do not expose serde

Author: CTTY

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

@@ -96,10 +96,9 @@ impl ManifestEntryV1 {
     }
 }
 
-/// todo doc
 #[serde_as]
 #[derive(Serialize, Deserialize)]
-pub struct DataFileSerde {
+pub(super) struct DataFileSerde {
     #[serde(default)]
     content: i32,
     file_path: String,
@@ -127,7 +126,6 @@ pub struct DataFileSerde {
 }
 
 impl DataFileSerde {
-    /// todo doc
     pub fn try_from(
         value: super::DataFile,
         partition_type: &StructType,
@@ -162,7 +160,6 @@ impl DataFileSerde {
         })
     }
 
-    /// todo doc
     pub fn try_into(
         self,
         partition_spec_id: i32,

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

@@ -15,9 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-// todo fix encapsulation
 mod _serde;
-pub use _serde::*;
 
 mod data_file;
 pub use data_file::*;
@@ -35,7 +33,7 @@ use super::{
     Datum, FormatVersion, ManifestContentType, PartitionSpec, PrimitiveType, Schema, Struct,
     UNASSIGNED_SEQUENCE_NUMBER,
 };
-use crate::error::Result;
+use crate::error::{Error, ErrorKind, Result};
 
 /// A manifest contains metadata and a list of entries.
 #[derive(Debug, PartialEq, Eq, Clone)]
@@ -121,6 +119,38 @@ impl Manifest {
     }
 }
 
+/// Serialize a DataFile to a JSON string.
+pub fn serialize_data_file_to_json(
+    data_file: DataFile,
+    partition_type: &super::StructType,
+    is_version_1: bool,
+) -> Result<String> {
+    let serde = _serde::DataFileSerde::try_from(data_file, partition_type, is_version_1)?;
+    serde_json::to_string(&serde).map_err(|e| {
+        Error::new(
+            ErrorKind::DataInvalid,
+            format!("Failed to serialize DataFile to JSON: {}", e),
+        )
+    })
+}
+
+/// Deserialize a DataFile from a JSON string.
+pub fn deserialize_data_file_from_json(
+    json: &str,
+    partition_spec_id: i32,
+    partition_type: &super::StructType,
+    schema: &Schema,
+) -> Result<DataFile> {
+    let serde = serde_json::from_str::<_serde::DataFileSerde>(json).map_err(|e| {
+        Error::new(
+            ErrorKind::DataInvalid,
+            format!("Failed to deserialize JSON to DataFile: {}", e),
+        )
+    })?;
+
+    serde.try_into(partition_spec_id, partition_type, schema)
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;

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

@@ -33,7 +33,7 @@ use datafusion::physical_plan::{
 };
 use futures::StreamExt;
 use iceberg::Catalog;
-use iceberg::spec::{DataFile, DataFileSerde};
+use iceberg::spec::{DataFile, deserialize_data_file_from_json};
 use iceberg::table::Table;
 use iceberg::transaction::Transaction;
 
@@ -234,18 +234,18 @@ impl ExecutionPlan for IcebergCommitExec {
                     total_count += count_array.iter().flatten().sum::<u64>();
 
                     // Deserialize all data files from the StringArray
-                    let batch_files: Vec<DataFile> = (0..files_array.len())
-                        .map(|i| -> DFResult<DataFile> {
+                    let batch_files: Vec<DataFile> = files_array
+                        .into_iter()
+                        .flatten()
+                        .map(|f| -> DFResult<DataFile> {
                             // Parse JSON to DataFileSerde and convert to DataFile
-                            serde_json::from_str::<DataFileSerde>(files_array.value(i))
-                                .map_err(|e| {
-                                    DataFusionError::Internal(format!(
-                                        "Failed to deserialize data files: {}",
-                                        e
-                                    ))
-                                })?
-                                .try_into(spec_id, &partition_type, &current_schema)
-                                .map_err(to_datafusion_error)
+                            deserialize_data_file_from_json(
+                                f,
+                                spec_id,
+                                &partition_type,
+                                &current_schema,
+                            )
+                            .map_err(to_datafusion_error)
                         })
                         .collect::<datafusion::common::Result<_>>()?;
 

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

@@ -34,7 +34,7 @@ use datafusion::physical_plan::{
 };
 use futures::StreamExt;
 use iceberg::arrow::schema_to_arrow_schema;
-use iceberg::spec::{DataFileFormat, DataFileSerde, FormatVersion};
+use iceberg::spec::{DataFileFormat, FormatVersion, serialize_data_file_to_json};
 use iceberg::table::Table;
 use iceberg::writer::CurrentFileStatus;
 use iceberg::writer::file_writer::location_generator::{
@@ -208,19 +208,10 @@ impl ExecutionPlan for IcebergWriteExec {
                         DataFusionError::Execution(format!("Failed to build data file: {}", e))
                     })?;
 
-                    // Convert to DataFileSerde
-                    let serde = DataFileSerde::try_from(data_file, &partition_type, is_version_1)
-                        .map_err(|e| {
-                        DataFusionError::Execution(format!(
-                            "Failed to convert to DataFileSerde: {}",
-                            e
-                        ))
-                    })?;
-
                     // Serialize to JSON
-                    let json = serde_json::to_string(&serde).map_err(|e| {
-                        DataFusionError::Execution(format!("Failed to serialize to JSON: {}", e))
-                    })?;
+                    let json =
+                        serialize_data_file_to_json(data_file, &partition_type, is_version_1)
+                            .map_err(to_datafusion_error)?;
 
                     println!("Serialized data file: {}", json); // todo remove log
                     Ok(json)
@@ -244,8 +235,8 @@ mod tests {
 
     use datafusion::arrow::array::StringArray;
     use iceberg::spec::{
-        DataFile, DataFileBuilder, DataFileFormat, DataFileSerde, PartitionSpec, PrimitiveType,
-        Schema, Struct, Type,
+        DataFile, DataFileBuilder, DataFileFormat, PartitionSpec, PrimitiveType, Schema, Struct,
+        Type, deserialize_data_file_from_json, serialize_data_file_to_json,
     };
 
     // todo move this to DataFileSerde?
@@ -316,8 +307,7 @@ mod tests {
         let serialized_files = data_files
             .into_iter()
             .map(|f| {
-                let serde = DataFileSerde::try_from(f, &partition_type, is_version_1).unwrap();
-                let json = serde_json::to_string(&serde).unwrap();
+                let json = serialize_data_file_to_json(f, &partition_type, is_version_1).unwrap();
                 println!("Test serialized data file: {}", json);
                 json
             })
@@ -343,14 +333,13 @@ mod tests {
         let deserialized_files: Vec<DataFile> = serialized_files
             .into_iter()
             .map(|json| {
-                // First deserialize to DataFileSerde
-                let data_file_serde: DataFileSerde =
-                    serde_json::from_str(&json).expect("Failed to deserialize to DataFileSerde");
-
-                // Then convert to DataFile
-                let data_file = data_file_serde
-                    .try_into(partition_spec.spec_id(), &partition_type, &schema)
-                    .expect("Failed to convert DataFileSerde to DataFile");
+                let data_file = deserialize_data_file_from_json(
+                    &json,
+                    partition_spec.spec_id(),
+                    &partition_type,
+                    &schema,
+                )
+                .unwrap();
 
                 println!("Deserialized DataFile: {:?}", data_file);
                 data_file