Support Datafusion insert_into

Author: CTTY

crates/iceberg/src/arrow/nan_val_cnt_visitor.rs

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

crates/iceberg/src/arrow/value.rs

@@ -440,10 +440,12 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
         Ok(schema_partner)
     }
 
+    // todo generate field_pos in datafusion instead of passing to here
     fn field_partner<'a>(
         &self,
         struct_partner: &'a ArrayRef,
         field: &NestedField,
+        field_pos: Option<usize>,
     ) -> Result<&'a ArrayRef> {
         let struct_array = struct_partner
             .as_any()
@@ -455,6 +457,13 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
                 )
             })?;
 
+        println!(
+            "!!!Accessor struct array from struct partner: {:?}",
+            struct_array
+        );
+
+        println!("!!!field: {:?}", field);
+
         let field_pos = struct_array
             .fields()
             .iter()
@@ -463,12 +472,12 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
                     .map(|id| id == field.id)
                     .unwrap_or(false)
             })
-            .ok_or_else(|| {
+            .unwrap_or(field_pos.ok_or_else(|| {
                 Error::new(
                     ErrorKind::DataInvalid,
                     format!("Field id {} not found in struct array", field.id),
                 )
-            })?;
+            })?);
 
         Ok(struct_array.column(field_pos))
     }

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

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

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

@@ -15,7 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
+// todo fix encapsulation
 mod _serde;
+pub use _serde::*;
 
 mod data_file;
 pub use data_file::*;

crates/iceberg/src/spec/schema/visitor.rs

@@ -192,7 +192,12 @@ pub trait PartnerAccessor<P> {
     /// Get the struct partner from schema partner.
     fn struct_parner<'a>(&self, schema_partner: &'a P) -> Result<&'a P>;
     /// Get the field partner from struct partner.
-    fn field_partner<'a>(&self, struct_partner: &'a P, field: &NestedField) -> Result<&'a P>;
+    fn field_partner<'a>(
+        &self,
+        struct_partner: &'a P,
+        field: &NestedField,
+        field_pos: Option<usize>,
+    ) -> Result<&'a P>;
     /// Get the list element partner from list partner.
     fn list_element_partner<'a>(&self, list_partner: &'a P) -> Result<&'a P>;
     /// Get the map key partner from map partner.
@@ -253,8 +258,8 @@ pub fn visit_struct_with_partner<P, V: SchemaWithPartnerVisitor<P>, A: PartnerAc
     accessor: &A,
 ) -> Result<V::T> {
     let mut results = Vec::with_capacity(s.fields().len());
-    for field in s.fields() {
-        let field_partner = accessor.field_partner(partner, field)?;
+    for (pos, field) in s.fields().iter().enumerate() {
+        let field_partner = accessor.field_partner(partner, field, Some(pos))?;
         visitor.before_struct_field(field, field_partner)?;
         let result = visit_type_with_partner(&field.field_type, field_partner, visitor, accessor)?;
         visitor.after_struct_field(field, field_partner)?;

crates/integrations/datafusion/Cargo.toml

@@ -34,9 +34,12 @@ async-trait = { workspace = true }
 datafusion = { workspace = true }
 futures = { workspace = true }
 iceberg = { workspace = true }
+parquet = { workspace = true }
 tokio = { workspace = true }
+serde_json = { workspace = true }
 
 [dev-dependencies]
 expect-test = { workspace = true }
+iceberg-catalog-memory = { workspace = true }
 parquet = { workspace = true }
 tempfile = { workspace = true }