Merge branch 'main' into hjiang/delete-column-implementation

Author: dentiny

crates/catalog/rest/src/catalog.rs

@@ -320,7 +320,7 @@ impl RestCatalog {
             None => None,
         };
 
-        let file_io = match warehouse_path.or(metadata_location) {
+        let file_io = match metadata_location.or(warehouse_path) {
             Some(url) => FileIO::from_path(url)?
                 .with_props(props)
                 .with_extensions(self.file_io_extensions.clone())

crates/iceberg/src/delete_vector.rs

@@ -21,14 +21,16 @@ use roaring::RoaringTreemap;
 use roaring::bitmap::Iter;
 use roaring::treemap::BitmapIter;
 
+use crate::{Error, ErrorKind, Result};
+
 #[derive(Debug, Default)]
 pub struct DeleteVector {
     inner: RoaringTreemap,
 }
 
 impl DeleteVector {
     #[allow(unused)]
-    pub(crate) fn new(roaring_treemap: RoaringTreemap) -> DeleteVector {
+    pub fn new(roaring_treemap: RoaringTreemap) -> DeleteVector {
         DeleteVector {
             inner: roaring_treemap,
         }
@@ -43,6 +45,25 @@ impl DeleteVector {
         self.inner.insert(pos)
     }
 
+    /// Marks the given `positions` as deleted and returns the number of elements appended.
+    ///
+    /// The input slice must be strictly ordered in ascending order, and every value must be greater than all existing values already in the set.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the precondition is not met.
+    #[allow(dead_code)]
+    pub fn insert_positions(&mut self, positions: &[u64]) -> Result<usize> {
+        if let Err(err) = self.inner.append(positions.iter().copied()) {
+            return Err(Error::new(
+                ErrorKind::PreconditionFailed,
+                "failed to marks rows as deleted".to_string(),
+            )
+            .with_source(err));
+        }
+        Ok(positions.len())
+    }
+
     #[allow(unused)]
     pub fn len(&self) -> u64 {
         self.inner.len()
@@ -120,3 +141,61 @@ impl BitOrAssign for DeleteVector {
         self.inner.bitor_assign(&other.inner);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_insertion_and_iteration() {
+        let mut dv = DeleteVector::default();
+        assert!(dv.insert(42));
+        assert!(dv.insert(100));
+        assert!(!dv.insert(42));
+
+        let mut items: Vec<u64> = dv.iter().collect();
+        items.sort();
+        assert_eq!(items, vec![42, 100]);
+        assert_eq!(dv.len(), 2);
+    }
+
+    #[test]
+    fn test_successful_insert_positions() {
+        let mut dv = DeleteVector::default();
+        let positions = vec![1, 2, 3, 1000, 1 << 33];
+        assert_eq!(dv.insert_positions(&positions).unwrap(), 5);
+
+        let mut collected: Vec<u64> = dv.iter().collect();
+        collected.sort();
+        assert_eq!(collected, positions);
+    }
+
+    /// Testing scenario: bulk insertion fails because input positions are not strictly increasing.
+    #[test]
+    fn test_failed_insertion_unsorted_elements() {
+        let mut dv = DeleteVector::default();
+        let positions = vec![1, 3, 5, 4];
+        let res = dv.insert_positions(&positions);
+        assert!(res.is_err());
+    }
+
+    /// Testing scenario: bulk insertion fails because input positions have intersection with existing ones.
+    #[test]
+    fn test_failed_insertion_with_intersection() {
+        let mut dv = DeleteVector::default();
+        let positions = vec![1, 3, 5];
+        assert_eq!(dv.insert_positions(&positions).unwrap(), 3);
+
+        let res = dv.insert_positions(&[2, 4]);
+        assert!(res.is_err());
+    }
+
+    /// Testing scenario: bulk insertion fails because input positions have duplicates.
+    #[test]
+    fn test_failed_insertion_duplicate_elements() {
+        let mut dv = DeleteVector::default();
+        let positions = vec![1, 3, 5, 5];
+        let res = dv.insert_positions(&positions);
+        assert!(res.is_err());
+    }
+}

crates/iceberg/src/spec/manifest/_serde.rs

@@ -21,7 +21,7 @@ use serde_derive::{Deserialize, Serialize};
 use serde_with::serde_as;
 
 use super::{Datum, ManifestEntry, Schema, Struct};
-use crate::spec::{Literal, RawLiteral, StructType, Type};
+use crate::spec::{FormatVersion, Literal, RawLiteral, StructType, Type};
 use crate::{Error, ErrorKind};
 
 #[derive(Serialize, Deserialize)]
@@ -40,7 +40,7 @@ impl ManifestEntryV2 {
             snapshot_id: value.snapshot_id,
             sequence_number: value.sequence_number,
             file_sequence_number: value.file_sequence_number,
-            data_file: DataFileSerde::try_from(value.data_file, partition_type, false)?,
+            data_file: DataFileSerde::try_from(value.data_file, partition_type, FormatVersion::V2)?,
         })
     }
 
@@ -74,7 +74,7 @@ impl ManifestEntryV1 {
         Ok(Self {
             status: value.status as i32,
             snapshot_id: value.snapshot_id.unwrap_or_default(),
-            data_file: DataFileSerde::try_from(value.data_file, partition_type, true)?,
+            data_file: DataFileSerde::try_from(value.data_file, partition_type, FormatVersion::V1)?,
         })
     }
 
@@ -129,9 +129,13 @@ impl DataFileSerde {
     pub fn try_from(
         value: super::DataFile,
         partition_type: &StructType,
-        is_version_1: bool,
+        format_version: FormatVersion,
     ) -> Result<Self, Error> {
-        let block_size_in_bytes = if is_version_1 { Some(0) } else { None };
+        let block_size_in_bytes = if format_version == FormatVersion::V1 {
+            Some(0)
+        } else {
+            None
+        };
         Ok(Self {
             content: value.content as i32,
             file_path: value.file_path,
@@ -292,16 +296,23 @@ fn parse_i64_entry(v: Vec<I64Entry>) -> Result<HashMap<i32, u64>, Error> {
     Ok(m)
 }
 
+#[allow(unused_mut)]
 fn to_i64_entry(entries: HashMap<i32, u64>) -> Result<Vec<I64Entry>, Error> {
-    entries
+    let mut i64_entries = entries
         .iter()
         .map(|e| {
             Ok(I64Entry {
                 key: *e.0,
                 value: (*e.1).try_into()?,
             })
         })
-        .collect()
+        .collect::<Result<Vec<_>, Error>>()?;
+
+    // Ensure that the order is deterministic during testing
+    #[cfg(test)]
+    i64_entries.sort_by_key(|e| e.key);
+
+    Ok(i64_entries)
 }
 
 #[cfg(test)]
@@ -432,4 +443,154 @@ mod tests {
 
         assert_eq!(actual_data_file[0].content, DataContentType::Data)
     }
+
+    #[test]
+    fn test_manifest_entry_v1_to_v2_projection() {
+        use crate::spec::manifest::_serde::{DataFileSerde, ManifestEntryV1};
+        use crate::spec::{Literal, RawLiteral, Struct, StructType};
+
+        let partition = RawLiteral::try_from(
+            Literal::Struct(Struct::empty()),
+            &Type::Struct(StructType::new(vec![])),
+        )
+        .unwrap();
+
+        // Create a V1 manifest entry struct (lacks V2 sequence number fields)
+        let v1_entry = ManifestEntryV1 {
+            status: 1, // Added
+            snapshot_id: 12345,
+            data_file: DataFileSerde {
+                content: 0, // DataFileSerde is shared between V1/V2
+                file_path: "test/path.parquet".to_string(),
+                file_format: "PARQUET".to_string(),
+                partition,
+                record_count: 100,
+                file_size_in_bytes: 1024,
+                block_size_in_bytes: Some(0), // V1 includes this field
+                column_sizes: None,
+                value_counts: None,
+                null_value_counts: None,
+                nan_value_counts: None,
+                lower_bounds: None,
+                upper_bounds: None,
+                key_metadata: None,
+                split_offsets: None,
+                equality_ids: None, // Will be converted to empty vec
+                sort_order_id: None,
+                first_row_id: None,
+                referenced_data_file: None,
+                content_offset: None,
+                content_size_in_bytes: None,
+            },
+        };
+
+        // Test the explicit V1→V2 conversion logic in ManifestEntryV1::try_into()
+        let v2_entry = v1_entry
+            .try_into(
+                0, // partition_spec_id
+                &StructType::new(vec![]),
+                &schema(),
+            )
+            .unwrap();
+
+        // Verify that V1→V2 conversion adds the missing V2 sequence number fields
+        assert_eq!(
+            v2_entry.sequence_number,
+            Some(0),
+            "ManifestEntryV1::try_into() should set sequence_number to 0"
+        );
+        assert_eq!(
+            v2_entry.file_sequence_number,
+            Some(0),
+            "ManifestEntryV1::try_into() should set file_sequence_number to 0"
+        );
+        assert_eq!(
+            v2_entry.snapshot_id,
+            Some(12345),
+            "snapshot_id should be preserved during conversion"
+        );
+
+        // Verify that DataFileSerde conversion applies V2 defaults
+        assert_eq!(
+            v2_entry.data_file.content,
+            DataContentType::Data,
+            "DataFileSerde should convert content 0 to DataContentType::Data"
+        );
+        assert_eq!(
+            v2_entry.data_file.equality_ids,
+            Vec::<i32>::new(),
+            "DataFileSerde should convert None equality_ids to empty vec"
+        );
+
+        // Verify other fields are preserved during conversion
+        assert_eq!(v2_entry.data_file.file_path, "test/path.parquet");
+        assert_eq!(v2_entry.data_file.record_count, 100);
+        assert_eq!(v2_entry.data_file.file_size_in_bytes, 1024);
+    }
+
+    #[test]
+    fn test_data_file_serde_v1_field_defaults() {
+        use crate::spec::manifest::_serde::DataFileSerde;
+        use crate::spec::{Literal, RawLiteral, Struct, StructType};
+
+        let partition = RawLiteral::try_from(
+            Literal::Struct(Struct::empty()),
+            &Type::Struct(StructType::new(vec![])),
+        )
+        .unwrap();
+
+        // Create a DataFileSerde that simulates V1 deserialization behavior
+        // (missing V2 fields would be None due to #[serde(default)])
+        let v1_style_data_file = DataFileSerde {
+            content: 0, // V1 doesn't have this field, defaults to 0 via #[serde(default)]
+            file_path: "test/data.parquet".to_string(),
+            file_format: "PARQUET".to_string(),
+            partition,
+            record_count: 500,
+            file_size_in_bytes: 2048,
+            block_size_in_bytes: Some(1024), // V1 includes this field, V2 skips it
+            column_sizes: None,
+            value_counts: None,
+            null_value_counts: None,
+            nan_value_counts: None,
+            lower_bounds: None,
+            upper_bounds: None,
+            key_metadata: None,
+            split_offsets: None,
+            equality_ids: None, // V1 doesn't have this field, defaults to None via #[serde(default)]
+            sort_order_id: None,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
+        };
+
+        // Test the DataFileSerde::try_into() conversion that handles V1 field defaults
+        let data_file = v1_style_data_file
+            .try_into(
+                0, // partition_spec_id
+                &StructType::new(vec![]),
+                &schema(),
+            )
+            .unwrap();
+
+        // Verify that DataFileSerde::try_into() applies correct defaults for missing V2 fields
+        assert_eq!(
+            data_file.content,
+            DataContentType::Data,
+            "content 0 should convert to DataContentType::Data"
+        );
+        assert_eq!(
+            data_file.equality_ids,
+            Vec::<i32>::new(),
+            "None equality_ids should convert to empty vec via unwrap_or_default()"
+        );
+
+        // Verify other fields are handled correctly during conversion
+        assert_eq!(data_file.file_path, "test/data.parquet");
+        assert_eq!(data_file.file_format, DataFileFormat::Parquet);
+        assert_eq!(data_file.record_count, 500);
+        assert_eq!(data_file.file_size_in_bytes, 2048);
+        assert_eq!(data_file.partition_spec_id, 0);
+    }
 }

crates/iceberg/src/spec/manifest/data_file.rs

@@ -297,8 +297,12 @@ pub fn write_data_files_to_avro<W: Write>(
     let mut writer = AvroWriter::new(&avro_schema, writer);
 
     for data_file in data_files {
-        let value = to_value(DataFileSerde::try_from(data_file, partition_type, true)?)?
-            .resolve(&avro_schema)?;
+        let value = to_value(DataFileSerde::try_from(
+            data_file,
+            partition_type,
+            FormatVersion::V1,
+        )?)?
+        .resolve(&avro_schema)?;
         writer.append(value)?;
     }
 
@@ -333,9 +337,10 @@ pub fn read_data_files_from_avro<R: Read>(
 
 /// Type of content stored by the data file: data, equality deletes, or
 /// position deletes (all v1 files are data files)
-#[derive(Debug, PartialEq, Eq, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Clone, Copy, Serialize, Deserialize, Default)]
 pub enum DataContentType {
     /// value: 0
+    #[default]
     Data = 0,
     /// value: 1
     PositionDeletes = 1,
@@ -399,3 +404,17 @@ impl std::fmt::Display for DataFileFormat {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use crate::spec::DataContentType;
+    #[test]
+    fn test_data_content_type_default() {
+        assert_eq!(DataContentType::default(), DataContentType::Data);
+    }
+
+    #[test]
+    fn test_data_content_type_default_value() {
+        assert_eq!(DataContentType::default() as i32, 0);
+    }
+}