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

Author: zachschuermann

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.logical_schema().clone().into()
+    write_context.schema().clone().into()
 }
 
 /// Get write path from WriteContext handle.

kernel/src/engine/arrow_expression/evaluate_expression.rs

@@ -102,14 +102,6 @@ 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())
@@ -939,49 +931,6 @@ 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.physical_schema();
+        let output_schema = write_context.schema();
         let logical_to_physical_expr = self.evaluation_handler().new_expression_evaluator(
             input_schema.into(),
             transform.clone(),

kernel/src/transaction/mod.rs

@@ -555,23 +555,9 @@ 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),
         )
     }
@@ -886,22 +872,15 @@ impl Transaction {
 /// [`Transaction`]: struct.Transaction.html
 pub struct WriteContext {
     target_dir: Url,
-    logical_schema: SchemaRef,
-    physical_schema: SchemaRef,
+    schema: SchemaRef,
     logical_to_physical: ExpressionRef,
 }
 
 impl WriteContext {
-    fn new(
-        target_dir: Url,
-        logical_schema: SchemaRef,
-        physical_schema: SchemaRef,
-        logical_to_physical: ExpressionRef,
-    ) -> Self {
+    fn new(target_dir: Url, schema: SchemaRef, logical_to_physical: ExpressionRef) -> Self {
         WriteContext {
             target_dir,
-            logical_schema,
-            physical_schema,
+            schema,
             logical_to_physical,
         }
     }
@@ -910,12 +889,8 @@ impl WriteContext {
         &self.target_dir
     }
 
-    pub fn logical_schema(&self) -> &SchemaRef {
-        &self.logical_schema
-    }
-
-    pub fn physical_schema(&self) -> &SchemaRef {
-        &self.physical_schema
+    pub fn schema(&self) -> &SchemaRef {
+        &self.schema
     }
 
     pub fn logical_to_physical(&self) -> ExpressionRef {
@@ -1132,44 +1107,4 @@ 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(())
-    }
 }