feat!: error on surplus columns in output schema (#1488)

## What changes are proposed in this pull request?
Fixes [#1248](#1248)
by adding field count validation to Struct expression evaluation,
preventing silent data loss when the expression produces fewer fields
than the output schema expects.

The validation exposed a bug (test `test_append_partitioned` in
`kernel/tests/write.rs` started failing) in partitioned table writes
where the full logical schema (with partition columns) was incorrectly
used instead of the physical schema (without partition columns), which
is fixed by adding a `physical_schema` field to `WriteContext` that
properly excludes partition columns.


## How was this change tested?
Added addional test case `test_struct_expression_schema_validation` in
`kernel/src/engine/arrow_expression/evaluate_expression.rs::944`

---------

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

Author: aleksandarskrbic

ffi/src/transaction/write_context.rs

@@ -43,7 +43,7 @@ pub unsafe extern "C" fn get_write_schema(
     write_context: Handle<SharedWriteContext>,
 ) -> Handle<SharedSchema> {
     let write_context = unsafe { write_context.as_ref() };
-    write_context.schema().clone().into()
+    write_context.logical_schema().clone().into()
 }
 
 /// Get write path from WriteContext handle.

kernel/src/engine/arrow_expression/evaluate_expression.rs

@@ -102,6 +102,14 @@ fn evaluate_struct_expression(
     batch: &RecordBatch,
     output_schema: &StructType,
 ) -> DeltaResult<ArrayRef> {
+    if fields.len() != output_schema.num_fields() {
+        return Err(Error::generic(format!(
+            "Struct expression field count mismatch: {} fields in expression but {} in schema",
+            fields.len(),
+            output_schema.num_fields()
+        )));
+    }
+
     let output_cols: Vec<ArrayRef> = fields
         .iter()
         .zip(output_schema.fields())
@@ -931,6 +939,49 @@ mod tests {
             .contains("Data type is required"));
     }
 
+    #[test]
+    fn test_struct_expression_schema_validation() {
+        let batch = create_test_batch();
+
+        let test_cases = vec![
+            (
+                "too many schema fields",
+                Expr::Struct(vec![column_expr_ref!("a"), column_expr_ref!("b")]),
+                StructType::new_unchecked(vec![
+                    StructField::not_null("a", DataType::INTEGER),
+                    StructField::not_null("b", DataType::INTEGER),
+                    StructField::not_null("c", DataType::INTEGER),
+                ]),
+            ),
+            (
+                "too few schema fields",
+                Expr::Struct(vec![
+                    column_expr_ref!("a"),
+                    column_expr_ref!("b"),
+                    column_expr_ref!("c"),
+                ]),
+                StructType::new_unchecked(vec![
+                    StructField::not_null("a", DataType::INTEGER),
+                    StructField::not_null("b", DataType::INTEGER),
+                ]),
+            ),
+        ];
+
+        for (name, expr, schema) in test_cases {
+            let result =
+                evaluate_expression(&expr, &batch, Some(&DataType::Struct(Box::new(schema))));
+            assert!(result.is_err(), "Test case '{}' should fail", name);
+            assert!(
+                result
+                    .unwrap_err()
+                    .to_string()
+                    .contains("field count mismatch"),
+                "Test case '{}' should contain 'field count mismatch' error",
+                name
+            );
+        }
+    }
+
     #[test]
     fn test_coalesce_arrays_same_type() {
         // Test with Int32 arrays

kernel/src/engine/default/mod.rs

@@ -99,7 +99,7 @@ impl<E: TaskExecutor> DefaultEngine<E> {
     ) -> DeltaResult<Box<dyn EngineData>> {
         let transform = write_context.logical_to_physical();
         let input_schema = Schema::try_from_arrow(data.record_batch().schema())?;
-        let output_schema = write_context.schema();
+        let output_schema = write_context.physical_schema();
         let logical_to_physical_expr = self.evaluation_handler().new_expression_evaluator(
             input_schema.into(),
             transform.clone(),

kernel/src/transaction/mod.rs

@@ -557,9 +557,23 @@ impl Transaction {
         let target_dir = self.read_snapshot.table_root();
         let snapshot_schema = self.read_snapshot.schema();
         let logical_to_physical = self.generate_logical_to_physical();
+
+        // Compute physical schema: exclude partition columns since they're stored in the path
+        let partition_columns = self
+            .read_snapshot
+            .table_configuration()
+            .metadata()
+            .partition_columns();
+        let physical_fields = snapshot_schema
+            .fields()
+            .filter(|f| !partition_columns.contains(f.name()))
+            .cloned();
+        let physical_schema = Arc::new(StructType::new_unchecked(physical_fields));
+
         WriteContext::new(
             target_dir.clone(),
             snapshot_schema,
+            physical_schema,
             Arc::new(logical_to_physical),
         )
     }
@@ -874,15 +888,22 @@ impl Transaction {
 /// [`Transaction`]: struct.Transaction.html
 pub struct WriteContext {
     target_dir: Url,
-    schema: SchemaRef,
+    logical_schema: SchemaRef,
+    physical_schema: SchemaRef,
     logical_to_physical: ExpressionRef,
 }
 
 impl WriteContext {
-    fn new(target_dir: Url, schema: SchemaRef, logical_to_physical: ExpressionRef) -> Self {
+    fn new(
+        target_dir: Url,
+        logical_schema: SchemaRef,
+        physical_schema: SchemaRef,
+        logical_to_physical: ExpressionRef,
+    ) -> Self {
         WriteContext {
             target_dir,
-            schema,
+            logical_schema,
+            physical_schema,
             logical_to_physical,
         }
     }
@@ -891,8 +912,12 @@ impl WriteContext {
         &self.target_dir
     }
 
-    pub fn schema(&self) -> &SchemaRef {
-        &self.schema
+    pub fn logical_schema(&self) -> &SchemaRef {
+        &self.logical_schema
+    }
+
+    pub fn physical_schema(&self) -> &SchemaRef {
+        &self.physical_schema
     }
 
     pub fn logical_to_physical(&self) -> ExpressionRef {
@@ -1109,4 +1134,44 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_physical_schema_excludes_partition_columns() -> Result<(), Box<dyn std::error::Error>> {
+        let engine = SyncEngine::new();
+        let path = std::fs::canonicalize(PathBuf::from("./tests/data/basic_partitioned/")).unwrap();
+        let url = url::Url::from_directory_path(path).unwrap();
+        let snapshot = Snapshot::builder_for(url).build(&engine).unwrap();
+        let txn = snapshot
+            .transaction(Box::new(FileSystemCommitter::new()))?
+            .with_engine_info("default engine");
+
+        let write_context = txn.get_write_context();
+        let logical_schema = write_context.logical_schema();
+        let physical_schema = write_context.physical_schema();
+
+        // Logical schema should include the partition column
+        assert!(
+            logical_schema.contains("letter"),
+            "Logical schema should contain partition column 'letter'"
+        );
+
+        // Physical schema should exclude the partition column
+        assert!(
+            !physical_schema.contains("letter"),
+            "Physical schema should not contain partition column 'letter' (stored in path)"
+        );
+
+        // Both should contain the non-partition columns
+        assert!(
+            logical_schema.contains("number"),
+            "Logical schema should contain data column 'number'"
+        );
+
+        assert!(
+            physical_schema.contains("number"),
+            "Physical schema should contain data column 'number'"
+        );
+
+        Ok(())
+    }
 }