fix fmt, input boundedness

Author: CTTY

crates/iceberg/src/arrow/nan_val_cnt_visitor.rs

@@ -159,6 +159,7 @@ impl NanValueCountVisitor {
         let arrow_arr_partner_accessor = ArrowArrayAccessor {};
 
         let struct_arr = Arc::new(StructArray::from(batch)) as ArrayRef;
+        // todo remove these log lines
         println!("----StructArray from record stream: {:?}", struct_arr);
         println!("----Schema.as_struct from table: {:?}", schema.as_struct());
         visit_struct_with_partner(

crates/iceberg/src/arrow/value.rs

@@ -457,6 +457,7 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
                 )
             })?;
 
+        // todo remove unneeded log lines
         println!(
             "!!!Accessor struct array from struct partner: {:?}",
             struct_array

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

@@ -74,7 +74,7 @@ impl IcebergCommitExec {
         PlanProperties::new(
             EquivalenceProperties::new(schema),
             Partitioning::UnknownPartitioning(1),
-            EmissionType::Incremental,
+            EmissionType::Final,
             Boundedness::Bounded,
         )
     }
@@ -148,7 +148,8 @@ impl ExecutionPlan for IcebergCommitExec {
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
         if children.len() != 1 {
             return Err(DataFusionError::Internal(
-                "IcebergCommitExec expects exactly one child, but provided {children.len()}".to_string(),
+                "IcebergCommitExec expects exactly one child, but provided {children.len()}"
+                    .to_string(),
             ));
         }
 

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

@@ -28,6 +28,7 @@ use datafusion::common::Result as DFResult;
 use datafusion::error::DataFusionError;
 use datafusion::execution::{SendableRecordBatchStream, TaskContext};
 use datafusion::physical_expr::{EquivalenceProperties, Partitioning};
+use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
 use datafusion::physical_plan::{
     DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties,
@@ -78,8 +79,8 @@ impl IcebergWriteExec {
         PlanProperties::new(
             EquivalenceProperties::new(schema),
             Partitioning::UnknownPartitioning(input.output_partitioning().partition_count()),
-            input.pipeline_behavior(),
-            input.boundedness(),
+            EmissionType::Final,
+            Boundedness::Bounded,
         )
     }