fix(data-imports): force nullable on delta columns added via schema evolution

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py

@@ -305,8 +305,14 @@ async def _evolve_delta_schema(self, schema: pa.Schema) -> deltalake.DeltaTable:
 
         delta_table_schema = pyarrow_schema_from_arrow_exportable(delta_table.schema())
 
+        # Columns added here always predate their own addition: every file the table already
+        # holds was written without this column, so it must tolerate absent values on those
+        # rows. Forcing nullable regardless of the incoming batch's own nullability (which
+        # reflects only whether *this* batch happened to contain nulls) is what lets a later
+        # `optimize.compact()` read those old files at all — a non-nullable add otherwise fails
+        # compaction with "Non-nullable column '<name>' is missing from the physical schema".
         new_fields = [
-            deltalake.Field.from_arrow(field)
+            deltalake.Field.from_arrow(field.with_nullable(True))
             for field in ensure_delta_compatible_arrow_schema(schema)
             if field.name not in delta_table_schema.names
         ]

products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py

@@ -762,6 +762,43 @@ async def test_integer_overflow_batch_raises_clean_non_retryable(self, tmp_path:
             )
 
 
+class TestSchemaEvolutionNullability:
+    """A column added mid-table-lifetime always predates its own addition: every file the
+    table already holds was written without it, so `optimize.compact()` must be able to
+    treat those rows as null for that column. If schema evolution adds the column as NOT
+    NULL — which happens whenever the batch that introduces it has no nulls, since delta-rs
+    takes the new field's nullability straight from the incoming Arrow field — compaction
+    later fails with "Non-nullable column '<name>' is missing from the physical schema"."""
+
+    @pytest.mark.asyncio
+    async def test_compact_survives_a_column_added_by_an_all_non_null_batch(self, tmp_path: Path) -> None:
+        delta_path = str(tmp_path / "table")
+        deltalake.write_deltalake(delta_path, pa.table({"id": pa.array([1, 2], type=pa.int64())}))
+
+        helper = _make_local_helper(delta_path)
+
+        # The incoming field is non-nullable because every value in *this* batch is
+        # non-null — exactly how upstream Arrow construction infers it, unrelated to
+        # whether the column can appear in prior or future batches.
+        fields: list[pa.Field] = [pa.field("id", pa.int64()), pa.field("status", pa.string(), nullable=False)]
+        batch_schema = pa.schema(fields)
+        batch = pa.table(
+            {"id": pa.array([3, 4], type=pa.int64()), "status": pa.array(["ok", "ok"])}, schema=batch_schema
+        )
+
+        result = await helper.write_to_deltalake(
+            data=batch, write_type="append", should_overwrite_table=False, primary_keys=None
+        )
+        status_field = next(f for f in result.schema().fields if f.name == "status")
+        assert status_field.nullable is True
+
+        await helper.compact_table()
+
+        final = result.to_pyarrow_table()
+        by_id = dict(zip(final.column("id").to_pylist(), final.column("status").to_pylist()))
+        assert by_id == {1: None, 2: None, 3: "ok", 4: "ok"}
+
+
 class TestIncrementalBatchDeduplication:
     """Duplicate PKs in a source batch must never reach the Delta write.