add merge test

Signed-off-by: Frank Portman <frank1214@gmail.com>

Author: FrankPortman

crates/core/src/operations/merge/mod.rs

@@ -1810,6 +1810,108 @@ mod tests {
 
         assert_merge(table, metrics).await;
     }
+    #[tokio::test]
+    async fn test_merge_preserves_nullability_without_schema_merge() {
+        // Test that nullability constraints are preserved when merge_schema is false (default)
+        let delta_schema = vec![
+            StructField::new(
+                "id".to_string(),
+                DataType::Primitive(PrimitiveType::String),
+                false, // non-nullable
+            ),
+            StructField::new(
+                "value".to_string(),
+                DataType::Primitive(PrimitiveType::Integer),
+                false, // non-nullable
+            ),
+            StructField::new(
+                "modified".to_string(),
+                DataType::Primitive(PrimitiveType::String),
+                true, // nullable
+            ),
+        ];
+
+        let table = DeltaOps::new_in_memory()
+            .create()
+            .with_save_mode(SaveMode::ErrorIfExists)
+            .with_columns(delta_schema)
+            .await
+            .unwrap();
+
+        // Verify initial schema nullability
+        let initial_fields: Vec<_> = table
+            .snapshot()
+            .unwrap()
+            .schema()
+            .fields()
+            .cloned()
+            .collect();
+        assert!(
+            !initial_fields[0].is_nullable(),
+            "id should be non-nullable"
+        );
+        assert!(
+            !initial_fields[1].is_nullable(),
+            "value should be non-nullable"
+        );
+        assert!(
+            initial_fields[2].is_nullable(),
+            "modified should be nullable"
+        );
+
+        // Source data with all nullable fields (typical from external sources)
+        let source_schema = Arc::new(ArrowSchema::new(vec![
+            Field::new("id", ArrowDataType::Utf8, true), // nullable in source
+            Field::new("value", ArrowDataType::Int32, true), // nullable in source
+            Field::new("modified", ArrowDataType::Utf8, true), // nullable in source
+        ]));
+
+        let ctx = SessionContext::new();
+        let batch = RecordBatch::try_new(
+            source_schema,
+            vec![
+                Arc::new(arrow::array::StringArray::from(vec![Some("A"), Some("B")])),
+                Arc::new(arrow::array::Int32Array::from(vec![Some(1), Some(2)])),
+                Arc::new(arrow::array::StringArray::from(vec![
+                    Some("2021-02-02"),
+                    None,
+                ])),
+            ],
+        )
+        .unwrap();
+        let source = ctx.read_batch(batch).unwrap();
+
+        let (merged_table, _) = DeltaOps(table)
+            .merge(source, col("target.id").eq(col("source.id")))
+            .with_source_alias("source")
+            .with_target_alias("target")
+            // merge_schema is false by default
+            .when_not_matched_insert(|insert| {
+                insert
+                    .set("id", col("source.id"))
+                    .set("value", col("source.value"))
+                    .set("modified", col("source.modified"))
+            })
+            .unwrap()
+            .await
+            .unwrap();
+
+        // Verify schema nullability is preserved after merge
+        let final_fields: Vec<_> = merged_table.snapshot().unwrap().schema().fields().collect();
+        assert!(
+            !final_fields[0].is_nullable(),
+            "id should remain non-nullable after merge"
+        );
+        assert!(
+            !final_fields[1].is_nullable(),
+            "value should remain non-nullable after merge"
+        );
+        assert!(
+            final_fields[2].is_nullable(),
+            "modified should remain nullable after merge"
+        );
+    }
+
     #[tokio::test]
     async fn test_merge_with_schema_merge_no_change_of_schema() {
         let (table, _) = setup().await;

crates/core/src/operations/write/mod.rs

@@ -2284,4 +2284,111 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_schema_preserved_with_replace_where() -> TestResult {
+        // Test that schema is preserved when using overwrite with predicate (replaceWhere)
+        use arrow_array::{BooleanArray, Int32Array, Int64Array, RecordBatch, StringArray};
+        use arrow_schema::{DataType, Field, Schema as ArrowSchema};
+        use std::sync::Arc;
+
+        // Create initial table with mixed nullability
+        let initial_schema = Arc::new(ArrowSchema::new(vec![
+            Field::new("id", DataType::Int64, false), // non-nullable
+            Field::new("name", DataType::Utf8, true), // nullable
+            Field::new("active", DataType::Boolean, false), // non-nullable
+            Field::new("count", DataType::Int32, false), // non-nullable
+        ]));
+
+        let initial_batch = RecordBatch::try_new(
+            initial_schema.clone(),
+            vec![
+                Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5])),
+                Arc::new(StringArray::from(vec![
+                    Some("Alice"),
+                    Some("Bob"),
+                    None,
+                    Some("David"),
+                    Some("Eve"),
+                ])),
+                Arc::new(BooleanArray::from(vec![true, false, true, false, true])),
+                Arc::new(Int32Array::from(vec![10, 20, 30, 40, 50])),
+            ],
+        )?;
+
+        let table = DeltaOps::new_in_memory()
+            .write(vec![initial_batch])
+            .with_save_mode(SaveMode::Overwrite)
+            .await?;
+
+        // Capture initial schema
+        let initial_fields: Vec<_> = table
+            .snapshot()
+            .unwrap()
+            .schema()
+            .fields()
+            .cloned()
+            .collect();
+
+        // Create new data with all nullable fields (typical from Pandas)
+        let new_schema = Arc::new(ArrowSchema::new(vec![
+            Field::new("id", DataType::Int64, true), // nullable in new data
+            Field::new("name", DataType::Utf8, true), // nullable
+            Field::new("active", DataType::Boolean, true), // nullable
+            Field::new("count", DataType::Int32, true), // nullable
+        ]));
+
+        let replacement_batch = RecordBatch::try_new(
+            new_schema.clone(),
+            vec![
+                Arc::new(Int64Array::from(vec![Some(2), Some(4)])), // Replace ids 2 and 4
+                Arc::new(StringArray::from(vec![Some("Bob2"), Some("David2")])),
+                Arc::new(BooleanArray::from(vec![Some(true), Some(true)])),
+                Arc::new(Int32Array::from(vec![Some(200), Some(400)])),
+            ],
+        )?;
+
+        // Use replaceWhere to selectively overwrite
+        let table = DeltaOps(table)
+            .write(vec![replacement_batch])
+            .with_save_mode(SaveMode::Overwrite)
+            .with_replace_where("id = 2 OR id = 4")
+            .await?;
+
+        // Verify schema is preserved
+        let final_fields: Vec<_> = table.snapshot().unwrap().schema().fields().collect();
+
+        for (i, field) in final_fields.iter().enumerate() {
+            assert_eq!(
+                field.is_nullable(),
+                initial_fields[i].is_nullable(),
+                "Field '{}' nullability should be preserved with replaceWhere",
+                field.name()
+            );
+        }
+
+        // Now test that constraints are still enforced with replaceWhere
+        let invalid_batch = RecordBatch::try_new(
+            new_schema,
+            vec![
+                Arc::new(Int64Array::from(vec![None, Some(3)])), // NULL in non-nullable id!
+                Arc::new(StringArray::from(vec![Some("Invalid"), Some("Valid")])),
+                Arc::new(BooleanArray::from(vec![Some(false), Some(false)])),
+                Arc::new(Int32Array::from(vec![Some(999), Some(333)])),
+            ],
+        )?;
+
+        let result = DeltaOps(table)
+            .write(vec![invalid_batch])
+            .with_save_mode(SaveMode::Overwrite)
+            .with_replace_where("id = 1 OR id = 3")
+            .await;
+
+        assert!(
+            result.is_err(),
+            "replaceWhere should still enforce non-nullable constraints"
+        );
+
+        Ok(())
+    }
 }