refactor!: use kernel TableProperties

Signed-off-by: Robert Pack <[email protected]>

Author: roeap

crates/core/src/delta_datafusion/mod.rs

@@ -95,6 +95,7 @@ use crate::kernel::{
 };
 use crate::logstore::LogStoreRef;
 use crate::table::builder::ensure_table_uri;
+use crate::table::config::TablePropertiesExt as _;
 use crate::table::state::DeltaTableState;
 use crate::table::{Constraint, GeneratedColumn};
 use crate::{open_table, open_table_with_storage_options, DeltaTable};

crates/core/src/kernel/snapshot/mod.rs

@@ -26,6 +26,7 @@ use delta_kernel::path::{LogPathFileType, ParsedLogPath};
 use delta_kernel::scan::scan_row_schema;
 use delta_kernel::schema::SchemaRef;
 use delta_kernel::snapshot::Snapshot as KernelSnapshot;
+use delta_kernel::table_properties::TableProperties;
 use delta_kernel::{PredicateRef, Version};
 use futures::stream::BoxStream;
 use futures::{StreamExt, TryStreamExt};
@@ -42,7 +43,6 @@ use crate::kernel::parse::read_removes;
 use crate::kernel::transaction::CommitData;
 use crate::kernel::{ActionType, Add, StructType};
 use crate::logstore::{LogStore, LogStoreExt};
-use crate::table::config::TableConfig;
 use crate::{DeltaResult, DeltaTableConfig, DeltaTableError};
 
 pub use self::log_data::*;
@@ -208,8 +208,8 @@ impl Snapshot {
     }
 
     /// Well known table configuration
-    pub fn table_config(&self) -> TableConfig<'_> {
-        TableConfig(self.inner.metadata().configuration())
+    pub fn table_config(&self) -> &TableProperties {
+        self.inner.table_properties()
     }
 
     /// Get the active files for the current snapshot.
@@ -514,7 +514,7 @@ impl EagerSnapshot {
     }
 
     /// Well known table configuration
-    pub fn table_config(&self) -> TableConfig<'_> {
+    pub fn table_config(&self) -> &TableProperties {
         self.snapshot.table_config()
     }
 

crates/core/src/kernel/transaction/conflict_checker.rs

@@ -1,6 +1,8 @@
 //! Helper module to check if a transaction can be committed in case of conflicting commits.
 use std::collections::HashSet;
 
+use delta_kernel::table_properties::IsolationLevel;
+
 use super::CommitInfo;
 #[cfg(feature = "datafusion")]
 use crate::delta_datafusion::DataFusionMixins;
@@ -10,7 +12,7 @@ use crate::kernel::Transaction;
 use crate::kernel::{Action, Add, Metadata, Protocol, Remove};
 use crate::logstore::{get_actions, LogStore};
 use crate::protocol::DeltaOperation;
-use crate::table::config::IsolationLevel;
+use crate::table::config::TablePropertiesExt as _;
 use crate::DeltaTableError;
 
 #[cfg(feature = "datafusion")]

crates/core/src/kernel/transaction/mod.rs

@@ -79,6 +79,7 @@ use std::sync::Arc;
 use bytes::Bytes;
 use chrono::Utc;
 use conflict_checker::ConflictChecker;
+use delta_kernel::table_properties::TableProperties;
 use futures::future::BoxFuture;
 use object_store::path::Path;
 use object_store::Error as ObjectStoreError;
@@ -97,7 +98,7 @@ use crate::logstore::{CommitOrBytes, LogStoreRef};
 use crate::operations::CustomExecuteHandler;
 use crate::protocol::DeltaOperation;
 use crate::protocol::{cleanup_expired_logs_for, create_checkpoint_for};
-use crate::table::config::TableConfig;
+use crate::table::config::TablePropertiesExt as _;
 use crate::table::state::DeltaTableState;
 use crate::{crate_version, DeltaResult};
 
@@ -235,7 +236,7 @@ impl From<CommitBuilderError> for DeltaTableError {
 /// Reference to some structure that contains mandatory attributes for performing a commit.
 pub trait TableReference: Send + Sync {
     /// Well known table configuration
-    fn config(&self) -> TableConfig;
+    fn config(&self) -> &TableProperties;
 
     /// Get the table protocol of the snapshot
     fn protocol(&self) -> &Protocol;
@@ -256,7 +257,7 @@ impl TableReference for EagerSnapshot {
         EagerSnapshot::metadata(self)
     }
 
-    fn config(&self) -> TableConfig {
+    fn config(&self) -> &TableProperties {
         self.table_config()
     }
 
@@ -266,7 +267,7 @@ impl TableReference for EagerSnapshot {
 }
 
 impl TableReference for DeltaTableState {
-    fn config(&self) -> TableConfig {
+    fn config(&self) -> &TableProperties {
         self.snapshot.config()
     }
 
@@ -857,7 +858,7 @@ impl PostCommit {
             return Ok(false);
         }
 
-        let checkpoint_interval = table_state.config().checkpoint_interval() as i64;
+        let checkpoint_interval = table_state.config().checkpoint_interval().get() as i64;
         if ((version + 1) % checkpoint_interval) == 0 {
             create_checkpoint_for(version as u64, log_store.as_ref(), Some(operation_id)).await?;
             Ok(true)

crates/core/src/kernel/transaction/protocol.rs

@@ -8,6 +8,7 @@ use crate::kernel::{
     contains_timestampntz, Action, EagerSnapshot, Protocol, ProtocolExt as _, Schema,
 };
 use crate::protocol::DeltaOperation;
+use crate::table::config::TablePropertiesExt as _;
 use crate::table::state::DeltaTableState;
 
 static READER_V2: LazyLock<HashSet<ReaderFeature>> =

crates/core/src/operations/cdc.rs

@@ -2,6 +2,7 @@
 //! The CDC module contains private tools for managing CDC files
 //!
 
+use crate::table::config::TablePropertiesExt as _;
 use crate::table::state::DeltaTableState;
 use crate::DeltaResult;
 

crates/core/src/operations/constraints.rs

@@ -242,6 +242,7 @@ mod tests {
     use arrow_schema::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
     use datafusion::logical_expr::{col, lit};
 
+    use crate::table::config::TablePropertiesExt as _;
     use crate::writer::test_utils::{create_bare_table, get_arrow_schema, get_record_batch};
     use crate::{DeltaOps, DeltaResult, DeltaTable};
 

crates/core/src/operations/delete.rs

@@ -59,6 +59,7 @@ use crate::operations::write::execution::{write_execution_plan, write_execution_
 use crate::operations::write::WriterStatsConfig;
 use crate::operations::CustomExecuteHandler;
 use crate::protocol::DeltaOperation;
+use crate::table::config::TablePropertiesExt as _;
 use crate::table::state::DeltaTableState;
 use crate::{DeltaTable, DeltaTableError};
 
@@ -237,7 +238,8 @@ async fn execute_non_empty_expr(
         snapshot.table_config().num_indexed_cols(),
         snapshot
             .table_config()
-            .stats_columns()
+            .data_skipping_stats_columns
+            .as_ref()
             .map(|v| v.iter().map(|v| v.to_string()).collect::<Vec<String>>()),
     );
 
@@ -257,7 +259,7 @@ async fn execute_non_empty_expr(
             filter.clone(),
             table_partition_cols.clone(),
             log_store.object_store(Some(operation_id)),
-            Some(snapshot.table_config().target_file_size() as usize),
+            Some(snapshot.table_config().target_file_size().get() as usize),
             None,
             writer_properties.clone(),
             writer_stats_config.clone(),
@@ -293,7 +295,7 @@ async fn execute_non_empty_expr(
             cdc_filter,
             table_partition_cols.clone(),
             log_store.object_store(Some(operation_id)),
-            Some(snapshot.table_config().target_file_size() as usize),
+            Some(snapshot.table_config().target_file_size().get() as usize),
             None,
             writer_properties,
             writer_stats_config,

crates/core/src/operations/merge/mod.rs

@@ -91,6 +91,7 @@ use crate::operations::write::generated_columns::{
 };
 use crate::operations::write::WriterStatsConfig;
 use crate::protocol::{DeltaOperation, MergePredicate};
+use crate::table::config::TablePropertiesExt as _;
 use crate::table::state::DeltaTableState;
 use crate::{DeltaResult, DeltaTable, DeltaTableError};
 
@@ -1378,7 +1379,8 @@ async fn execute(
         snapshot.table_config().num_indexed_cols(),
         snapshot
             .table_config()
-            .stats_columns()
+            .data_skipping_stats_columns
+            .as_ref()
             .map(|v| v.iter().map(|v| v.to_string()).collect::<Vec<String>>()),
     );
 
@@ -1388,7 +1390,7 @@ async fn execute(
         write,
         table_partition_cols.clone(),
         log_store.object_store(Some(operation_id)),
-        Some(snapshot.table_config().target_file_size() as usize),
+        Some(snapshot.table_config().target_file_size().get() as usize),
         None,
         writer_properties.clone(),
         writer_stats_config.clone(),

crates/core/src/operations/mod.rs

@@ -7,6 +7,7 @@
 //! with a [data stream][datafusion::physical_plan::SendableRecordBatchStream],
 //! if the operation returns data as well.
 use async_trait::async_trait;
+use delta_kernel::table_properties::{DataSkippingNumIndexedCols, TableProperties};
 use std::collections::HashMap;
 use std::sync::Arc;
 use update_field_metadata::UpdateFieldMetadataBuilder;
@@ -36,6 +37,7 @@ use self::{
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::logstore::LogStoreRef;
 use crate::table::builder::DeltaTableBuilder;
+use crate::table::config::{TablePropertiesExt as _, DEFAULT_NUM_INDEX_COLS};
 use crate::DeltaTable;
 
 pub mod add_column;
@@ -333,19 +335,33 @@ impl AsRef<DeltaTable> for DeltaOps {
 /// If table_config does not exist (only can occur in the first write action) it takes
 /// the configuration that was passed to the writerBuilder.
 pub fn get_num_idx_cols_and_stats_columns(
-    config: Option<crate::table::config::TableConfig<'_>>,
+    config: Option<&TableProperties>,
     configuration: HashMap<String, Option<String>>,
-) -> (i32, Option<Vec<String>>) {
+) -> (DataSkippingNumIndexedCols, Option<Vec<String>>) {
     let (num_index_cols, stats_columns) = match &config {
-        Some(conf) => (conf.num_indexed_cols(), conf.stats_columns()),
+        Some(conf) => (
+            conf.num_indexed_cols(),
+            conf.data_skipping_stats_columns
+                .clone()
+                .map(|v| v.iter().map(|v| v.to_string()).collect::<Vec<String>>()),
+        ),
         _ => (
             configuration
                 .get("delta.dataSkippingNumIndexedCols")
-                .and_then(|v| v.clone().map(|v| v.parse::<i32>().unwrap()))
-                .unwrap_or(crate::table::config::DEFAULT_NUM_INDEX_COLS),
+                .and_then(|v| {
+                    v.as_ref()
+                        .and_then(|vv| vv.parse::<u64>().ok())
+                        .map(DataSkippingNumIndexedCols::NumColumns)
+                })
+                .unwrap_or(DataSkippingNumIndexedCols::NumColumns(
+                    DEFAULT_NUM_INDEX_COLS,
+                )),
             configuration
                 .get("delta.dataSkippingStatsColumns")
-                .and_then(|v| v.as_ref().map(|v| v.split(',').collect::<Vec<&str>>())),
+                .and_then(|v| {
+                    v.as_ref()
+                        .map(|v| v.split(',').map(|s| s.to_string()).collect::<Vec<String>>())
+                }),
         ),
     };
     (
@@ -360,14 +376,14 @@ pub fn get_num_idx_cols_and_stats_columns(
 /// If table_config does not exist (only can occur in the first write action) it takes
 /// the configuration that was passed to the writerBuilder.
 pub(crate) fn get_target_file_size(
-    config: &Option<crate::table::config::TableConfig<'_>>,
+    config: Option<&TableProperties>,
     configuration: &HashMap<String, Option<String>>,
-) -> i64 {
+) -> u64 {
     match &config {
-        Some(conf) => conf.target_file_size(),
+        Some(conf) => conf.target_file_size().get(),
         _ => configuration
             .get("delta.targetFileSize")
-            .and_then(|v| v.clone().map(|v| v.parse::<i64>().unwrap()))
+            .and_then(|v| v.clone().map(|v| v.parse::<u64>().unwrap()))
             .unwrap_or(crate::table::config::DEFAULT_TARGET_FILE_SIZE),
     }
 }