parse table metadata.configuration as TableProperties

kernel/src/actions/mod.rs

@@ -9,8 +9,9 @@ use visitors::{AddVisitor, MetadataVisitor, ProtocolVisitor};
 
 use self::deletion_vector::DeletionVectorDescriptor;
 use crate::actions::schemas::GetStructField;
-use crate::features::{ReaderFeatures, WriterFeatures};
 use crate::schema::{SchemaRef, StructType};
+use crate::table_features::{ReaderFeatures, WriterFeatures};
+use crate::table_properties::TableProperties;
 use crate::{DeltaResult, EngineData};
 
 pub mod deletion_vector;
@@ -99,7 +100,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>,
 }
 
@@ -113,6 +114,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 get_table_properties(&self) -> DeltaResult<TableProperties> {
+        TableProperties::new(&self.configuration)
+    }
 }
 
 #[derive(Default, Debug, Clone, PartialEq, Eq, Schema, Serialize, Deserialize)]

kernel/src/expressions/column_names.rs

@@ -2,9 +2,12 @@ use std::borrow::Borrow;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::ops::Deref;
+
+use serde::Deserialize;
+
 /// A (possibly nested) column name.
 // TODO: Track name as a path rather than a single string
-#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord)]
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Deserialize)]
 pub struct ColumnName {
     path: String,
 }

kernel/src/lib.rs

@@ -62,7 +62,7 @@ pub mod actions;
 pub mod engine_data;
 pub mod error;
 pub mod expressions;
-pub mod features;
+pub mod table_features;
 
 #[cfg(feature = "developer-visibility")]
 pub mod path;
@@ -73,6 +73,7 @@ pub mod scan;
 pub mod schema;
 pub mod snapshot;
 pub mod table;
+pub mod table_properties;
 pub mod transaction;
 pub(crate) mod utils;
 

kernel/src/scan/mod.rs

@@ -10,10 +10,10 @@ use url::Url;
 use crate::actions::deletion_vector::{split_vector, treemap_to_bools, DeletionVectorDescriptor};
 use crate::actions::{get_log_add_schema, get_log_schema, ADD_NAME, REMOVE_NAME};
 use crate::expressions::{ColumnName, Expression, ExpressionRef, Scalar};
-use crate::features::ColumnMappingMode;
 use crate::scan::state::{DvInfo, Stats};
 use crate::schema::{DataType, Schema, SchemaRef, StructField, StructType};
 use crate::snapshot::Snapshot;
+use crate::table_features::ColumnMappingMode;
 use crate::{DeltaResult, Engine, EngineData, Error, FileMeta};
 
 use self::log_replay::scan_action_iter;
@@ -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));
         Ok(Scan {
@@ -247,7 +247,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/scan/state.rs

@@ -8,8 +8,8 @@ use crate::{
         visitors::visit_deletion_vector_at,
     },
     engine_data::{GetData, TypedGetData},
-    features::ColumnMappingMode,
     schema::SchemaRef,
+    table_features::ColumnMappingMode,
     DataVisitor, DeltaResult, Engine, EngineData, Error,
 };
 use serde::{Deserialize, Serialize};

kernel/src/schema.rs

@@ -9,7 +9,7 @@ use indexmap::IndexMap;
 use itertools::Itertools;
 use serde::{Deserialize, Serialize};
 
-use crate::features::ColumnMappingMode;
+use crate::table_features::ColumnMappingMode;
 use crate::utils::require;
 use crate::{DeltaResult, Error};
 

kernel/src/snapshot.rs

@@ -12,10 +12,10 @@ use url::Url;
 
 use crate::actions::{get_log_schema, Metadata, Protocol, METADATA_NAME, PROTOCOL_NAME};
 use crate::expressions::column_expr;
-use crate::features::{ColumnMappingMode, COLUMN_MAPPING_MODE_KEY};
 use crate::path::ParsedLogPath;
 use crate::scan::ScanBuilder;
 use crate::schema::{Schema, SchemaRef};
+use crate::table_properties::TableProperties;
 use crate::utils::require;
 use crate::{DeltaResult, Engine, Error, FileMeta, FileSystemClient, Version};
 use crate::{EngineData, Expression, ExpressionRef};
@@ -134,7 +134,7 @@ pub struct Snapshot {
     metadata: Metadata,
     protocol: Protocol,
     schema: Schema,
-    pub(crate) column_mapping_mode: ColumnMappingMode,
+    table_properties: TableProperties,
 }
 
 impl Drop for Snapshot {
@@ -230,18 +230,20 @@ impl Snapshot {
             .read_metadata(engine)?
             .ok_or(Error::MissingMetadata)?;
         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),
-        }?;
+        // FIXME
+        // 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.get_table_properties()?;
         Ok(Self {
             table_root: location,
             log_segment,
             version,
             metadata,
             protocol,
             schema,
-            column_mapping_mode,
+            table_properties,
         })
     }
 
@@ -278,8 +280,8 @@ impl Snapshot {
     /// 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
+    pub fn table_properties(&self) -> &TableProperties {
+        &self.table_properties
     }
 
     /// Create a [`ScanBuilder`] for an `Arc<Snapshot>`.

kernel/src/features/column_mapping.rs → 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/features/mod.rs → kernel/src/table_features/mod.rs

@@ -1,7 +1,6 @@
 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;