Load package `schema_update` omits arrow tables on filesystem destination (trace under-reports schema migrations)

[sh-rp]

Observation

On a filesystem destination, a table loaded from arrow data does not appear in the load package's schema_update (the per-package schema migration recorded in the trace / load info). The same load on a SQL destination (e.g. duckdb) reports it correctly. This makes the trace under-report which tables/columns were created or altered by a load.

This surfaced for us via dlthub materialized transformations (materialization emits arrow, and delta forces materialization), but it reproduces with a plain pipeline.run(<pyarrow table>) — no transformations, and independent of table_format (plain parquet and delta both hit it).

Repro

import os, shutil, dlt, pyarrow as pa

def schema_update_tables(pipeline):
    for step in pipeline.last_trace.steps:
        if step.step != "load":
            continue
        for pkg in step.step_info.load_packages:
            return list(pkg.schema_update.keys())

data = pa.table({"id": [1, 2], "name": ["a", "b"]})

# filesystem — user table 'things' is MISSING
W = "/tmp/tkt_fs"; shutil.rmtree(W, ignore_errors=True); os.makedirs(W); os.chdir(W)
fs = dlt.pipeline("p", destination=dlt.destinations.filesystem(bucket_url=f"file://{W}/lake"), dataset_name="d")
fs.run(data, table_name="things")
print("filesystem:", schema_update_tables(fs))

# duckdb — user table 'things' is present
W2 = "/tmp/tkt_duck"; shutil.rmtree(W2, ignore_errors=True); os.makedirs(W2); os.chdir(W2)
dk = dlt.pipeline("p", destination="duckdb", dataset_name="d")
dk.run(data, table_name="things")
print("duckdb    :", schema_update_tables(dk))

Output:

filesystem: ['_dlt_pipeline_state']
duckdb    : ['_dlt_pipeline_state', '_dlt_loads', '_dlt_version', 'things']

The behavior is deterministic (not intermittent) and also affects schema evolution: a subsequent arrow load that adds a column to things produces an empty schema_update (tables=[]) on filesystem, while duckdb records the added column. Loading the same data as dict/JSON rows (instead of arrow) reports correctly on filesystem too.

Versions: dlt==1.27.0, deltalake==0.25.4, Python 3.12.11.

What is expected in the trace

Each load package carries a schema_update (serialized as tables in the load-package info / trace step info). It is meant to describe the schema migration that this load applied at the destination — the set of tables and columns that were newly created or added by this specific load. Consumers rely on it to answer "what changed at the destination in this run" (e.g. surfacing per-run schema/DDL migrations in observability tooling). For a freshly created data table with a load job, that table (with its columns) should be present in schema_update; for an evolved table, the added columns should be present — regardless of whether the data arrived as arrow or JSON, and regardless of destination type.

Why it isn't working (brief)

The load-package schema_update is derived from the diff the normalize step produces. For arrow input, the column schema is resolved at extract time (from the arrow schema), so normalize sees no change and records an empty diff. SQL destinations recompute the actually-applied update at load time in update_stored_schema (via _execute_schema_update_sql, diffing the schema against the destination's stored state), so they report the table anyway. The filesystem client's update_stored_schema instead returns the (empty) expected_update verbatim, so the table is never reported.

[sh-rp]

Note

This comment is Claude-generated (proposed fix directions). A dlt developer should validate these before implementation — the tradeoffs and edge cases below are hypotheses from reading the code, not verified fixes.

Two candidate fix directions:

Variant A — recompute the applied update in the filesystem client (localized).
Make FilesystemClient.update_stored_schema (dlt/destinations/impl/filesystem/filesystem.py) build applied_update by diffing the current self.schema (restricted to only_tables) against the previously-stored schema file, instead of returning expected_update verbatim. This mirrors what the SQL client already does in _execute_schema_update_sql / _build_schema_update_sql, reusing self.schema.get_new_table_columns(table_name, storage_columns, case_sensitive=...). The "previous state" would come from the newest stored schema for the schema name (_get_stored_schema_by_hash_or_newest(schema_name=...)) rather than DB introspection.

• Points to check: schema-evolution path needs the newest prior schema (a version_hash-only lookup returns None for a changed hash, so a second schema_name= lookup is needed); confirm only_tables already contains tables-with-jobs at the call site (dlt/load/utils.py init_client); confirm this is consistent with how other non-SQL clients report.

Variant B — fix the diff at its source in normalize (broad).
This is exactly the existing TODO in dlt/normalize/normalize.py (near the save_schema_updates call): "do not use schema_updates to generate a diff. compare the schema in the storage to the schema being saved and create unified diff. schema may change in several places (also extraction)." Computing schema_update as a real stored-vs-saved diff would make expected_update correct for all destinations at once (arrow tables resolve their columns during extract, which the current per-normalize-worker accumulation misses).

• Points to check: it's more invasive (parallel-worker retry path, and extract already persists schema changes to the local schema store, so "what extract added" may not be cleanly recoverable at normalize); needs care not to regress SQL destinations that currently recompute independently.

Variant A is the smaller, lower-risk change and keeps the destination as the source of truth for "what was actually applied here." Variant B addresses the root approximation but is a bigger change. A maintainer should decide which aligns with the intended contract of schema_update.