Merge branch 'apache:main' into dev

sundy-li · web-flow · commit 21f7e243da9a · 2025-05-07T11:11:26.000+08:00
diff --git a/crates/iceberg/src/arrow/schema.rs b/crates/iceberg/src/arrow/schema.rs
@@ -167,6 +167,7 @@ fn visit_type<V: ArrowSchemaVisitor>(r#type: &DataType, visitor: &mut V) -> Resu
             )),
         },
         DataType::Struct(fields) => visit_struct(fields, visitor),
+        DataType::Dictionary(_key_type, value_type) => visit_type(value_type, visitor),
         other => Err(Error::new(
             ErrorKind::DataInvalid,
             format!("Cannot visit Arrow data type: {other}"),
@@ -1029,6 +1030,7 @@ mod tests {
             Arc::new(simple_field(DEFAULT_MAP_FIELD_NAME, r#struct, false, "17")),
             false,
         );
+        let dictionary = DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
 
         let fields = Fields::from(vec![
             simple_field("aa", DataType::Int32, false, "18"),
@@ -1108,6 +1110,7 @@ mod tests {
             ),
             simple_field("map", map, false, "16"),
             simple_field("struct", r#struct, false, "17"),
+            simple_field("dictionary", dictionary, false, "30"),
         ])
     }
 
@@ -1285,6 +1288,12 @@ mod tests {
                             }
                         ]
                     }
+                },
+                {
+                    "id":30,
+                    "name":"dictionary",
+                    "required":true,
+                    "type":"string"
                 }
             ],
             "identifier-field-ids":[]
@@ -1299,7 +1308,7 @@ mod tests {
         let arrow_schema = arrow_schema_for_arrow_schema_to_schema_test();
         let schema = iceberg_schema_for_arrow_schema_to_schema_test();
         let converted_schema = arrow_schema_to_schema(&arrow_schema).unwrap();
-        assert_eq!(converted_schema, schema);
+        pretty_assertions::assert_eq!(converted_schema, schema);
     }
 
     fn arrow_schema_for_schema_to_arrow_schema_test() -> ArrowSchema {
diff --git a/crates/iceberg/src/expr/visitors/expression_evaluator.rs b/crates/iceberg/src/expr/visitors/expression_evaluator.rs
@@ -347,6 +347,10 @@ mod tests {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
@@ -371,6 +375,10 @@ mod tests {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
diff --git a/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs b/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs
@@ -1997,6 +1997,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
@@ -2019,6 +2023,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
@@ -2077,6 +2085,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
     fn get_test_file_2() -> DataFile {
@@ -2104,6 +2116,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
@@ -2132,6 +2148,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
@@ -2160,6 +2180,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 }
diff --git a/crates/iceberg/src/expr/visitors/strict_metrics_evaluator.rs b/crates/iceberg/src/expr/visitors/strict_metrics_evaluator.rs
@@ -582,6 +582,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
@@ -604,6 +608,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
@@ -626,6 +634,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
@@ -649,6 +661,10 @@ mod test {
             equality_ids: vec![],
             sort_order_id: None,
             partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }
     }
 
diff --git a/crates/iceberg/src/spec/manifest/_serde.rs b/crates/iceberg/src/spec/manifest/_serde.rs
@@ -119,6 +119,10 @@ pub(super) struct DataFileSerde {
     #[serde(default)]
     equality_ids: Option<Vec<i32>>,
     sort_order_id: Option<i32>,
+    first_row_id: Option<i64>,
+    referenced_data_file: Option<String>,
+    content_offset: Option<i64>,
+    content_size_in_bytes: Option<i64>,
 }
 
 impl DataFileSerde {
@@ -149,6 +153,10 @@ impl DataFileSerde {
             split_offsets: Some(value.split_offsets),
             equality_ids: Some(value.equality_ids),
             sort_order_id: value.sort_order_id,
+            first_row_id: value.first_row_id,
+            referenced_data_file: value.referenced_data_file,
+            content_offset: value.content_offset,
+            content_size_in_bytes: value.content_size_in_bytes,
         })
     }
 
@@ -215,6 +223,10 @@ impl DataFileSerde {
             equality_ids: self.equality_ids.unwrap_or_default(),
             sort_order_id: self.sort_order_id,
             partition_spec_id,
+            first_row_id: self.first_row_id,
+            referenced_data_file: self.referenced_data_file,
+            content_offset: self.content_offset,
+            content_size_in_bytes: self.content_size_in_bytes,
         })
     }
 }
@@ -359,7 +371,11 @@ mod tests {
             split_offsets: vec![4],
             equality_ids: vec![],
             sort_order_id: Some(0),
-            partition_spec_id: 0
+            partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
         }];
 
         let mut buffer = Vec::new();
diff --git a/crates/iceberg/src/spec/manifest/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs
@@ -43,7 +43,7 @@ pub struct DataFile {
     pub(crate) file_path: String,
     /// field id: 101
     ///
-    /// String file format name, avro, orc or parquet
+    /// String file format name, `avro`, `orc`, `parquet`, or `puffin`
     pub(crate) file_format: DataFileFormat,
     /// field id: 102
     ///
@@ -52,7 +52,7 @@ pub struct DataFile {
     pub(crate) partition: Struct,
     /// field id: 103
     ///
-    /// Number of records in this file
+    /// Number of records in this file, or the cardinality of a deletion vector
     pub(crate) record_count: u64,
     /// field id: 104
     ///
@@ -148,9 +148,35 @@ pub struct DataFile {
     /// delete files.
     #[builder(default, setter(strip_option))]
     pub(crate) sort_order_id: Option<i32>,
+    /// field id: 142
+    ///
+    /// The _row_id for the first row in the data file.
+    /// For more details, refer to https://github.com/apache/iceberg/blob/main/format/spec.md#first-row-id-inheritance
+    #[builder(default)]
+    pub(crate) first_row_id: Option<i64>,
     /// This field is not included in spec. It is just store in memory representation used
     /// in process.
     pub(crate) partition_spec_id: i32,
+    /// field id: 143
+    ///
+    /// Fully qualified location (URI with FS scheme) of a data file that all deletes reference.
+    /// Position delete metadata can use `referenced_data_file` when all deletes tracked by the
+    /// entry are in a single data file. Setting the referenced file is required for deletion vectors.
+    #[builder(default)]
+    pub(crate) referenced_data_file: Option<String>,
+    /// field: 144
+    ///
+    /// The offset in the file where the content starts.
+    /// The `content_offset` and `content_size_in_bytes` fields are used to reference a specific blob
+    /// for direct access to a deletion vector. For deletion vectors, these values are required and must
+    /// exactly match the `offset` and `length` stored in the Puffin footer for the deletion vector blob.
+    #[builder(default)]
+    pub(crate) content_offset: Option<i64>,
+    /// field: 145
+    ///
+    /// The length of a referenced content stored in the file; required if `content_offset` is present
+    #[builder(default)]
+    pub(crate) content_size_in_bytes: Option<i64>,
 }
 
 impl DataFile {
@@ -226,6 +252,10 @@ impl DataFile {
     pub fn equality_ids(&self) -> &[i32] {
         &self.equality_ids
     }
+    /// Get the first row id in the data file.
+    pub fn first_row_id(&self) -> Option<i64> {
+        self.first_row_id
+    }
     /// Get the sort order id of the data file.
     /// Only data files and equality delete files should be
     /// written with a non-null order id. Position deletes are required to be
@@ -235,6 +265,21 @@ impl DataFile {
     pub fn sort_order_id(&self) -> Option<i32> {
         self.sort_order_id
     }
+    /// Get the fully qualified referenced location for the corresponding data file.
+    /// Positional delete files could have the field set, and deletion vectors must the field set.
+    pub fn referenced_data_file(&self) -> Option<String> {
+        self.referenced_data_file.clone()
+    }
+    /// Get the offset in the file where the blob content starts.
+    /// Only meaningful for puffin blobs, and required for deletion vectors.
+    pub fn content_offset(&self) -> Option<i64> {
+        self.content_offset
+    }
+    /// Get the length of a puffin blob.
+    /// Only meaningful for puffin blobs, and required for deletion vectors.
+    pub fn content_size_in_bytes(&self) -> Option<i64> {
+        self.content_size_in_bytes
+    }
 }
 
 /// Convert data files to avro bytes and write to writer.
@@ -323,6 +368,8 @@ pub enum DataFileFormat {
     Orc,
     /// Parquet file format: <https://parquet.apache.org/>
     Parquet,
+    /// Puffin file format: <https://iceberg.apache.org/puffin-spec/>
+    Puffin,
 }
 
 impl FromStr for DataFileFormat {
@@ -333,6 +380,7 @@ impl FromStr for DataFileFormat {
             "avro" => Ok(Self::Avro),
             "orc" => Ok(Self::Orc),
             "parquet" => Ok(Self::Parquet),
+            "puffin" => Ok(Self::Puffin),
             _ => Err(Error::new(
                 ErrorKind::DataInvalid,
                 format!("Unsupported data file format: {}", s),
@@ -347,6 +395,7 @@ impl std::fmt::Display for DataFileFormat {
             DataFileFormat::Avro => write!(f, "avro"),
             DataFileFormat::Orc => write!(f, "orc"),
             DataFileFormat::Parquet => write!(f, "parquet"),
+            DataFileFormat::Puffin => write!(f, "puffin"),
         }
     }
 }
diff --git a/crates/iceberg/src/spec/manifest/entry.rs b/crates/iceberg/src/spec/manifest/entry.rs
@@ -469,6 +469,46 @@ static SORT_ORDER_ID: Lazy<NestedFieldRef> = {
     })
 };
 
+static FIRST_ROW_ID: Lazy<NestedFieldRef> = {
+    Lazy::new(|| {
+        Arc::new(NestedField::optional(
+            142,
+            "first_row_id",
+            Type::Primitive(PrimitiveType::Long),
+        ))
+    })
+};
+
+static REFERENCE_DATA_FILE: Lazy<NestedFieldRef> = {
+    Lazy::new(|| {
+        Arc::new(NestedField::optional(
+            143,
+            "referenced_data_file",
+            Type::Primitive(PrimitiveType::String),
+        ))
+    })
+};
+
+static CONTENT_OFFSET: Lazy<NestedFieldRef> = {
+    Lazy::new(|| {
+        Arc::new(NestedField::optional(
+            144,
+            "content_offset",
+            Type::Primitive(PrimitiveType::Long),
+        ))
+    })
+};
+
+static CONTENT_SIZE_IN_BYTES: Lazy<NestedFieldRef> = {
+    Lazy::new(|| {
+        Arc::new(NestedField::optional(
+            145,
+            "content_size_in_bytes",
+            Type::Primitive(PrimitiveType::Long),
+        ))
+    })
+};
+
 fn data_file_fields_v2(partition_type: &StructType) -> Vec<NestedFieldRef> {
     vec![
         CONTENT.clone(),
@@ -491,6 +531,10 @@ fn data_file_fields_v2(partition_type: &StructType) -> Vec<NestedFieldRef> {
         SPLIT_OFFSETS.clone(),
         EQUALITY_IDS.clone(),
         SORT_ORDER_ID.clone(),
+        FIRST_ROW_ID.clone(),
+        REFERENCE_DATA_FILE.clone(),
+        CONTENT_OFFSET.clone(),
+        CONTENT_SIZE_IN_BYTES.clone(),
     ]
 }
 
diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs
diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs
diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs

Original file line number	Diff line number	Diff line change
`@@ -347,6 +347,10 @@ mod tests {`
`347`	`347`	`equality_ids: vec![],`
`348`	`348`	`sort_order_id: None,`
`349`	`349`	`partition_spec_id: 0,`
	`350`	`+ first_row_id: None,`
	`351`	`+ referenced_data_file: None,`
	`352`	`+ content_offset: None,`
	`353`	`+ content_size_in_bytes: None,`
`350`	`354`	`}`
`351`	`355`	`}`
`352`	`356`
`@@ -371,6 +375,10 @@ mod tests {`
`371`	`375`	`equality_ids: vec![],`
`372`	`376`	`sort_order_id: None,`
`373`	`377`	`partition_spec_id: 0,`
	`378`	`+ first_row_id: None,`
	`379`	`+ referenced_data_file: None,`
	`380`	`+ content_offset: None,`
	`381`	`+ content_size_in_bytes: None,`
`374`	`382`	`}`
`375`	`383`	`}`
`376`	`384`
Original file line number	Diff line number	Diff line change
`@@ -1997,6 +1997,10 @@ mod test {`
`1997`	`1997`	`equality_ids: vec![],`
`1998`	`1998`	`sort_order_id: None,`
`1999`	`1999`	`partition_spec_id: 0,`
	`2000`	`+ first_row_id: None,`
	`2001`	`+ referenced_data_file: None,`
	`2002`	`+ content_offset: None,`
	`2003`	`+ content_size_in_bytes: None,`
`2000`	`2004`	`}`
`2001`	`2005`	`}`
`2002`	`2006`
`@@ -2019,6 +2023,10 @@ mod test {`
`2019`	`2023`	`equality_ids: vec![],`
`2020`	`2024`	`sort_order_id: None,`
`2021`	`2025`	`partition_spec_id: 0,`
	`2026`	`+ first_row_id: None,`
	`2027`	`+ referenced_data_file: None,`
	`2028`	`+ content_offset: None,`
	`2029`	`+ content_size_in_bytes: None,`
`2022`	`2030`	`}`
`2023`	`2031`	`}`
`2024`	`2032`
`@@ -2077,6 +2085,10 @@ mod test {`
`2077`	`2085`	`equality_ids: vec![],`
`2078`	`2086`	`sort_order_id: None,`
`2079`	`2087`	`partition_spec_id: 0,`
	`2088`	`+ first_row_id: None,`
	`2089`	`+ referenced_data_file: None,`
	`2090`	`+ content_offset: None,`
	`2091`	`+ content_size_in_bytes: None,`
`2080`	`2092`	`}`
`2081`	`2093`	`}`
`2082`	`2094`	`fn get_test_file_2() -> DataFile {`
`@@ -2104,6 +2116,10 @@ mod test {`
`2104`	`2116`	`equality_ids: vec![],`
`2105`	`2117`	`sort_order_id: None,`
`2106`	`2118`	`partition_spec_id: 0,`
	`2119`	`+ first_row_id: None,`
	`2120`	`+ referenced_data_file: None,`
	`2121`	`+ content_offset: None,`
	`2122`	`+ content_size_in_bytes: None,`
`2107`	`2123`	`}`
`2108`	`2124`	`}`
`2109`	`2125`
`@@ -2132,6 +2148,10 @@ mod test {`
`2132`	`2148`	`equality_ids: vec![],`
`2133`	`2149`	`sort_order_id: None,`
`2134`	`2150`	`partition_spec_id: 0,`
	`2151`	`+ first_row_id: None,`
	`2152`	`+ referenced_data_file: None,`
	`2153`	`+ content_offset: None,`
	`2154`	`+ content_size_in_bytes: None,`
`2135`	`2155`	`}`
`2136`	`2156`	`}`
`2137`	`2157`
`@@ -2160,6 +2180,10 @@ mod test {`
`2160`	`2180`	`equality_ids: vec![],`
`2161`	`2181`	`sort_order_id: None,`
`2162`	`2182`	`partition_spec_id: 0,`
	`2183`	`+ first_row_id: None,`
	`2184`	`+ referenced_data_file: None,`
	`2185`	`+ content_offset: None,`
	`2186`	`+ content_size_in_bytes: None,`
`2163`	`2187`	`}`
`2164`	`2188`	`}`
`2165`	`2189`	`}`
Original file line number	Diff line number	Diff line change
`@@ -582,6 +582,10 @@ mod test {`
`582`	`582`	`equality_ids: vec![],`
`583`	`583`	`sort_order_id: None,`
`584`	`584`	`partition_spec_id: 0,`
	`585`	`+ first_row_id: None,`
	`586`	`+ referenced_data_file: None,`
	`587`	`+ content_offset: None,`
	`588`	`+ content_size_in_bytes: None,`
`585`	`589`	`}`
`586`	`590`	`}`
`587`	`591`
`@@ -604,6 +608,10 @@ mod test {`
`604`	`608`	`equality_ids: vec![],`
`605`	`609`	`sort_order_id: None,`
`606`	`610`	`partition_spec_id: 0,`
	`611`	`+ first_row_id: None,`
	`612`	`+ referenced_data_file: None,`
	`613`	`+ content_offset: None,`
	`614`	`+ content_size_in_bytes: None,`
`607`	`615`	`}`
`608`	`616`	`}`
`609`	`617`
`@@ -626,6 +634,10 @@ mod test {`
`626`	`634`	`equality_ids: vec![],`
`627`	`635`	`sort_order_id: None,`
`628`	`636`	`partition_spec_id: 0,`
	`637`	`+ first_row_id: None,`
	`638`	`+ referenced_data_file: None,`
	`639`	`+ content_offset: None,`
	`640`	`+ content_size_in_bytes: None,`
`629`	`641`	`}`
`630`	`642`	`}`
`631`	`643`
`@@ -649,6 +661,10 @@ mod test {`
`649`	`661`	`equality_ids: vec![],`
`650`	`662`	`sort_order_id: None,`
`651`	`663`	`partition_spec_id: 0,`
	`664`	`+ first_row_id: None,`
	`665`	`+ referenced_data_file: None,`
	`666`	`+ content_offset: None,`
	`667`	`+ content_size_in_bytes: None,`
`652`	`668`	`}`
`653`	`669`	`}`
`654`	`670`