parse table metadata.configuration as TableProperties

ffi/src/lib.rs

@@ -382,6 +382,7 @@ pub enum KernelError {
     InvalidCommitInfo,
     FileAlreadyExists,
     MissingCommitInfo,
+    InvalidTableProperties,
     UnsupportedError,
 }
 
@@ -434,6 +435,7 @@ impl From<Error> for KernelError {
             Error::InvalidCommitInfo(_) => KernelError::InvalidCommitInfo,
             Error::FileAlreadyExists(_) => KernelError::FileAlreadyExists,
             Error::MissingCommitInfo => KernelError::MissingCommitInfo,
+            Error::InvalidTableProperties(_) => KernelError::InvalidTableProperties,
             Error::Unsupported(_) => KernelError::UnsupportedError,
         }
     }

kernel/src/actions/mod.rs

@@ -16,6 +16,7 @@ use crate::table_features::{
 };
 use crate::utils::require;
 use crate::{DeltaResult, EngineData, Error};
+use crate::table_properties::TableProperties;
 use visitors::{AddVisitor, MetadataVisitor, ProtocolVisitor};
 
 use delta_kernel_derive::Schema;
@@ -107,7 +108,7 @@ pub struct Metadata {
     pub partition_columns: Vec<String>,
     /// The time when this metadata action is created, in milliseconds since the Unix epoch
     pub created_time: Option<i64>,
-    /// Configuration options for the metadata action
+    /// Configuration options for the metadata action. These are parsed into [`TableProperties`].
     pub configuration: HashMap<String, String>,
 }
 
@@ -121,6 +122,11 @@ impl Metadata {
     pub fn schema(&self) -> DeltaResult<StructType> {
         Ok(serde_json::from_str(&self.schema_string)?)
     }
+
+    /// Parse the metadata configuration HashMap<String, String> into a TableProperties struct.
+    pub fn parse_table_properties(&self) -> DeltaResult<TableProperties> {
+        TableProperties::new(self.configuration.clone())
+    }
 }
 
 #[derive(Default, Debug, Clone, PartialEq, Eq, Schema, Serialize, Deserialize)]

kernel/src/error.rs

@@ -176,6 +176,10 @@ pub enum Error {
     #[error("File already exists: {0}")]
     FileAlreadyExists(String),
 
+    /// The `metadata` actions's `configuration` field was unable to parse into `TableProperties`
+    #[error("Invalid table properties: {0}")]
+    InvalidTableProperties(String),
+
     /// Some functionality is currently unsupported
     #[error("Unsupported: {0}")]
     Unsupported(String),
@@ -230,6 +234,9 @@ impl Error {
     pub(crate) fn invalid_log_path(msg: impl ToString) -> Self {
         Self::InvalidLogPath(msg.to_string())
     }
+    pub(crate) fn invalid_table_properties(msg: impl ToString) -> Self {
+        Self::InvalidTableProperties(msg.to_string())
+    }
 
     pub fn internal_error(msg: impl ToString) -> Self {
         Self::InternalError(msg.to_string()).with_backtrace()

kernel/src/lib.rs

@@ -63,8 +63,16 @@ pub mod actions;
 pub mod engine_data;
 pub mod error;
 pub mod expressions;
-pub(crate) mod predicates;
 pub mod table_features;
+pub mod scan;
+pub mod schema;
+pub mod snapshot;
+pub mod table;
+pub mod table_properties;
+pub mod transaction;
+
+pub(crate) mod predicates;
+pub(crate) mod utils;
 
 #[cfg(feature = "developer-visibility")]
 pub mod path;
@@ -76,13 +84,6 @@ pub mod log_segment;
 #[cfg(not(feature = "developer-visibility"))]
 pub(crate) mod log_segment;
 
-pub mod scan;
-pub mod schema;
-pub mod snapshot;
-pub mod table;
-pub mod transaction;
-pub(crate) mod utils;
-
 pub use delta_kernel_derive;
 pub use engine_data::{DataVisitor, EngineData};
 pub use error::{DeltaResult, Error};

kernel/src/scan/mod.rs

@@ -95,7 +95,7 @@ impl ScanBuilder {
         let (all_fields, read_fields, have_partition_cols) = get_state_info(
             logical_schema.as_ref(),
             &self.snapshot.metadata().partition_columns,
-            self.snapshot.column_mapping_mode,
+            self.snapshot.table_properties().get_column_mapping_mode(),
         )?;
         let physical_schema = Arc::new(StructType::new(read_fields));
 
@@ -251,7 +251,7 @@ impl Scan {
             partition_columns: self.snapshot.metadata().partition_columns.clone(),
             logical_schema: self.logical_schema.clone(),
             read_schema: self.physical_schema.clone(),
-            column_mapping_mode: self.snapshot.column_mapping_mode,
+            column_mapping_mode: self.snapshot.table_properties().get_column_mapping_mode(),
         }
     }
 

kernel/src/snapshot.rs

@@ -1,6 +1,5 @@
 //! In-memory representation of snapshots of tables (snapshot is a table at given point in time, it
 //! has schema etc.)
-//!
 
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -11,7 +10,7 @@ use crate::actions::{Metadata, Protocol};
 use crate::log_segment::LogSegment;
 use crate::scan::ScanBuilder;
 use crate::schema::Schema;
-use crate::table_features::{ColumnMappingMode, COLUMN_MAPPING_MODE_KEY};
+use crate::table_properties::TableProperties;
 use crate::{DeltaResult, Engine, Error, FileSystemClient, Version};
 
 const LAST_CHECKPOINT_FILE_NAME: &str = "_last_checkpoint";
@@ -26,7 +25,7 @@ pub struct Snapshot {
     metadata: Metadata,
     protocol: Protocol,
     schema: Schema,
-    pub(crate) column_mapping_mode: ColumnMappingMode,
+    table_properties: TableProperties,
 }
 
 impl Drop for Snapshot {
@@ -77,17 +76,14 @@ impl Snapshot {
     ) -> DeltaResult<Self> {
         let (metadata, protocol) = log_segment.read_metadata(engine)?;
         let schema = metadata.schema()?;
-        let column_mapping_mode = match metadata.configuration.get(COLUMN_MAPPING_MODE_KEY) {
-            Some(mode) if protocol.min_reader_version() >= 2 => mode.as_str().try_into(),
-            _ => Ok(ColumnMappingMode::None),
-        }?;
+        let table_properties = metadata.parse_table_properties()?;
         Ok(Self {
             table_root: location,
             log_segment,
             metadata,
             protocol,
             schema,
-            column_mapping_mode,
+            table_properties,
         })
     }
 
@@ -121,11 +117,9 @@ impl Snapshot {
         &self.protocol
     }
 
-    /// Get the [column mapping
-    /// mode](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#column-mapping) at this
-    /// `Snapshot`s version.
-    pub fn column_mapping_mode(&self) -> ColumnMappingMode {
-        self.column_mapping_mode
+    /// Get the [`TableProperties`] for this [`Snapshot`].
+    pub fn table_properties(&self) -> &TableProperties {
+        &self.table_properties
     }
 
     /// Create a [`ScanBuilder`] for an `Arc<Snapshot>`.

kernel/src/table_features/column_mapping.rs

@@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
 use crate::{DeltaResult, Error};
 
 /// Modes of column mapping a table can be in
-#[derive(Serialize, Deserialize, Debug, Copy, Clone, PartialEq)]
+#[derive(Serialize, Deserialize, Debug, Copy, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub enum ColumnMappingMode {
     /// No column mapping is applied
@@ -17,9 +17,6 @@ pub enum ColumnMappingMode {
     Name,
 }
 
-// key to look in metadata.configuration for to get column mapping mode
-pub(crate) const COLUMN_MAPPING_MODE_KEY: &str = "delta.columnMapping.mode";
-
 impl TryFrom<&str> for ColumnMappingMode {
     type Error = Error;
 

kernel/src/table_features/mod.rs

@@ -4,7 +4,6 @@ use std::sync::LazyLock;
 use serde::{Deserialize, Serialize};
 
 pub use column_mapping::ColumnMappingMode;
-pub(crate) use column_mapping::COLUMN_MAPPING_MODE_KEY;
 use strum::{AsRefStr, Display as StrumDisplay, EnumString, VariantNames};
 
 mod column_mapping;