feat!: Introduce metadata column API

Author: lbhm

ffi/src/schema.rs

@@ -222,7 +222,7 @@ pub unsafe extern "C" fn visit_schema(
 fn visit_schema_impl(schema: &StructType, visitor: &mut EngineSchemaVisitor) -> usize {
     // Visit all the fields of a struct and return the list of children
     fn visit_struct_fields(visitor: &EngineSchemaVisitor, s: &StructType) -> usize {
-        let child_list_id = (visitor.make_field_list)(visitor.data, s.fields.len());
+        let child_list_id = (visitor.make_field_list)(visitor.data, s.num_fields());
         for field in s.fields() {
             visit_schema_item(
                 field.name(),

ffi/src/transaction/mod.rs

@@ -274,11 +274,17 @@ mod tests {
             // Ensure we get the correct schema
             let write_schema = unsafe { get_write_schema(write_context.shallow_copy()) };
             let write_schema_ref = unsafe { write_schema.as_ref() };
-            assert_eq!(write_schema_ref.fields.len(), 2);
-            assert_eq!(write_schema_ref.fields[0].name, "number");
-            assert_eq!(write_schema_ref.fields[0].data_type, DataType::INTEGER);
-            assert_eq!(write_schema_ref.fields[1].name, "string");
-            assert_eq!(write_schema_ref.fields[1].data_type, DataType::STRING);
+            assert_eq!(write_schema_ref.num_fields(), 2);
+            assert_eq!(write_schema_ref.field_at_index(0).unwrap().name, "number");
+            assert_eq!(
+                write_schema_ref.field_at_index(0).unwrap().data_type,
+                DataType::INTEGER
+            );
+            assert_eq!(write_schema_ref.field_at_index(1).unwrap().name, "string");
+            assert_eq!(
+                write_schema_ref.field_at_index(1).unwrap().data_type,
+                DataType::STRING
+            );
 
             // Ensure the ffi returns the correct table path
             let write_path = unsafe { get_write_path(write_context.shallow_copy(), allocate_str) };

kernel/src/actions/mod.rs

@@ -176,7 +176,7 @@ impl TryFrom<Format> for Scalar {
         )
         .map(Scalar::Map)?;
         Ok(Scalar::Struct(StructData::try_new(
-            Format::to_schema().fields().cloned().collect(),
+            Format::to_schema().into_fields().collect(),
             vec![provider, options],
         )?))
     }
@@ -190,6 +190,7 @@ impl TryFrom<Format> for Scalar {
 )]
 #[internal_api]
 pub(crate) struct Metadata {
+    // TODO: Make the struct fields private to force using the try_new function.
     /// Unique identifier for this table
     pub(crate) id: String,
     /// User-provided identifier for this table
@@ -219,6 +220,16 @@ impl Metadata {
         created_time: i64,
         configuration: HashMap<String, String>,
     ) -> DeltaResult<Self> {
+        // Validate that the schema does not contain metadata columns
+        // Note: We don't have to look for nested metadata columns because that is already validated
+        // when creating a StructType.
+        if let Some(metadata_field) = schema.fields().find(|field| field.is_metadata_column()) {
+            return Err(Error::Schema(format!(
+                "Table schema must not contain metadata columns. Found metadata column: '{}'",
+                metadata_field.name
+            )));
+        }
+
         Ok(Self {
             id: uuid::Uuid::new_v4().to_string(),
             name,

kernel/src/engine/arrow_expression/evaluate_expression.rs

@@ -179,11 +179,11 @@ fn evaluate_transform_expression(
     }
 
     // Verify that the lengths match before attempting to zip them below.
-    if output_cols.len() != output_schema.fields_len() {
+    if output_cols.len() != output_schema.num_fields() {
         return Err(Error::generic(format!(
             "Expression count ({}) doesn't match output schema field count ({})",
             output_cols.len(),
-            output_schema.fields_len()
+            output_schema.num_fields()
         )));
     }
 

kernel/src/engine/arrow_expression/mod.rs

@@ -182,7 +182,7 @@ impl Scalar {
                 // insert an entry for each child builder, even when we're inserting NULL.
                 let builder = builder_as!(array::StructBuilder);
                 require!(
-                    builder.num_fields() == stype.fields_len(),
+                    builder.num_fields() == stype.num_fields(),
                     Error::generic("Struct builder has wrong number of fields")
                 );
                 for _ in 0..num_rows {

kernel/src/engine/arrow_utils.rs

@@ -330,8 +330,8 @@ fn get_indices(
     fields: &ArrowFields,
     mask_indices: &mut Vec<usize>,
 ) -> DeltaResult<(usize, Vec<ReorderIndex>)> {
-    let mut found_fields = HashSet::with_capacity(requested_schema.fields.len());
-    let mut reorder_indices = Vec::with_capacity(requested_schema.fields.len());
+    let mut found_fields = HashSet::with_capacity(requested_schema.num_fields());
+    let mut reorder_indices = Vec::with_capacity(requested_schema.num_fields());
     let mut parquet_offset = start_parquet_offset;
     // for each field, get its position in the parquet (via enumerate), a reference to the arrow
     // field, and info about where it appears in the requested_schema, or None if the field is not
@@ -507,10 +507,17 @@ fn get_indices(
         }
     }
 
-    if found_fields.len() != requested_schema.fields.len() {
+    if found_fields.len() != requested_schema.num_fields() {
         // some fields are missing, but they might be nullable, need to insert them into the reorder_indices
         for (requested_position, field) in requested_schema.fields().enumerate() {
             if !found_fields.contains(field.name()) {
+                if let Some(metadata_spec) = field.get_metadata_column_spec() {
+                    // We don't support reading any metadata columns yet
+                    // TODO: Implement row index support for the Parquet reader
+                    return Err(Error::Generic(format!(
+                        "Metadata column {metadata_spec:?} is not supported by the default parquet reader"
+                    )));
+                }
                 if field.nullable {
                     debug!("Inserting missing and nullable field: {}", field.name());
                     reorder_indices.push(ReorderIndex::missing(
@@ -582,11 +589,12 @@ fn match_parquet_fields<'k, 'p>(
             // Map the parquet ArrowField to the matching kernel KernelFieldInfo if present.
             let kernel_field_info =
                 kernel_schema
-                    .fields
-                    .get_full(field_name)
-                    .map(|(idx, _name, field)| KernelFieldInfo {
-                        parquet_index: idx,
-                        field,
+                    .field_with_index(field_name)
+                    .and_then(|(idx, field)| {
+                        (!field.is_metadata_column()).then_some(KernelFieldInfo {
+                            parquet_index: idx,
+                            field,
+                        })
                     });
 
             MatchedParquetField {
@@ -1533,6 +1541,41 @@ mod tests {
         ]))
     }
 
+    #[test]
+    fn test_match_parquet_fields_filters_metadata_columns() {
+        use crate::schema::MetadataColumnSpec;
+
+        let kernel_schema = StructType::new_unchecked([
+            StructField::not_null("regular_field", DataType::INTEGER),
+            StructField::create_metadata_column("row_index", MetadataColumnSpec::RowIndex),
+            StructField::nullable("another_field", DataType::STRING),
+        ]);
+
+        let parquet_fields: ArrowFields = vec![
+            ArrowField::new("regular_field", ArrowDataType::Int32, false),
+            ArrowField::new("row_index", ArrowDataType::Int64, false),
+            ArrowField::new("another_field", ArrowDataType::Utf8, true),
+        ]
+        .into();
+
+        let matched_fields: Vec<_> =
+            match_parquet_fields(&kernel_schema, &parquet_fields).collect();
+
+        assert_eq!(matched_fields.len(), 3);
+
+        // First field (regular_field) should have kernel_field_info
+        assert!(matched_fields[0].kernel_field_info.is_some());
+        assert_eq!(matched_fields[0].parquet_field.name(), "regular_field");
+
+        // Second field (row_index metadata column) should have None for kernel_field_info
+        assert!(matched_fields[1].kernel_field_info.is_none());
+        assert_eq!(matched_fields[1].parquet_field.name(), "row_index");
+
+        // Third field (another_field) should have kernel_field_info
+        assert!(matched_fields[2].kernel_field_info.is_some());
+        assert_eq!(matched_fields[2].parquet_field.name(), "another_field");
+    }
+
     #[test]
     fn nested_indices() {
         column_mapping_cases().into_iter().for_each(|mode| {

kernel/src/engine/ensure_data_types.rs

@@ -110,7 +110,7 @@ impl EnsureDataTypes {
                 // build a list of kernel fields that matches the order of the arrow fields
                 let mapped_fields = arrow_fields
                     .iter()
-                    .filter_map(|f| kernel_fields.fields.get(f.name()));
+                    .filter_map(|f| kernel_fields.field(f.name()));
 
                 // keep track of how many fields we matched up
                 let mut found_fields = 0;
@@ -122,12 +122,11 @@ impl EnsureDataTypes {
                 }
 
                 // require that we found the number of fields that we requested.
-                require!(kernel_fields.fields.len() == found_fields, {
+                require!(kernel_fields.num_fields() == found_fields, {
                     let arrow_field_map: HashSet<&String> =
                         HashSet::from_iter(arrow_fields.iter().map(|f| f.name()));
                     let missing_field_names = kernel_fields
-                        .fields
-                        .keys()
+                        .field_names()
                         .filter(|kernel_field| !arrow_field_map.contains(kernel_field))
                         .take(5)
                         .join(", ");

kernel/src/row_tracking.rs

@@ -112,7 +112,7 @@ impl RowVisitor for RowTrackingVisitor {
 
     fn visit<'a>(&mut self, row_count: usize, getters: &[&'a dyn GetData<'a>]) -> DeltaResult<()> {
         require!(
-            getters.len() == add_files_schema().fields_len(),
+            getters.len() == add_files_schema().num_fields(),
             Error::generic(format!(
                 "Wrong number of RowTrackingVisitor getters: {}",
                 getters.len()
@@ -176,7 +176,7 @@ mod tests {
         let schema = add_files_schema();
         let mut getters: Vec<&'a dyn GetData<'a>> = Vec::new();
 
-        for i in 0..schema.fields_len() {
+        for i in 0..schema.num_fields() {
             if i == RowTrackingVisitor::NUM_RECORDS_FIELD_INDEX {
                 getters.push(num_records_mock);
             } else {

kernel/src/scan/log_replay.rs

@@ -127,8 +127,8 @@ impl AddRemoveDedupVisitor<'_> {
         field_idx: usize,
         partition_values: &HashMap<String, String>,
     ) -> DeltaResult<(usize, (String, Scalar))> {
-        let field = self.logical_schema.fields.get_index(field_idx);
-        let Some((_, field)) = field else {
+        let field = self.logical_schema.field_at_index(field_idx);
+        let Some(field) = field else {
             return Err(Error::InternalError(format!(
                 "out of bounds partition column field index {field_idx}"
             )));

kernel/src/scan/mod.rs

@@ -872,7 +872,9 @@ impl StateInfo {
         column_mapping_mode: ColumnMappingMode,
     ) -> DeltaResult<Self> {
         let mut have_partition_cols = false;
-        let mut read_fields = Vec::with_capacity(logical_schema.fields.len());
+        let mut read_fields = Vec::with_capacity(logical_schema.num_fields());
+        let mut read_field_names = HashSet::with_capacity(logical_schema.num_fields());
+
         // Loop over all selected fields and note if they are columns that will be read from the
         // parquet file ([`ColumnType::Selected`]) or if they are partition columns and will need to
         // be filled in by evaluating an expression ([`ColumnType::Partition`])
@@ -881,6 +883,13 @@ impl StateInfo {
             .enumerate()
             .map(|(index, logical_field)| -> DeltaResult<_> {
                 if partition_columns.contains(logical_field.name()) {
+                    if logical_field.is_metadata_column() {
+                        return Err(Error::Schema(format!(
+                            "Metadata column names must not match partition columns: {}",
+                            logical_field.name()
+                        )));
+                    }
+
                     // Store the index into the schema for this field. When we turn it into an
                     // expression in the inner loop, we will index into the schema and get the name and
                     // data type, which we need to properly materialize the column.
@@ -893,10 +902,26 @@ impl StateInfo {
                     debug!("\n\n{logical_field:#?}\nAfter mapping: {physical_field:#?}\n\n");
                     let physical_name = physical_field.name.clone();
                     read_fields.push(physical_field);
+
+                    if !logical_field.is_metadata_column() {
+                        read_field_names.insert(physical_name.clone());
+                    }
+
                     Ok(ColumnType::Selected(physical_name))
                 }
             })
             .try_collect()?;
+
+        // This iteration runs in O(3) time since each metadata column can appear at most once in the schema
+        for metadata_column in logical_schema.metadata_columns() {
+            if read_field_names.contains(metadata_column.name()) {
+                return Err(Error::Schema(format!(
+                    "Metadata column names must not match physical columns: {}",
+                    metadata_column.name()
+                )));
+            }
+        }
+
         Ok(StateInfo {
             all_fields,
             read_fields,