Fix/sparse schema and allow column drops

src/dlt_iceberg/schema_casting.py

@@ -12,6 +12,19 @@
 logger = logging.getLogger(__name__)
 
 
+def _missing_required_target_fields(
+    source_schema: pa.Schema,
+    target_schema: pa.Schema,
+) -> List[pa.Field]:
+    """Return non-nullable target fields that are absent from the source schema."""
+    source_field_names = {field.name for field in source_schema}
+    return [
+        target_field
+        for target_field in target_schema
+        if target_field.name not in source_field_names and not target_field.nullable
+    ]
+
+
 def ensure_iceberg_compatible_arrow_schema(schema: pa.Schema) -> pa.Schema:
     """
     Convert Arrow schema to Iceberg-compatible schema.
@@ -332,11 +345,17 @@ def validate_cast(
     source_field_names = {field.name for field in source_schema}
     for target_field in target_schema:
         if target_field.name not in source_field_names:
-            # Missing fields will be filled with nulls - this is usually OK
-            result.add_warning(
-                f"Field '{target_field.name}' exists in target but not in source "
-                f"(will be null)"
-            )
+            if target_field.nullable:
+                # Missing nullable fields will be filled with nulls.
+                result.add_warning(
+                    f"Field '{target_field.name}' exists in target but not in source "
+                    f"(will be null)"
+                )
+            else:
+                result.add_error(
+                    f"Required field '{target_field.name}' exists in target but not "
+                    f"in source schema"
+                )
 
     return result
 
@@ -418,6 +437,17 @@ def cast_table_safe(
     for warning in validation.warnings:
         logger.warning(f"Cast warning: {warning}")
 
+    missing_required_fields = _missing_required_target_fields(table.schema, target_schema)
+    if missing_required_fields:
+        error_msg = (
+            "Cannot cast table with missing required target fields:\n"
+            + "\n".join(
+                f"Required field '{field.name}' exists in target but not in source schema"
+                for field in missing_required_fields
+            )
+        )
+        raise CastingError(error_msg)
+
     # In strict mode, fail if there are errors
     if strict and not validation.is_safe():
         error_msg = "Cannot cast table safely. Errors:\n" + "\n".join(validation.errors)
@@ -436,6 +466,17 @@ def cast_table_safe(
         f"to schema with {len(target_schema)} fields"
     )
 
+    # Add null columns for any field in the target schema missing from the source.
+    # This handles sparse incoming data where the Iceberg table has columns that
+    # the current batch doesn't contain.
+    source_field_names = {field.name for field in table.schema}
+    for target_field in target_schema:
+        if target_field.name not in source_field_names:
+            table = table.append_column(
+                target_field,
+                pa.nulls(len(table), type=target_field.type),
+            )
+
     # Reorder columns to match target schema before casting
     # PyArrow's cast() requires fields to be in the same order
     target_field_names = [field.name for field in target_schema]

src/dlt_iceberg/schema_evolution.py

@@ -37,6 +37,19 @@ class SchemaEvolutionError(Exception):
     pass
 
 
+def _required_dropped_fields(
+    existing_schema: Schema,
+    dropped_fields: List[str],
+) -> List[str]:
+    """Return dropped existing fields that cannot be represented as sparse nulls."""
+    dropped_field_names = set(dropped_fields)
+    return [
+        field.name
+        for field in existing_schema.fields
+        if field.name in dropped_field_names and field.required
+    ]
+
+
 def can_promote_type(from_type: IcebergType, to_type: IcebergType) -> bool:
     """
     Check if one Iceberg type can be safely promoted to another.
@@ -144,13 +157,6 @@ def validate_schema_changes(
     """
     errors = []
 
-    # Check dropped columns
-    if dropped_fields and not allow_column_drops:
-        errors.append(
-            f"Columns dropped (not safe): {', '.join(dropped_fields)}. "
-            f"Dropping columns is not supported by default to prevent data loss."
-        )
-
     # Check type changes
     for field_name, old_type, new_type in type_changes:
         if not can_promote_type(old_type, new_type):
@@ -169,7 +175,8 @@ def validate_schema_changes(
 def apply_schema_evolution(
     table,
     added_fields: List[NestedField],
-    type_changes: List[Tuple[str, IcebergType, IcebergType]]
+    type_changes: List[Tuple[str, IcebergType, IcebergType]],
+    dropped_fields: Optional[List[str]] = None,
 ) -> None:
     """
     Apply schema evolution changes to an Iceberg table.
@@ -178,14 +185,16 @@ def apply_schema_evolution(
         table: PyIceberg table instance
         added_fields: New fields to add
         type_changes: Type promotions to apply
+        dropped_fields: Fields to remove from the schema
     """
-    if not added_fields and not type_changes:
+    if not added_fields and not type_changes and not dropped_fields:
         logger.info("No schema changes to apply")
         return
 
     logger.info(
         f"Applying schema evolution: "
-        f"{len(added_fields)} new columns, {len(type_changes)} type promotions"
+        f"{len(added_fields)} new columns, {len(type_changes)} type promotions, "
+        f"{len(dropped_fields or [])} dropped columns"
     )
 
     # Apply changes using update_schema transaction
@@ -208,6 +217,11 @@ def apply_schema_evolution(
                 field_type=new_type
             )
 
+        # Delete dropped columns
+        for field_name in (dropped_fields or []):
+            logger.info(f"  Dropping column: {field_name}")
+            update.delete_column(field_name)
+
     logger.info("Schema evolution applied successfully")
 
 
@@ -238,24 +252,54 @@ def evolve_schema_if_needed(
     added_fields, type_changes, dropped_fields = compare_schemas(
         existing_schema, new_schema
     )
+    missing_required_fields = (
+        _required_dropped_fields(existing_schema, dropped_fields)
+        if not allow_column_drops
+        else []
+    )
 
     # Log what we found
     if added_fields:
         logger.info(f"Detected {len(added_fields)} new columns: {[f.name for f in added_fields]}")
     if type_changes:
         logger.info(f"Detected {len(type_changes)} type changes: {[(name, str(old), str(new)) for name, old, new in type_changes]}")
     if dropped_fields:
-        logger.warning(f"Detected {len(dropped_fields)} dropped columns: {dropped_fields}")
+        if allow_column_drops:
+            logger.info(f"Detected {len(dropped_fields)} columns to drop: {dropped_fields}")
+        elif missing_required_fields:
+            logger.error(
+                f"Detected missing required columns in incoming data: "
+                f"{missing_required_fields}. Cannot fill required columns with nulls."
+            )
+        else:
+            logger.warning(
+                f"Detected {len(dropped_fields)} sparse columns (not in incoming data): "
+                f"{dropped_fields}. Columns will remain in schema; new rows will have nulls."
+            )
 
     # If no changes, nothing to do
     if not added_fields and not type_changes and not dropped_fields:
         logger.debug("No schema changes detected")
         return False
 
+    if missing_required_fields:
+        raise SchemaEvolutionError(
+            "Incoming data is missing required existing columns and cannot be "
+            "treated as sparse data: " + ", ".join(missing_required_fields)
+        )
+
     # Validate changes are safe
     validate_schema_changes(added_fields, type_changes, dropped_fields, allow_column_drops)
 
-    # Apply evolution
-    apply_schema_evolution(table, added_fields, type_changes)
+    # When allow_column_drops=False and only dropped fields were detected,
+    # the table schema is already correct — no evolution needed.
+    if not allow_column_drops and not added_fields and not type_changes:
+        return False
+
+    # Apply evolution, passing dropped_fields only when allow_column_drops=True
+    apply_schema_evolution(
+        table, added_fields, type_changes,
+        dropped_fields=dropped_fields if allow_column_drops else None,
+    )
 
     return True

tests/test_schema_casting.py

@@ -370,6 +370,23 @@ def test_extra_field_in_target_warning():
     assert "not in source" in validation.warnings[0].lower()
 
 
+def test_required_extra_field_in_target_error():
+    """Test that missing required target fields are unsafe."""
+    source_schema = pa.schema([
+        pa.field("id", pa.int64(), nullable=False),
+    ])
+    target_schema = pa.schema([
+        pa.field("id", pa.int64(), nullable=False),
+        pa.field("name", pa.string(), nullable=False),
+    ])
+
+    validation = validate_cast(source_schema, target_schema)
+
+    assert not validation.is_safe()
+    assert any("required field 'name'" in err.lower() for err in validation.errors)
+    assert not validation.warnings
+
+
 def test_complex_multi_field_validation():
     """Test validation with multiple fields and mixed safety."""
     source_schema = pa.schema([

tests/test_schema_evolution.py

@@ -280,13 +280,9 @@ def test_schema_evolution_unsafe_changes():
     assert "value" in dropped, "Should detect 'value' was dropped"
     print(f"   Detected dropped columns: {dropped}")
 
-    # Should raise error by default
-    try:
-        validate_schema_changes(added, type_changes, dropped, allow_column_drops=False)
-        assert False, "Should have raised SchemaEvolutionError for dropped column"
-    except SchemaEvolutionError as e:
-        assert "value" in str(e).lower() or "dropped" in str(e).lower()
-        print(f"   Correctly rejected: {str(e)}")
+    # Dropped columns should not raise - sparse data is handled by filling nulls
+    validate_schema_changes(added, type_changes, dropped, allow_column_drops=False)
+    print(f"   Sparse columns accepted (nulls will be filled at write time)")
 
     # Test 2: Unsafe type narrowing should be detected
     print("\nTest 2: Unsafe type narrowing detection")
@@ -377,5 +373,45 @@ def test_schema_evolution_unit_compare():
     print("\n   All unit tests passed")
 
 
+def test_evolve_schema_rejects_missing_required_before_applying_changes():
+    """
+    Missing required existing columns are rejected before adding/promoting columns.
+    """
+    from dlt_iceberg.schema_evolution import (
+        evolve_schema_if_needed,
+        SchemaEvolutionError,
+    )
+
+    existing_schema = Schema(
+        NestedField(1, "id", IntegerType(), required=True),
+        NestedField(2, "name", StringType(), required=False),
+    )
+    incoming_schema = Schema(
+        NestedField(2, "name", StringType(), required=False),
+        NestedField(3, "new_col", StringType(), required=False),
+    )
+
+    class RejectingTable:
+        def __init__(self, schema):
+            self._schema = schema
+            self.update_schema_called = False
+
+        def schema(self):
+            return self._schema
+
+        def update_schema(self):
+            self.update_schema_called = True
+            raise AssertionError("schema evolution should not be applied")
+
+    table = RejectingTable(existing_schema)
+
+    with pytest.raises(SchemaEvolutionError) as exc_info:
+        evolve_schema_if_needed(table, incoming_schema, allow_column_drops=False)
+
+    assert "missing required existing columns" in str(exc_info.value).lower()
+    assert "id" in str(exc_info.value).lower()
+    assert not table.update_schema_called
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v", "-s"])