make data_files constant

Author: CTTY

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

@@ -37,6 +37,7 @@ use iceberg::spec::{DataFile, deserialize_data_file_from_json};
 use iceberg::table::Table;
 use iceberg::transaction::Transaction;
 
+use crate::physical_plan::DATA_FILES_COL_NAME;
 use crate::to_datafusion_error;
 
 /// IcebergCommitExec is responsible for collecting results from multiple IcebergWriteExec
@@ -199,7 +200,7 @@ impl ExecutionPlan for IcebergCommitExec {
                     let batch = batch_result?;
 
                     let files_array = batch
-                        .column_by_name("data_files")
+                        .column_by_name(DATA_FILES_COL_NAME)
                         .ok_or_else(|| {
                             DataFusionError::Internal(
                                 "Expected 'data_files' column in input batch".to_string(),

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

@@ -20,3 +20,5 @@ pub(crate) mod expr_to_predicate;
 pub(crate) mod metadata_scan;
 pub(crate) mod scan;
 pub(crate) mod write;
+
+pub(crate) const DATA_FILES_COL_NAME: &str = "data_files";

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

@@ -51,6 +51,7 @@ use iceberg::{Error, ErrorKind};
 use parquet::file::properties::WriterProperties;
 use uuid::Uuid;
 
+use crate::physical_plan::DATA_FILES_COL_NAME;
 use crate::to_datafusion_error;
 
 pub(crate) struct IcebergWriteExec {
@@ -88,15 +89,16 @@ impl IcebergWriteExec {
     fn make_result_batch(data_files: Vec<String>) -> DFResult<RecordBatch> {
         let files_array = Arc::new(StringArray::from(data_files)) as ArrayRef;
 
-        RecordBatch::try_from_iter_with_nullable(vec![("data_files", files_array, false)]).map_err(
-            |e| DataFusionError::ArrowError(e, Some("Failed to make result batch".to_string())),
-        )
+        RecordBatch::try_from_iter_with_nullable(vec![(DATA_FILES_COL_NAME, files_array, false)])
+            .map_err(|e| {
+                DataFusionError::ArrowError(e, Some("Failed to make result batch".to_string()))
+            })
     }
 
     fn make_result_schema() -> ArrowSchemaRef {
         // Define a schema.
         Arc::new(ArrowSchema::new(vec![Field::new(
-            "data_files",
+            DATA_FILES_COL_NAME,
             DataType::Utf8,
             false,
         )]))