feat: implement builder for StructType (#1492)

## What changes are proposed in this pull request?
Fixes #1284



## How was this change tested?
Added couple of unit tests

---------

Co-authored-by: aleksandarskrbic <aleksandarskrbic@users.noreply.github.com>

Author: aleksandarskrbic

kernel/src/schema/mod.rs

@@ -547,6 +547,43 @@ pub struct StructType {
     metadata_columns: HashMap<MetadataColumnSpec, usize>,
 }
 
+pub struct StructTypeBuilder {
+    fields: IndexMap<String, StructField>,
+}
+
+impl Default for StructTypeBuilder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl StructTypeBuilder {
+    pub fn new() -> Self {
+        Self {
+            fields: IndexMap::new(),
+        }
+    }
+
+    pub fn from_schema(schema: &StructType) -> Self {
+        Self {
+            fields: schema.fields.clone(),
+        }
+    }
+
+    pub fn add_field(mut self, field: StructField) -> Self {
+        self.fields.insert(field.name.clone(), field);
+        self
+    }
+
+    pub fn build(self) -> DeltaResult<StructType> {
+        StructType::try_new(self.fields.into_values())
+    }
+
+    pub fn build_arc_unchecked(self) -> Arc<StructType> {
+        Arc::new(StructType::new_unchecked(self.fields.into_values()))
+    }
+}
+
 impl StructType {
     /// Creates a new [`StructType`] from the given fields.
     ///
@@ -600,6 +637,10 @@ impl StructType {
             .process_results(|iter| Self::try_new(iter))?
     }
 
+    pub fn builder() -> StructTypeBuilder {
+        StructTypeBuilder::new()
+    }
+
     /// Creates a new [`StructType`] from the given fields without validating them.
     ///
     /// This should only be used when you are sure that the fields are valid.
@@ -3158,4 +3199,38 @@ mod tests {
 ");
         Ok(())
     }
+
+    #[test]
+    fn test_builder_empty() {
+        let schema = StructType::builder().build().unwrap();
+        assert_eq!(schema.num_fields(), 0)
+    }
+
+    #[test]
+    fn test_builder_add_fields() {
+        let schema = StructType::builder()
+            .add_field(StructField::new("id", DataType::INTEGER, false))
+            .add_field(StructField::new("name", DataType::STRING, true))
+            .build()
+            .unwrap();
+
+        assert_eq!(schema.num_fields(), 2);
+        assert_eq!(schema.field_at_index(0).unwrap().name(), "id");
+        assert_eq!(schema.field_at_index(1).unwrap().name(), "name");
+    }
+
+    #[test]
+    fn test_builder_from_schema() {
+        let base_schema =
+            StructType::try_new([StructField::new("id", DataType::INTEGER, false)]).unwrap();
+
+        let extended_schema = StructTypeBuilder::from_schema(&base_schema)
+            .add_field(StructField::new("name", DataType::STRING, true))
+            .build()
+            .unwrap();
+
+        assert_eq!(extended_schema.num_fields(), 2);
+        assert_eq!(extended_schema.field_at_index(0).unwrap().name(), "id");
+        assert_eq!(extended_schema.field_at_index(1).unwrap().name(), "name");
+    }
 }

kernel/src/transaction/mod.rs

@@ -24,7 +24,7 @@ use crate::scan::log_replay::{
     BASE_ROW_ID_NAME, DEFAULT_ROW_COMMIT_VERSION_NAME, FILE_CONSTANT_VALUES_NAME, TAGS_NAME,
 };
 use crate::scan::scan_row_schema;
-use crate::schema::{ArrayType, MapType, SchemaRef, StructField, StructType};
+use crate::schema::{ArrayType, MapType, SchemaRef, StructField, StructType, StructTypeBuilder};
 use crate::snapshot::SnapshotRef;
 use crate::utils::{current_time_ms, require};
 use crate::{
@@ -66,9 +66,9 @@ pub(crate) static BASE_ADD_FILES_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(||
         DataType::struct_type_unchecked(vec![StructField::nullable("numRecords", DataType::LONG)]),
     );
 
-    Arc::new(StructType::new_unchecked(
-        mandatory_add_file_schema().fields().cloned().chain([stats]),
-    ))
+    StructTypeBuilder::from_schema(mandatory_add_file_schema())
+        .add_field(stats)
+        .build_arc_unchecked()
 });
 
 static DATA_CHANGE_COLUMN: LazyLock<StructField> =
@@ -86,30 +86,28 @@ static ADD_FILES_SCHEMA_WITH_DATA_CHANGE: LazyLock<SchemaRef> = LazyLock::new(||
     Arc::new(StructType::new_unchecked(fields.into_iter().cloned()))
 });
 
-// NOTE: The following two methods are a workaround for the fact that we do not have a proper SchemaBuilder yet.
-// See https://github.com/delta-io/delta-kernel-rs/issues/1284
 /// Extend a schema with a statistics column and return a new SchemaRef.
 ///
 /// The stats column is of type string as required by the spec.
 ///
 /// Note that this method is only useful to extend an Add action schema.
 fn with_stats_col(schema: &SchemaRef) -> SchemaRef {
-    let fields = schema
-        .fields()
-        .cloned()
-        .chain([StructField::nullable("stats", DataType::STRING)]);
-    Arc::new(StructType::new_unchecked(fields))
+    StructTypeBuilder::from_schema(schema)
+        .add_field(StructField::nullable("stats", DataType::STRING))
+        .build_arc_unchecked()
 }
 
 /// Extend a schema with row tracking columns and return a new SchemaRef.
 ///
 /// Note that this method is only useful to extend an Add action schema.
 fn with_row_tracking_cols(schema: &SchemaRef) -> SchemaRef {
-    let fields = schema.fields().cloned().chain([
-        StructField::nullable("baseRowId", DataType::LONG),
-        StructField::nullable("defaultRowCommitVersion", DataType::LONG),
-    ]);
-    Arc::new(StructType::new_unchecked(fields))
+    StructTypeBuilder::from_schema(schema)
+        .add_field(StructField::nullable("baseRowId", DataType::LONG))
+        .add_field(StructField::nullable(
+            "defaultRowCommitVersion",
+            DataType::LONG,
+        ))
+        .build_arc_unchecked()
 }
 
 /// A transaction represents an in-progress write to a table. After creating a transaction, changes