From 7a134b2e3cae2c1ae995f598fa543c67b22cf80e Mon Sep 17 00:00:00 2001 From: Tom Owers Date: Wed, 29 Jul 2026 16:00:42 +0100 Subject: [PATCH] fix(data-imports): force nullable on delta columns added via schema evolution Columns added to a Delta table mid-lifetime always predate their own addition, so every file the table already holds was written without them. `_evolve_delta_schema` was adding new columns with whatever nullability the incoming Arrow batch happened to have, which is non-nullable whenever the introducing batch contains no nulls. A later `optimize.compact()` then fails trying to reconcile older files against a NOT NULL column they don't have at all. Force newly-added columns to nullable=True regardless of the incoming field's own nullability, since older files can never retroactively gain the column's values. Generated-By: PostHog Code Task-Id: d3c0f944-c36f-4a90-9ebe-e1c05b7a5c99 --- .../pipelines/core/delta_table_helper.py | 8 +++- .../core/test/test_delta_table_helper.py | 37 +++++++++++++++++++ 2 files changed, 44 insertions(+), 1 deletion(-) diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py index 4176f587bcc9..4648db53ea32 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py +++ b/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 '' 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 ] diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py index ed155fae90d2..32e1dc655f04 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py +++ b/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 '' 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.