use format version when serde datafiles

Author: CTTY

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,

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)?;
     }
 

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

@@ -123,9 +123,9 @@ impl Manifest {
 pub fn serialize_data_file_to_json(
     data_file: DataFile,
     partition_type: &super::StructType,
-    is_version_1: bool,
+    format_version: FormatVersion,
 ) -> Result<String> {
-    let serde = _serde::DataFileSerde::try_from(data_file, partition_type, is_version_1)?;
+    let serde = _serde::DataFileSerde::try_from(data_file, partition_type, format_version)?;
     serde_json::to_string(&serde).map_err(|e| {
         Error::new(
             ErrorKind::DataInvalid,
@@ -1120,9 +1120,6 @@ mod tests {
         // Get partition type from the partition spec
         let partition_type = partition_spec.partition_type(&schema).unwrap();
 
-        // Set version flag
-        let is_version_1 = false;
-
         // Create a vector of DataFile objects
         let data_files = vec![
             DataFileBuilder::default()
@@ -1157,7 +1154,8 @@ mod tests {
         let serialized_files = data_files
             .into_iter()
             .map(|f| {
-                let json = serialize_data_file_to_json(f, &partition_type, is_version_1).unwrap();
+                let json =
+                    serialize_data_file_to_json(f, &partition_type, FormatVersion::V2).unwrap();
                 println!("Test serialized data file: {}", json);
                 json
             })

crates/integrations/datafusion/src/physical_plan/write.rs

@@ -174,7 +174,7 @@ impl ExecutionPlan for IcebergWriteExec {
         // todo non-default partition spec?
         let spec_id = self.table.metadata().default_partition_spec_id();
         let partition_type = self.table.metadata().default_partition_type().clone();
-        let is_version_1 = self.table.metadata().format_version() == FormatVersion::V1;
+        let format_version = self.table.metadata().format_version();
 
         // Check data file format
         let file_format = DataFileFormat::from_str(
@@ -245,7 +245,7 @@ impl ExecutionPlan for IcebergWriteExec {
                 .map(|data_file| -> DFResult<String> {
                     // Serialize to JSON
                     let json =
-                        serialize_data_file_to_json(data_file, &partition_type, is_version_1)
+                        serialize_data_file_to_json(data_file, &partition_type, format_version)
                             .map_err(to_datafusion_error)?;
 
                     println!("Serialized data file: {}", json); // todo remove log