feat: adopt kernel schema types

Cargo.toml

@@ -26,6 +26,9 @@ debug = true
 debug = "line-tables-only"
 
 [workspace.dependencies]
+delta_kernel = { version = "0.1" }
+# delta_kernel = { path = "../delta-kernel-rs/kernel" }
+
 # arrow
 arrow = { version = "51" }
 arrow-arith = { version = "51" }

crates/aws/tests/integration_s3_dynamodb.rs

@@ -390,7 +390,7 @@ async fn prepare_table(context: &IntegrationContext, table_name: &str) -> TestRe
     // create delta table
     let table = DeltaOps(table)
         .create()
-        .with_columns(schema.fields().clone())
+        .with_columns(schema.fields().cloned())
         .await?;
     println!("table created: {table:?}");
     Ok(table)

crates/core/Cargo.toml

@@ -15,6 +15,8 @@ rust-version.workspace = true
 features = ["datafusion", "json", "unity-experimental"]
 
 [dependencies]
+delta_kernel.workspace = true
+
 # arrow
 arrow = { workspace = true }
 arrow-arith = { workspace = true }

crates/core/src/delta_datafusion/expr.rs

@@ -542,7 +542,7 @@ mod test {
 
         let table = DeltaOps::new_in_memory()
             .create()
-            .with_columns(schema.fields().clone())
+            .with_columns(schema.fields().cloned())
             .await
             .unwrap();
         assert_eq!(table.version(), 0);

crates/core/src/delta_datafusion/mod.rs

@@ -36,7 +36,6 @@ use arrow::record_batch::RecordBatch;
 use arrow_array::types::UInt16Type;
 use arrow_array::{Array, DictionaryArray, StringArray, TypedDictionaryArray};
 use arrow_cast::display::array_value_to_string;
-
 use arrow_schema::Field;
 use async_trait::async_trait;
 use chrono::{DateTime, TimeZone, Utc};
@@ -78,15 +77,14 @@ use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 use datafusion_sql::planner::ParserOptions;
 use either::Either;
 use futures::TryStreamExt;
-
 use itertools::Itertools;
 use object_store::ObjectMeta;
 use serde::{Deserialize, Serialize};
 use url::Url;
 
 use crate::delta_datafusion::expr::parse_predicate_expression;
 use crate::errors::{DeltaResult, DeltaTableError};
-use crate::kernel::{Add, DataCheck, EagerSnapshot, Invariant, Snapshot};
+use crate::kernel::{Add, DataCheck, EagerSnapshot, Invariant, Snapshot, StructTypeExt};
 use crate::logstore::LogStoreRef;
 use crate::table::builder::ensure_table_uri;
 use crate::table::state::DeltaTableState;
@@ -202,13 +200,11 @@ fn _arrow_schema(snapshot: &Snapshot, wrap_partitions: bool) -> DeltaResult<Arro
     let fields = meta
         .schema()?
         .fields()
-        .iter()
         .filter(|f| !meta.partition_columns.contains(&f.name().to_string()))
         .map(|f| f.try_into())
         .chain(
             meta.schema()?
                 .fields()
-                .iter()
                 .filter(|f| meta.partition_columns.contains(&f.name().to_string()))
                 .map(|f| {
                     let field = Field::try_from(f)?;
@@ -2049,7 +2045,7 @@ mod tests {
 
         let table = crate::DeltaOps::new_in_memory()
             .create()
-            .with_columns(get_delta_schema().fields().clone())
+            .with_columns(get_delta_schema().fields().cloned())
             .with_partition_columns(["modified", "id"])
             .await
             .unwrap();

crates/core/src/errors.rs

@@ -11,6 +11,9 @@ pub type DeltaResult<T> = Result<T, DeltaTableError>;
 #[allow(missing_docs)]
 #[derive(thiserror::Error, Debug)]
 pub enum DeltaTableError {
+    #[error("Kernel error: {0}")]
+    KernelError(#[from] delta_kernel::error::Error),
+
     #[error("Delta protocol violation: {source}")]
     Protocol { source: ProtocolError },