From 8b5f2580eacdb8466751a784af9c67895c0e95d1 Mon Sep 17 00:00:00 2001 From: TiM Date: Fri, 23 Oct 2026 07:46:15 +1300 Subject: [PATCH 1/5] Incremental: Add the `upsert` and `cdc` strategies `--incremental-strategy upsert` inserts and updates by primary key and deletes nothing. `--incremental-strategy cdc` keeps the destination equal to the source, deleting the rows the source no longer holds, which until now only `replace` could do, by rewriting the whole table. Both are dlt merge strategies, so they unflatten onto a `merge` write disposition that names them, the same path `scd2` takes. The `scd2` special case becomes one named-merge-strategy path. cdc deletes whatever a run does not read, so it shares scd2's refusal of partial reads, and is accepted for SQL database sources only: other sources can keep a cursor or a default window of their own that cannot be seen before the read. A source that owns its disposition has cdc refused rather than silently dropped. Both strategies need a primary key, checked before the load so a keyless table gets a one-line error instead of dlt's traceback at the load step. The partial-read check also covers `--interval-start` / `--interval-end` on a `query:` table that binds them into its placeholders, which windows the read without an incremental key. That applies to scd2 too. dlt 1.31 is the minimum version, the first with the cdc strategy. --- .../config/vocabularies/omniload/accept.txt | 1 + docs/changelog.md | 10 + docs/commands/ingest.md | 2 +- docs/getting-started/incremental-loading.md | 60 ++- docs/supported-sources/clickhouse.md | 2 +- docs/supported-sources/cratedb.md | 3 +- docs/supported-sources/turso.md | 2 +- pyproject.toml | 2 +- src/omniload/api.py | 137 ++++-- src/omniload/model.py | 2 + .../filesystem/test_incremental_strategy.py | 19 +- tests/main/test_incremental_strategy_cdc.py | 431 ++++++++++++++++++ 12 files changed, 637 insertions(+), 34 deletions(-) create mode 100644 tests/main/test_incremental_strategy_cdc.py diff --git a/docs/_vale/styles/config/vocabularies/omniload/accept.txt b/docs/_vale/styles/config/vocabularies/omniload/accept.txt index 0b6d40cc0..aa4b2c720 100644 --- a/docs/_vale/styles/config/vocabularies/omniload/accept.txt +++ b/docs/_vale/styles/config/vocabularies/omniload/accept.txt @@ -54,3 +54,4 @@ Apprise [Cc]onfig cron ntfy +[Uu]pserts? diff --git a/docs/changelog.md b/docs/changelog.md index e6290cb7a..2e089f304 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -4,6 +4,15 @@ - Connector: Verified support for DuckDB 2.x +- **Incremental strategies `upsert` and `cdc`.** `upsert` inserts and updates by + primary key without deleting. `cdc` keeps the destination equal to the source, + deleting the rows the source no longer holds; it is accepted for SQL database + sources only, and like `scd2` it rejects options that read part of a table. Both + need a primary key, declared by the table or named with `--primary-key`. dlt 1.31 + is now the minimum version. `scd2` now also rejects `--interval-start` / + `--interval-end` on a `query:` table that binds them, which windowed the read the + same way. See [Incremental loading]. + ## 2026/10/21 v0.20.0 - Keep datetime normalization compatible with dlt 1.31 while retaining support for @@ -590,5 +599,6 @@ [dlt-filesystem]: https://github.com/dlt-contrib/dlt-filesystem [ingestr v0.14.155]: https://github.com/bruin-data/ingestr/tree/v0.14.155 +[Incremental loading]: getting-started/incremental-loading.md [Notifications]: getting-started/notifications.md [SCD2]: https://en.wikipedia.org/wiki/Slowly_changing_dimension#Type_2:_add_new_row diff --git a/docs/commands/ingest.md b/docs/commands/ingest.md index 6f61b359c..3f1403fdf 100644 --- a/docs/commands/ingest.md +++ b/docs/commands/ingest.md @@ -23,7 +23,7 @@ omniload ingest \ - `--dest-table TEXT`: Designates the destination table to save the data. If not specified, defaults to the value of `--source-table`. - `--incremental-key TEXT`: Identifies the key used for incremental data strategies. Defaults to `None`. -- `--incremental-strategy TEXT`: Defines the strategy for incremental updates. Options include `replace`, `append`, `delete+insert`, `merge`, or `scd2`. When omitted, the default is `append` for filesystem-family sources (`file://`, `s3://`, `gs://`, `az://`, `sftp://`, ...) and `replace` for ordinary sources; sources that manage their own incrementality (many SaaS/streaming sources) keep their own per-resource disposition instead of applying a run-level one. Filesystem-family sources honour an explicit `append` or `replace`, but reject the key-based strategies (`delete+insert`, `merge`, `scd2`), which need an incremental or merge key they can't supply. See [Filesystem sources](../getting-started/incremental-loading.md#filesystem-sources). +- `--incremental-strategy TEXT`: Defines the strategy for incremental updates. Options include `replace`, `append`, `delete+insert`, `merge`, `scd2`, `upsert`, or `cdc`. When omitted, the default is `append` for filesystem-family sources (`file://`, `s3://`, `gs://`, `az://`, `sftp://`, ...) and `replace` for ordinary sources; sources that manage their own incrementality (many SaaS/streaming sources) keep their own per-resource disposition instead of applying a run-level one. Filesystem-family sources honour an explicit `append` or `replace`, but reject the key-based strategies (`delete+insert`, `merge`, `scd2`, `upsert`), which need an incremental or merge key they can't supply. `cdc` is accepted for SQL database sources only. See [Filesystem sources](../getting-started/incremental-loading.md#filesystem-sources). - `--filesystem-incremental`: For filesystem-family sources, filters the metadata-only file listing by modification time and appends rows only from files not recorded by the previous run. This opt-in mode is append-only. With the default temporary pipeline directory, the destination must support dlt state sync; otherwise, reuse an explicit persistent `--pipelines-dir`. Equal-boundary files are deduplicated by file URL, while an older-mtime backfill needs `--full-refresh`. Can also be set via the `FILESYSTEM_INCREMENTAL` / `OMNILOAD_FILESYSTEM_INCREMENTAL` environment variables. Defaults to `False`. See [Filesystem sources](../getting-started/incremental-loading.md#filesystem-sources). - `--interval-start`: Sets the start of the interval for the incremental key. Defaults to `None`. - `--interval-end`: Sets the end of the interval for the incremental key. Defaults to `None`. diff --git a/docs/getting-started/incremental-loading.md b/docs/getting-started/incremental-loading.md index 7d9783602..c2aa03862 100644 --- a/docs/getting-started/incremental-loading.md +++ b/docs/getting-started/incremental-loading.md @@ -18,6 +18,8 @@ Before you use incremental loading, you should understand three important settin - `merge`: merge the new rows with the existing rows in the destination table, insert the new ones and update the existing ones with the new values. - `delete+insert`: delete the existing rows in the destination table that match the incremental key and then insert the new rows. - `scd2`: retire the existing row and open a new one when a row changes, keeping the history of both. + - `upsert`: insert the new rows and update the changed ones, matched on the primary key, without deleting any. + - `cdc`: make the destination equal to the source on every run, deleting the rows the source no longer holds. @@ -236,7 +238,7 @@ The `delete+insert` strategy is useful when you want to keep the destination tab SCD2 (slowly changing dimension type 2) keeps the history of every row rather than overwriting it. When a row changes in the source, the existing record is retired and a new one is opened alongside it, so the destination records what the data looked like at any point in time. Each run stamps two extra columns, `_dlt_valid_from` and `_dlt_valid_to`. The record that is currently active is the one with an empty `_dlt_valid_to`. -SCD2 spots a change by comparing the whole source table against the records it is holding open, and retires whatever it no longer finds there. Every run must therefore read the table in full, so `scd2` rejects `--incremental-key`, `--sql-limit`, and `--yield-limit`: each reads back part of the table, which SCD2 cannot tell apart from the rest of the rows having been deleted. The filesystem family rejects `scd2` altogether ([Filesystem sources](#filesystem-sources)). +SCD2 spots a change by comparing the whole source table against the records it is holding open, and retires whatever it no longer finds there. Every run must therefore read the table in full, so `scd2` rejects `--incremental-key`, `--sql-limit`, `--yield-limit`, and `--interval-start` / `--interval-end` on a `query:` table that uses the `:interval_start` / `:interval_end` placeholders: each reads back part of the table, which SCD2 cannot tell apart from the rest of the rows having been deleted. The filesystem family rejects `scd2` altogether ([Filesystem sources](#filesystem-sources)). Where the source table is an expression rather than a plain table name, such as a `query:` for a SQL source, that expression's result is the table SCD2 tracks and carries the same requirement: it must return the tracked rows in full on every run. An expression whose result varies for any reason other than the data changing, such as one carrying its own limit, retires the rows it omits. @@ -300,6 +302,60 @@ SCD2 compares the row values it receives, so anything that rewrites a value diff SCD2 compares rows by a hash that dlt computes only for row-oriented data, so it does not work on data read as Arrow tables. For SCD2, SQL sources are read with the `sqlalchemy` backend, which omniload selects for you. Passing `--sql-backend pyarrow` or `--sql-backend connectorx` alongside `scd2` is rejected, as is `scd2` on an `mmap://` source, which is Arrow whichever backend you name. ::: +(incremental-loading-upsert)= + +## Upsert + +Upsert inserts the source rows the destination does not have yet and updates the ones it already holds, matching them on the primary key. It deletes nothing: a row that disappears from the source stays in the destination. Where `merge` deletes the matching rows and inserts them again on most SQL destinations, `upsert` updates them in place. + +Upsert needs a primary key. A SQL table that declares one is used as it is; for any other table, name the key columns with `--primary-key` on every run. A run without either is rejected before anything is loaded. + +As it deletes nothing, `upsert` can be combined with `--incremental-key` to read only the rows that changed since the previous run. + +The following example upserts the `my_schema.some_data` table from Postgres into DuckDB, matching rows on `id`. + +```bash +omniload ingest \ + --source-uri 'postgresql://admin:admin@localhost:8837/web?sslmode=disable' \ + --source-table 'my_schema.some_data' \ + --dest-uri 'duckdb:///warehouse.duckdb' \ + --incremental-strategy upsert \ + --primary-key id +``` + +(incremental-loading-cdc)= + +## CDC + +CDC (change data capture) makes the destination table equal to the source table on every run. Rows new to the source are inserted, changed rows are updated, and rows the source no longer holds are deleted, all matched on the primary key. It keeps a replica in sync without the whole-table rewrite of `replace`. + +CDC deletes every row a run does not read, so every run must read the table in full. Like `scd2`, `cdc` therefore rejects `--incremental-key`, `--sql-limit`, `--yield-limit`, and `--interval-start` / `--interval-end` on a `query:` table that uses the `:interval_start` / `:interval_end` placeholders. A `query:` table must return all of its rows on every run: one carrying its own limit deletes the rows it leaves out. + +`cdc` is accepted for SQL database sources only. Other sources can narrow what they read by themselves, for example to a default date window, in a way omniload cannot see before reading, so `cdc` is rejected for them. That includes the filesystem family and the sources that manage their own incrementality. + +CDC needs a primary key, as upsert does. + +The following example keeps the `my_schema.some_data` table in DuckDB in sync with Postgres. + +```bash +omniload ingest \ + --source-uri 'postgresql://admin:admin@localhost:8837/web?sslmode=disable' \ + --source-table 'my_schema.some_data' \ + --dest-uri 'duckdb:///warehouse.duckdb' \ + --incremental-strategy cdc \ + --primary-key id +``` + +:::{caution} +- An empty source table is handled according to the SQL backend. With `sqlalchemy` and `pyarrow`, which is the default, a run that reads no rows loads nothing and deletes nothing, so the destination keeps the rows it had. With `connectorx`, the same run deletes every row in the destination. +- A destination table loaded earlier with `delete+insert` keeps its incremental key as a merge key, and a later `cdc` run then deletes only among the values of that key it has read. Add `--full-refresh` to the first `cdc` run into such a table. +- Masking a primary-key column with `--mask` changes the key rows are matched on. A mask that draws a fresh value per run deletes and inserts every row again on every run, and a mask that maps two keys to one value merges those rows. +::: + +:::{note} +Which destinations support `upsert` and `cdc` is up to dlt. With dlt 1.31, both work on BigQuery, Databricks, DuckDB 1.4 or later, MotherDuck, Microsoft SQL Server, Postgres, Snowflake, and Athena with Iceberg tables. ClickHouse, CrateDB, Redshift, Synapse, Turso, and the destinations written through SQLAlchemy, such as MySQL and SQLite, support neither. +::: + (incremental-loading-filesystem)= ## Filesystem sources @@ -383,7 +439,7 @@ omniload ingest \ --incremental-strategy replace ``` -The key-dependent strategies (`delete+insert`, `merge`, `scd2`) are rejected with an error, because filesystem sources expose no incremental or merge key for them to use. Use `append` or `replace` instead. +The key-dependent strategies (`delete+insert`, `merge`, `scd2`, `upsert`) are rejected with an error, because filesystem sources expose no incremental or merge key for them to use. `cdc` is rejected too, as for every source that is not a SQL database. Use `append` or `replace` instead. ### Resetting the destination with `--full-refresh` diff --git a/docs/supported-sources/clickhouse.md b/docs/supported-sources/clickhouse.md index 51c56c159..392d1fc15 100644 --- a/docs/supported-sources/clickhouse.md +++ b/docs/supported-sources/clickhouse.md @@ -56,5 +56,5 @@ omniload ingest \ --> :::{warning} -Clickhouse currently doesn't support `delete+insert`, `merge` or `scd2` incremental strategies. +Clickhouse currently doesn't support `cdc`, `delete+insert`, `merge`, `scd2` or `upsert` incremental strategies. ::: diff --git a/docs/supported-sources/cratedb.md b/docs/supported-sources/cratedb.md index 46680e71e..72bcc3048 100644 --- a/docs/supported-sources/cratedb.md +++ b/docs/supported-sources/cratedb.md @@ -122,7 +122,8 @@ uvx crash -c 'SELECT * FROM doc.sample' :::{warning} CrateDB supports the `replace` incremental materialization strategy, but -currently does not support the `delete+insert`, `merge`, or `scd2` strategies. +currently does not support the `cdc`, `delete+insert`, `merge`, `scd2`, or `upsert` +strategies. ::: ## Appendix diff --git a/docs/supported-sources/turso.md b/docs/supported-sources/turso.md index b406a746b..47388b9c9 100644 --- a/docs/supported-sources/turso.md +++ b/docs/supported-sources/turso.md @@ -90,7 +90,7 @@ default temporary pipelines directory, each run restores it from the server. :::{note} - **Dispositions**: the destination takes `append` and `replace` only. `merge`, - `delete+insert` and `scd2` are refused before anything is written, as are sources + `delete+insert`, `scd2`, `upsert` and `cdc` are refused before anything is written, as are sources that set a merge disposition on their own resources. dlt merges through a staging database it `ATTACH`es, which Turso treats as experimental, and with that enabled Turso resolves the merge's database-qualified correlated subquery against the wrong diff --git a/pyproject.toml b/pyproject.toml index 9069d2ed3..f9d3c8dfa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -84,7 +84,7 @@ dependencies = [ "databricks-sql-connector<5", "databricks-sqlalchemy<3", "dataclasses-json<0.7", - "dlt[deltalake]>=1.22,<1.32", + "dlt[deltalake]>=1.31,<1.32", "dlt-cratedb<0.2", "dlt-filesystem", "duckdb<2.1", diff --git a/src/omniload/api.py b/src/omniload/api.py index b683e4115..4dd13bb5e 100644 --- a/src/omniload/api.py +++ b/src/omniload/api.py @@ -84,6 +84,28 @@ def _mark(context: _RunContext | None, stage: Stage) -> None: ) +#: Strategies that are a dlt merge strategy named on a `merge` disposition, rather than a +#: write disposition of their own. +NAMED_MERGE_STRATEGIES = ( + IncrementalStrategy.scd2, + IncrementalStrategy.upsert, + IncrementalStrategy.cdc, +) + +#: Why each strategy that reads absence as deletion cannot take a partial read. +_ABSENCE_CONSEQUENCE = { + IncrementalStrategy.scd2: ( + "scd2 spots a change by comparing the whole source table against the records " + "it holds open, so a run that reads only part of the table retires every " + "record it leaves out." + ), + IncrementalStrategy.cdc: ( + "cdc syncs the destination to the whole source table, so a run that reads " + "only part of the table deletes every record it leaves out." + ), +} + + def _coerce(value, enum_cls): """Accept either an enum member or its string value (the CLI form).""" if value is None or isinstance(value, enum_cls): @@ -138,7 +160,8 @@ def _first_merging_resource(dlt_source, run_disposition) -> str | None: run-level disposition overrides every resource's hint (dlt applies it with ``apply_hints``), and without one each resource keeps its own, which sources that manage their incrementality set themselves. Merge is the string - ``"merge"`` or a dict whose ``disposition`` is ``"merge"`` (scd2's form). A + ``"merge"`` or a dict whose ``disposition`` is ``"merge"`` (the form a named + merge strategy such as scd2 takes). A callable hint is resolved per data item, so nothing here can show it does not merge, and it counts as merging. """ @@ -155,6 +178,26 @@ def check(res) -> None: return found[0] if found else None +def _require_primary_key(resource, strategy: IncrementalStrategy) -> None: + """Raise unless ``resource`` declares a primary key ``strategy`` can match rows on. + + A resource whose table hints depend on the data cannot be inspected before it is + read, so it is left to dlt's own check at the load step. + """ + from dlt.extract.exceptions import DataItemRequiredForDynamicTableHints + + try: + columns = resource.compute_table_schema().get("columns", {}) + except DataItemRequiredForDynamicTableHints: + return + if not any(column.get("primary_key") for column in columns.values()): + raise ValidationError( + f"Incremental strategy '{strategy.value}' needs a primary key to match " + f"records on, and the '{resource.name}' table declares none. Name its key " + "columns with '--primary-key'." + ) + + def run_ingest(**kwargs) -> LoadInfo | None: """Load data from ``source_uri`` into ``dest_uri`` and return the dlt ``LoadInfo``. @@ -541,6 +584,23 @@ def parse_columns(columns: list[str]) -> dict: column_hints: dict[str, TColumnSchema] = {} original_incremental_strategy = incremental_strategy + # cdc deletes every destination record a load leaves out, so it is only safe on a + # read of the whole table. A SQL table reads whole unless an option windows or + # truncates it, which the partial-read check below refuses. Other sources can keep a + # cursor or a default window of their own that nothing here can see before the read, + # so cdc is refused for them outright, ahead of the suppression that would otherwise + # drop it to `none` for sources that own their disposition. + if ( + original_incremental_strategy == IncrementalStrategy.cdc + and factory.source_scheme not in SQL_SOURCE_SCHEMES + ): + raise ValidationError( + "Incremental strategy 'cdc' is only supported for SQL database sources, " + f"and '{factory.source_scheme}' is not one. cdc deletes every destination " + "record a run does not read, so it needs a source omniload can read in " + "full." + ) + supports_filesystem_incremental = getattr( source, "supports_filesystem_incremental", lambda: False )() @@ -583,12 +643,12 @@ def parse_columns(columns: list[str]) -> dict: column_hints[column_name] = {"data_type": column_type} # `IncrementalStrategy` flattens dlt's two axes (a write disposition of - # skip/append/replace/merge, and a merge strategy of delete-insert/scd2/upsert/ - # insert-only) onto one, so its two merge-strategy values unflatten back onto dlt's - # shape here: both are a `merge` disposition carrying a strategy. `delete+insert` marks - # the incremental key as the merge key and relies on dlt's default strategy; `scd2` - # names its strategy on the disposition itself, applied below once the sources that own - # their own disposition have had their say. + # skip/append/replace/merge, and a merge strategy of delete-insert/scd2/upsert/cdc/ + # insert-only) onto one, so its merge-strategy values unflatten back onto dlt's shape + # here: each is a `merge` disposition carrying a strategy. `delete+insert` marks the + # incremental key as the merge key and relies on dlt's default strategy; `scd2`, + # `upsert` and `cdc` name their strategy on the disposition itself, applied below once + # the sources that own their own disposition have had their say. merge_key = None if incremental_strategy == IncrementalStrategy.delete_insert: merge_key = jr.incremental_key @@ -598,7 +658,7 @@ def parse_columns(columns: list[str]) -> dict: column_hints[jr.incremental_key] = {} column_hints[jr.incremental_key]["merge_key"] = True - elif incremental_strategy == IncrementalStrategy.scd2: + elif incremental_strategy in NAMED_MERGE_STRATEGIES: incremental_strategy = IncrementalStrategy.merge # TODO: What is this hash used for? @@ -716,8 +776,8 @@ def parse_columns(columns: list[str]) -> dict: pass elif original_incremental_strategy in ( IncrementalStrategy.merge, - IncrementalStrategy.scd2, IncrementalStrategy.delete_insert, + *NAMED_MERGE_STRATEGIES, ): raise ValidationError( f"Incremental strategy '{original_incremental_strategy.value}' needs an " @@ -734,25 +794,39 @@ def parse_columns(columns: list[str]) -> dict: if incremental_strategy is None: incremental_strategy = IncrementalStrategy.create_replace - # The strategy is final from here, so this is what actually reaches the load: scd2 as - # asked for, and not suppressed above by a source that owns its own disposition. - runs_scd2 = ( - original_incremental_strategy == IncrementalStrategy.scd2 + # The strategy is final from here, so this is what actually reaches the load: a named + # merge strategy as asked for, and not suppressed above by a source that owns its own + # disposition. + merge_strategy = ( + original_incremental_strategy + if original_incremental_strategy in NAMED_MERGE_STRATEGIES and incremental_strategy != IncrementalStrategy.none + else None ) + runs_scd2 = merge_strategy == IncrementalStrategy.scd2 + + # scd2 and cdc both read a record missing from the load as gone from the source: scd2 + # retires it, cdc deletes it. Anything that hands dlt less than the whole source state + # corrupts the destination rather than failing, so refuse it up front. + if merge_strategy in _ABSENCE_CONSEQUENCE: + # An incremental key windows the read, and either limit truncates it. An interval + # windows it too where a `query:` table binds it into its `:interval_start` / + # `:interval_end` placeholders, which needs no incremental key; on any other + # table it takes effect only through the incremental key. Whatever the run leaves + # out is absent from the batch. + def binds(placeholder: str) -> bool: + return source_table.startswith("query:") and placeholder in source_table - # scd2 has dlt compare the whole source state against the records it holds open, and - # retire whatever is missing from it. Anything that hands dlt less than that state, or - # rows it cannot hash, corrupts the history rather than failing, so refuse it up front. - if runs_scd2: - # An incremental key windows the read, and either limit truncates it. Whatever the - # run leaves out is absent from the batch and gets retired, though the source still - # holds it unchanged. partial_read = next( ( option for option, value in ( ("--incremental-key", jr.incremental_key), + ( + "--interval-start", + binds(":interval_start") and jr.interval_start, + ), + ("--interval-end", binds(":interval_end") and jr.interval_end), ("--sql-limit", jr.sql_limit), ("--yield-limit", jr.yield_limit), ) @@ -762,11 +836,12 @@ def parse_columns(columns: list[str]) -> dict: ) if partial_read: raise ValidationError( - f"Incremental strategy 'scd2' cannot be combined with '{partial_read}': " - "scd2 spots a change by comparing the whole source table against the " - "records it holds open, so a run that reads only part of the table retires " - "every record it leaves out." + f"Incremental strategy '{merge_strategy.value}' cannot be combined with " + f"'{partial_read}': {_ABSENCE_CONSEQUENCE[merge_strategy]}" ) + + # scd2 also hashes each row to spot a change, which some inputs cannot carry. + if runs_scd2: # A mask that draws a fresh value per run rewrites the rows scd2 compares, so an # untouched source row reads as a new version of itself on every load. unstable_masks = non_deterministic_masks(jr.mask) if jr.mask else [] @@ -851,8 +926,8 @@ def parse_columns(columns: list[str]) -> dict: data_item_format = "arrow" write_disposition: str | dict[str, str] | None = None - if runs_scd2: - write_disposition = {"disposition": "merge", "strategy": "scd2"} + if merge_strategy is not None: + write_disposition = {"disposition": "merge", "strategy": merge_strategy.value} elif incremental_strategy != IncrementalStrategy.none: write_disposition = incremental_strategy.value @@ -991,6 +1066,16 @@ def col_h(x): resource.for_each(dlt_source, col_h) + if ( + merge_strategy in (IncrementalStrategy.upsert, IncrementalStrategy.cdc) + and not jr.primary_key + ): + # dlt refuses both without a primary key, but only at the load step and as a + # traceback. A key given with `--primary-key` reaches dlt at `pipeline.run()` + # rather than on the resource, so it satisfies this outright; otherwise the + # resource must carry one, as a SQL table with a declared key does by reflection. + resource.for_each(dlt_source, lambda x: _require_primary_key(x, merge_strategy)) + if isinstance(destination, AthenaDestination) and jr.partition_by: hint.apply_athena_hints(dlt_source, jr.partition_by, column_hints) diff --git a/src/omniload/model.py b/src/omniload/model.py index d296a69e7..f79f520b1 100644 --- a/src/omniload/model.py +++ b/src/omniload/model.py @@ -29,6 +29,8 @@ class IncrementalStrategy(str, Enum): delete_insert = "delete+insert" merge = "merge" scd2 = "scd2" + upsert = "upsert" + cdc = "cdc" none = "none" diff --git a/tests/main/filesystem/test_incremental_strategy.py b/tests/main/filesystem/test_incremental_strategy.py index 841e5dfa0..961758043 100644 --- a/tests/main/filesystem/test_incremental_strategy.py +++ b/tests/main/filesystem/test_incremental_strategy.py @@ -140,7 +140,7 @@ def test_filesystem_explicit_none_appends(tmp_path, scheme): @pytest.mark.parametrize("scheme", LOCAL_SCHEMES) -@pytest.mark.parametrize("strategy", ["merge", "scd2", "delete+insert"]) +@pytest.mark.parametrize("strategy", ["merge", "scd2", "upsert", "delete+insert"]) def test_filesystem_key_strategies_are_rejected(tmp_path, strategy, scheme): """Key-dependent strategies error clearly instead of silently appending: filesystem sources can't supply the incremental/merge key these need. The message names the @@ -158,6 +158,23 @@ def test_filesystem_key_strategies_are_rejected(tmp_path, strategy, scheme): ) +@pytest.mark.parametrize("scheme", LOCAL_SCHEMES) +def test_filesystem_cdc_is_rejected_as_not_sql(tmp_path, scheme): + """cdc is refused for every source that is not a SQL database, ahead of the + key-strategy error above, because what rules it out is the source, not the key.""" + src = _write_people(tmp_path) + dest = tmp_path / "wh.duckdb" + with pytest.raises(ValidationError, match=f"and '{scheme}' is not one"): + run_ingest( + source_uri=f"{scheme}://{src}", + dest_uri=f"duckdb:///{dest}", + source_table="people", + dest_table="out.people", + incremental_strategy="cdc", + progress="log", + ) + + # --- regression guard: a non-filesystem handles_incrementality source --- diff --git a/tests/main/test_incremental_strategy_cdc.py b/tests/main/test_incremental_strategy_cdc.py new file mode 100644 index 000000000..331181cf8 --- /dev/null +++ b/tests/main/test_incremental_strategy_cdc.py @@ -0,0 +1,431 @@ +"""``--incremental-strategy cdc`` and ``upsert``: dlt merge strategies keyed on the +primary key. + +Like ``scd2``, both are a dlt merge *strategy* rather than a write disposition, so they +are unflattened onto a ``merge`` disposition that names them. ``upsert`` updates and +inserts by primary key and never deletes. ``cdc`` also deletes the destination records a +load leaves out, so it shares scd2's refusal of partial reads, and it is allowed only for +SQL database sources, the one family omniload can be sure it reads in full. + +Mock-only unit lane (no Docker, no credentials): a real ``sqlite://`` source loads into a +real embedded duckdb. Sources that are not SQL are small fakes. +""" + +import sqlite3 + +import dlt +import duckdb +import pyarrow as pa +import pyarrow.ipc as ipc +import pytest + +from omniload import ValidationError, run_ingest +from omniload.core.factory import SourceDestinationFactory + + +def _make_sqlite(path, rows, *, keyed=True): + """(Re)create a `people` table holding `rows`, standing in for a changing source.""" + conn = sqlite3.connect(str(path)) + conn.execute("DROP TABLE IF EXISTS people") + key = "PRIMARY KEY" if keyed else "" + conn.execute( + f"CREATE TABLE people (id INTEGER {key}, name TEXT, city TEXT, seen INTEGER)" + ) + conn.executemany("INSERT INTO people VALUES (?, ?, ?, ?)", rows) + conn.commit() + conn.close() + + +def _load(src, dest, strategy, **kwargs): + kwargs.setdefault("source_table", "main.people") + run_ingest( + source_uri=f"sqlite:///{src}", + dest_uri=f"duckdb:///{dest}", + dest_table="out.people", + incremental_strategy=strategy, + progress="log", + **kwargs, + ) + + +def _records(dest, table="out.people"): + con = duckdb.connect(str(dest)) + try: + return con.sql(f"select id, name, city from {table} order by id").fetchall() + finally: + con.close() + + +THREE = [ + (1, "Alice", "Auckland", 1), + (2, "Bob", "Wellington", 1), + (3, "Carol", "Dunedin", 1), +] +# Bob is gone and Carol has moved. +TWO = [(1, "Alice", "Auckland", 1), (3, "Carol", "Nelson", 2)] + + +def test_cdc_deletes_a_vanished_record(tmp_path): + """The point of cdc: after a full read the destination equals the source, so a row + the source no longer holds is deleted and a changed one is updated in place.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + + _make_sqlite(src, THREE) + _load(src, dest, "cdc") + _make_sqlite(src, TWO) + _load(src, dest, "cdc") + + assert _records(dest) == [(1, "Alice", "Auckland"), (3, "Carol", "Nelson")] + + +def test_upsert_keeps_a_vanished_record(tmp_path): + """upsert updates and inserts by primary key and deletes nothing.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + + _make_sqlite(src, THREE) + _load(src, dest, "upsert") + _make_sqlite(src, TWO) + _load(src, dest, "upsert") + + assert _records(dest) == [ + (1, "Alice", "Auckland"), + (2, "Bob", "Wellington"), + (3, "Carol", "Nelson"), + ] + + +def test_upsert_allows_an_incremental_key(tmp_path): + """upsert reads absence as nothing at all, so a windowed read is safe: the second run + reads only Carol's changed row, updates it, and leaves the rest alone.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + pipelines_dir = tmp_path / "pipelines" + + _make_sqlite(src, THREE) + _load(src, dest, "upsert", incremental_key="seen", pipelines_dir=str(pipelines_dir)) + _make_sqlite(src, TWO) + _load(src, dest, "upsert", incremental_key="seen", pipelines_dir=str(pipelines_dir)) + + assert _records(dest) == [ + (1, "Alice", "Auckland"), + (2, "Bob", "Wellington"), + (3, "Carol", "Nelson"), + ] + + +INTERVAL_QUERY = ( + "query:SELECT id, name, city FROM people " + "WHERE seen >= :interval_start AND seen <= :interval_end" +) + + +@pytest.mark.parametrize("strategy", ["scd2", "cdc"]) +@pytest.mark.parametrize( + ("option", "kwargs"), + [ + ("--incremental-key", {"incremental_key": "seen"}), + ("--sql-limit", {"sql_limit": 1}), + ("--yield-limit", {"yield_limit": 1}), + ( + "--interval-start", + {"source_table": INTERVAL_QUERY, "interval_start": "2026-01-01"}, + ), + ( + "--interval-end", + {"source_table": INTERVAL_QUERY, "interval_end": "2026-01-01"}, + ), + ], +) +def test_rejects_a_partial_read(tmp_path, strategy, option, kwargs): + """scd2 retires and cdc deletes every record a run leaves out, so anything that reads + only part of the table is refused before the read. A `query:` table binds the + interval into its placeholders with no incremental key, which windows it all the + same. One check serves both strategies, so each option is tried with each.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + _make_sqlite(src, THREE) + + with pytest.raises(ValidationError, match=f"cannot be combined with '{option}'"): + _load(src, dest, strategy, primary_key=["id"], **kwargs) + assert not dest.exists() + + +@pytest.mark.parametrize("strategy", ["scd2", "cdc"]) +def test_an_interval_on_a_plain_table_is_not_a_partial_read(tmp_path, strategy): + """Without an incremental key, an interval binds only into a `query:` table's + placeholders, so on a plain table it reads the whole table and is not refused.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + _make_sqlite(src, THREE) + + _load(src, dest, strategy, primary_key=["id"], interval_start="2026-01-01") + assert len(_records(dest)) == 3 + + +def test_cdc_after_delete_insert_keeps_its_merge_key(tmp_path): + """Current behaviour, pinned so a change shows up here: `delete+insert` marks the + incremental key as the table's merge key, dlt keeps that hint in the stored schema, + and a later cdc run then deletes only inside the merge-key values it loaded. A + partition that vanished from the source wholesale survives in the destination. The + documented remedy is a `--full-refresh` when switching a table to cdc.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + conn = sqlite3.connect(str(src)) + conn.execute("CREATE TABLE people (id INTEGER PRIMARY KEY, part INTEGER)") + conn.executemany("INSERT INTO people VALUES (?, ?)", [(1, 1), (2, 1), (3, 2)]) + conn.commit() + + def load(strategy, **kwargs): + run_ingest( + source_uri=f"sqlite:///{src}", + dest_uri=f"duckdb:///{dest}", + source_table="main.people", + dest_table="out.people", + incremental_strategy=strategy, + progress="log", + **kwargs, + ) + + def ids(): + con = duckdb.connect(str(dest)) + try: + return [ + r[0] + for r in con.sql("select id from out.people order by id").fetchall() + ] + finally: + con.close() + + load("delete+insert", incremental_key="part") + conn.execute("DELETE FROM people WHERE id IN (2, 3)") + conn.commit() + conn.close() + + load("cdc") + assert ids() == [1, 3] # 2 deleted inside partition 1; partition 2 left alone + + load("cdc", full_refresh=True) + assert ids() == [1] + + +class _DictSource: + """A source that is not SQL and keeps a cursor of its own, as several SaaS sources do, + while reporting that it leaves incrementality to omniload.""" + + def handles_incrementality(self) -> bool: + return False + + def dlt_source(self, uri, table, **kwargs): + @dlt.resource(name="rows", primary_key="id") + def rows( + updated=dlt.sources.incremental("updated", initial_value=0), # noqa: B008 + ): + yield from [{"id": 1, "updated": 1}, {"id": 2, "updated": 2}] + + return rows() + + +class _ManagedSource: + """A ``handles_incrementality`` source that sets its own resource-level disposition, + like the SaaS/streaming sources.""" + + def handles_incrementality(self) -> bool: + return True + + def dlt_source(self, uri, table, **kwargs): + @dlt.resource(name="rows") + def rows(): + yield from [{"id": 1}, {"id": 2}, {"id": 3}] + + return rows() + + +def _ingest_stub(monkeypatch, stub, dest, strategy, **kwargs): + monkeypatch.setattr(SourceDestinationFactory, "get_source", lambda self: stub) + run_ingest( + source_uri=kwargs.pop("source_uri", "mongodb://placeholder/db"), + dest_uri=f"duckdb:///{dest}", + source_table="rows", + dest_table="out.rows", + incremental_strategy=strategy, + progress="log", + **kwargs, + ) + + +def _count(dest, table="out.rows"): + con = duckdb.connect(str(dest)) + try: + return con.sql(f"select count(*) from {table}").fetchall()[0][0] + finally: + con.close() + + +@pytest.mark.parametrize("dry_run", [False, True]) +@pytest.mark.parametrize( + ("stub", "source_uri"), + [ + (_DictSource(), "mongodb://placeholder/db"), + (_ManagedSource(), "influxdb://placeholder"), + ], + ids=["own-cursor", "managed"], +) +def test_cdc_rejects_a_non_sql_source(tmp_path, monkeypatch, stub, source_uri, dry_run): + """A source that is not SQL can window its read with a cursor or a default range of + its own that omniload cannot see before reading, so cdc is refused for it, on a dry + run too. A source that owns its disposition would otherwise drop cdc to `none` + without a word, so it is refused rather than ignored.""" + dest = tmp_path / "wh.duckdb" + with pytest.raises( + ValidationError, match="only supported for SQL database sources" + ): + _ingest_stub( + monkeypatch, + stub, + dest, + "cdc", + source_uri=source_uri, + primary_key=["id"], + dry_run=dry_run, + ) + assert not dest.exists() + + +def test_cdc_rejects_an_arrow_source(tmp_path): + """`mmap://` is not a SQL source either.""" + src = tmp_path / "people.arrow" + table = pa.table({"id": [1, 2], "city": ["Auckland", "Wellington"]}) + with pa.OSFile(str(src), "wb") as sink: + with ipc.new_file(sink, table.schema) as writer: + writer.write_table(table) + + with pytest.raises(ValidationError, match="'mmap' is not one"): + run_ingest( + source_uri=f"mmap://{src}", + dest_uri=f"duckdb:///{tmp_path / 'wh.duckdb'}", + source_table="arrow_mmap", + dest_table="out.people", + incremental_strategy="cdc", + primary_key=["id"], + progress="log", + ) + + +def test_upsert_accepts_a_non_sql_source(tmp_path, monkeypatch): + """upsert deletes nothing, so a source's own cursor cannot cost it a record, and the + SQL-only rule does not apply.""" + dest = tmp_path / "wh.duckdb" + for _ in range(2): + _ingest_stub(monkeypatch, _DictSource(), dest, "upsert") + assert _count(dest) == 2 + + +def test_managed_source_still_ignores_upsert(tmp_path, monkeypatch): + """A source that owns its disposition keeps ignoring upsert, as it does scd2: its + resource-level append stands, so rows accumulate across runs.""" + dest = tmp_path / "wh.duckdb" + for _ in range(2): + _ingest_stub( + monkeypatch, + _ManagedSource(), + dest, + "upsert", + source_uri="file://placeholder.csv", + ) + assert _count(dest) == 6 + + +@pytest.mark.parametrize(("primary_key", "expected"), [(None, 6), (["id"], 3)]) +def test_influxdb_upsert_keeps_its_primary_key_merge( + tmp_path, monkeypatch, primary_key, expected +): + """InfluxDB owns its incrementality, so upsert is suppressed there like any strategy, + and its `--primary-key` override still turns the load into a plain merge.""" + dest = tmp_path / "wh.duckdb" + kwargs = {"primary_key": primary_key} if primary_key else {} + for _ in range(2): + _ingest_stub( + monkeypatch, + _ManagedSource(), + dest, + "upsert", + source_uri="influxdb://placeholder", + **kwargs, + ) + assert _count(dest) == expected + + +@pytest.mark.parametrize("backend", ["pyarrow", "default"]) +def test_cdc_accepts_an_arrow_backend(tmp_path, backend): + """cdc matches destination rows on the primary key in SQL and needs no row hash, so + unlike scd2 it works on data read as Arrow tables.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + + _make_sqlite(src, THREE) + _load(src, dest, "cdc", sql_backend=backend) + _make_sqlite(src, TWO) + _load(src, dest, "cdc", sql_backend=backend) + + assert _records(dest) == [(1, "Alice", "Auckland"), (3, "Carol", "Nelson")] + + +@pytest.mark.parametrize("strategy", ["cdc", "upsert"]) +def test_requires_a_primary_key(tmp_path, strategy): + """Both strategies match records on the primary key. dlt says so only at the load + step, as a traceback, so a table with none is refused first with a pointer to the + option that supplies one.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + _make_sqlite(src, THREE, keyed=False) + + with pytest.raises( + ValidationError, match="Name its key columns with '--primary-key'" + ): + _load(src, dest, strategy) + assert not dest.exists() + + +@pytest.mark.parametrize("strategy", ["cdc", "upsert"]) +def test_a_primary_key_option_satisfies_the_check(tmp_path, strategy): + """`--primary-key` names the key a keyless table lacks.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + _make_sqlite(src, THREE, keyed=False) + + _load(src, dest, strategy, primary_key=["id"]) + assert len(_records(dest)) == 3 + + +def test_cdc_takes_a_reflected_primary_key(tmp_path): + """A SQL table's declared key is reflected onto the resource, so cdc needs no + `--primary-key` for it (the other tests here rely on that too).""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + _make_sqlite(src, THREE) + + _load(src, dest, "cdc") + assert len(_records(dest)) == 3 + + +@pytest.mark.parametrize("backend", ["sqlalchemy", "pyarrow"]) +def test_cdc_on_an_empty_source_deletes_nothing(tmp_path, backend): + """Current dlt behaviour, pinned so a change shows up here: a table read on these + backends yields no item at all when it is empty, and dlt runs no merge for a load with + no rows (it synthesizes an empty job for `replace` only), so nothing is deleted. If + this starts failing, dlt has started treating an empty read as an empty table, which + is what cdc means, and the documented limitation can go.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + + _make_sqlite(src, THREE) + _load(src, dest, "cdc", sql_backend=backend) + _make_sqlite(src, []) + _load(src, dest, "cdc", sql_backend=backend) + + assert len(_records(dest)) == 3 + + +def test_cdc_on_an_empty_source_clears_it_with_connectorx(tmp_path): + """ConnectorX hands dlt an empty Arrow table for an empty source table, which does + reach the merge, so cdc deletes every destination record.""" + pytest.importorskip("connectorx") + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + + _make_sqlite(src, THREE) + _load(src, dest, "cdc", sql_backend="connectorx") + _make_sqlite(src, []) + _load(src, dest, "cdc", sql_backend="connectorx") + + assert _records(dest) == [] From 44cb84c0029e6ce7ea285ff86b2083dabe8f4895 Mon Sep 17 00:00:00 2001 From: TiM Date: Fri, 23 Oct 2026 07:59:58 +1300 Subject: [PATCH 2/5] fix(review): Name the merge strategy on a plain merge so a stored cdc strategy does not linger dlt keeps a table's merge strategy in its stored schema when a run names none, so after a cdc load a plain merge or delete+insert with a limited read still ran cdc and deleted every row it did not read. A bare merge disposition now names the destination's first supported strategy, which is the one dlt would default to. Also: name a query: table as the query in the primary-key error, test the deferral for dynamic hints, and say that sources managing their own incrementality ignore upsert. --- docs/changelog.md | 4 +- docs/getting-started/incremental-loading.md | 2 + src/omniload/api.py | 28 ++++++++-- tests/main/test_incremental_strategy_cdc.py | 59 +++++++++++++++++++++ 4 files changed, 88 insertions(+), 5 deletions(-) diff --git a/docs/changelog.md b/docs/changelog.md index 2e089f304..742ca6f46 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -11,7 +11,9 @@ need a primary key, declared by the table or named with `--primary-key`. dlt 1.31 is now the minimum version. `scd2` now also rejects `--interval-start` / `--interval-end` on a `query:` table that binds them, which windowed the read the - same way. See [Incremental loading]. + same way. A plain `merge` or `delete+insert` now names the destination's default + merge strategy, so a table loaded earlier with `cdc` does not keep deleting unread + rows under it. See [Incremental loading]. ## 2026/10/21 v0.20.0 diff --git a/docs/getting-started/incremental-loading.md b/docs/getting-started/incremental-loading.md index c2aa03862..e7ee9b597 100644 --- a/docs/getting-started/incremental-loading.md +++ b/docs/getting-started/incremental-loading.md @@ -310,6 +310,8 @@ Upsert inserts the source rows the destination does not have yet and updates the Upsert needs a primary key. A SQL table that declares one is used as it is; for any other table, name the key columns with `--primary-key` on every run. A run without either is rejected before anything is loaded. +Sources that manage their own incrementality, such as most SaaS and streaming sources, keep their own write dispositions and ignore `upsert`, as they ignore the other strategies. + As it deletes nothing, `upsert` can be combined with `--incremental-key` to read only the rows that changed since the previous run. The following example upserts the `my_schema.some_data` table from Postgres into DuckDB, matching rows on `id`. diff --git a/src/omniload/api.py b/src/omniload/api.py index 4dd13bb5e..d32aed62e 100644 --- a/src/omniload/api.py +++ b/src/omniload/api.py @@ -178,7 +178,9 @@ def check(res) -> None: return found[0] if found else None -def _require_primary_key(resource, strategy: IncrementalStrategy) -> None: +def _require_primary_key( + resource, strategy: IncrementalStrategy, source_table: str +) -> None: """Raise unless ``resource`` declares a primary key ``strategy`` can match rows on. A resource whose table hints depend on the data cannot be inspected before it is @@ -191,10 +193,16 @@ def _require_primary_key(resource, strategy: IncrementalStrategy) -> None: except DataItemRequiredForDynamicTableHints: return if not any(column.get("primary_key") for column in columns.values()): + # A `query:` table's resource carries a placeholder name, not one the user wrote. + table = ( + "the query" + if source_table.startswith("query:") + else f"the '{resource.name}' table" + ) raise ValidationError( f"Incremental strategy '{strategy.value}' needs a primary key to match " - f"records on, and the '{resource.name}' table declares none. Name its key " - "columns with '--primary-key'." + f"records on, and {table} declares none. Name its key columns with " + "'--primary-key'." ) @@ -937,6 +945,15 @@ def binds(placeholder: str) -> bool: if primary_key_merges: write_disposition = "merge" + # dlt stores a table's merge strategy in its schema and leaves it in place when a + # later run names none, so a table once loaded with `cdc` would go on deleting + # unread rows under a plain `merge` or `delete+insert`. Name the strategy dlt would + # default to, the destination's first, so the run loads with the one it asked for. + if write_disposition == "merge": + strategies = dlt_dest.capabilities().supported_merge_strategies + if strategies: + write_disposition = {"disposition": "merge", "strategy": strategies[0]} + supports_merge = getattr(destination, "supports_merge", lambda: True)() if not supports_merge and _is_merge(write_disposition): # Known from the request alone, so `--dry-run` reports it too. Dispositions a @@ -1074,7 +1091,10 @@ def col_h(x): # traceback. A key given with `--primary-key` reaches dlt at `pipeline.run()` # rather than on the resource, so it satisfies this outright; otherwise the # resource must carry one, as a SQL table with a declared key does by reflection. - resource.for_each(dlt_source, lambda x: _require_primary_key(x, merge_strategy)) + resource.for_each( + dlt_source, + lambda x: _require_primary_key(x, merge_strategy, source_table), + ) if isinstance(destination, AthenaDestination) and jr.partition_by: hint.apply_athena_hints(dlt_source, jr.partition_by, column_hints) diff --git a/tests/main/test_incremental_strategy_cdc.py b/tests/main/test_incremental_strategy_cdc.py index 331181cf8..70c524c3c 100644 --- a/tests/main/test_incremental_strategy_cdc.py +++ b/tests/main/test_incremental_strategy_cdc.py @@ -429,3 +429,62 @@ def test_cdc_on_an_empty_source_clears_it_with_connectorx(tmp_path): _load(src, dest, "cdc", sql_backend="connectorx") assert _records(dest) == [] + + +@pytest.mark.parametrize( + ("strategy", "kwargs"), + [ + ("merge", {"primary_key": ["id"]}), + ("delete+insert", {"incremental_key": "id"}), + ], +) +def test_a_later_merge_does_not_inherit_cdc(tmp_path, strategy, kwargs): + """dlt keeps a table's merge strategy in its stored schema when a run names none, so + after a cdc load a plain `merge` or `delete+insert` would still run cdc and delete + every row it did not read. Each run names its own strategy, so a limited read after + cdc leaves the unread rows in place.""" + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + _make_sqlite(src, THREE) + + _load(src, dest, "cdc") + _load(src, dest, strategy, sql_limit=1, **kwargs) + + assert [r[0] for r in _records(dest)] == [1, 2, 3] + + +def test_the_primary_key_error_names_a_query_as_such(tmp_path, monkeypatch): + """A `query:` table's resource has a placeholder name the user never wrote. Building + a custom-query source replaces `table_rows` in dlt for the rest of the process, so the + test restores it for the tests that run after it.""" + import dlt.sources.sql_database + + monkeypatch.setattr( + dlt.sources.sql_database, "table_rows", dlt.sources.sql_database.table_rows + ) + src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" + _make_sqlite(src, THREE, keyed=False) + + with pytest.raises(ValidationError, match="and the query declares none"): + _load(src, dest, "cdc", source_table="query:SELECT id, name, city FROM people") + + +def test_upsert_leaves_dynamic_hints_to_dlt(tmp_path, monkeypatch): + """A resource whose table name depends on each item cannot be inspected before the + read, so the primary-key check defers to dlt, which loads it with the key the + resource declares.""" + + class _DynamicSource: + def handles_incrementality(self) -> bool: + return False + + def dlt_source(self, uri, table, **kwargs): + @dlt.resource(name="rows", table_name=lambda item: "rows", primary_key="id") + def rows(): + yield from [{"id": 1}, {"id": 2}] + + return rows() + + dest = tmp_path / "wh.duckdb" + for _ in range(2): + _ingest_stub(monkeypatch, _DynamicSource(), dest, "upsert") + assert _count(dest) == 2 From c40f4211b33ebaefca232e977dd993361d51badf Mon Sep 17 00:00:00 2001 From: TiM Date: Fri, 23 Oct 2026 08:09:05 +1300 Subject: [PATCH 3/5] fix(review): Read merge strategies from the resolved destination A destination may hand dlt a factory callable, as MongoDB does, which has no capabilities of its own, so naming the default merge strategy on a plain merge raised AttributeError there, on --dry-run too. Read them from the pipeline's destination instead, which dlt has resolved. The changelog now says the named strategy replaces any strategy an earlier load stored, not only cdc. --- docs/changelog.md | 4 ++-- src/omniload/api.py | 4 +++- tests/main/test_incremental_strategy_cdc.py | 18 ++++++++++++++++++ 3 files changed, 23 insertions(+), 3 deletions(-) diff --git a/docs/changelog.md b/docs/changelog.md index 742ca6f46..084891e12 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -12,8 +12,8 @@ is now the minimum version. `scd2` now also rejects `--interval-start` / `--interval-end` on a `query:` table that binds them, which windowed the read the same way. A plain `merge` or `delete+insert` now names the destination's default - merge strategy, so a table loaded earlier with `cdc` does not keep deleting unread - rows under it. See [Incremental loading]. + merge strategy, so a strategy an earlier load stored on the table, such as `cdc`, + no longer carries over to it. See [Incremental loading]. ## 2026/10/21 v0.20.0 diff --git a/src/omniload/api.py b/src/omniload/api.py index d32aed62e..0f4a456b9 100644 --- a/src/omniload/api.py +++ b/src/omniload/api.py @@ -949,8 +949,10 @@ def binds(placeholder: str) -> bool: # later run names none, so a table once loaded with `cdc` would go on deleting # unread rows under a plain `merge` or `delete+insert`. Name the strategy dlt would # default to, the destination's first, so the run loads with the one it asked for. + # The pipeline's destination, not `dlt_dest`: a destination may hand dlt a factory + # callable (MongoDB does), which only the pipeline resolves into one with capabilities. if write_disposition == "merge": - strategies = dlt_dest.capabilities().supported_merge_strategies + strategies = pipeline.destination.capabilities().supported_merge_strategies if strategies: write_disposition = {"disposition": "merge", "strategy": strategies[0]} diff --git a/tests/main/test_incremental_strategy_cdc.py b/tests/main/test_incremental_strategy_cdc.py index 70c524c3c..316b2a539 100644 --- a/tests/main/test_incremental_strategy_cdc.py +++ b/tests/main/test_incremental_strategy_cdc.py @@ -488,3 +488,21 @@ def rows(): for _ in range(2): _ingest_stub(monkeypatch, _DynamicSource(), dest, "upsert") assert _count(dest) == 2 + + +def test_a_plain_merge_into_a_factory_destination_still_runs(tmp_path): + """Naming the default merge strategy reads the capabilities of the destination the + pipeline resolved, because a destination can hand dlt a factory callable, as MongoDB + does, which has no capabilities of its own. A dry run stops before any connection.""" + src = tmp_path / "src.db" + _make_sqlite(src, THREE) + run_ingest( + source_uri=f"sqlite:///{src}", + dest_uri="mongodb://localhost:1/db", + source_table="main.people", + dest_table="people", + incremental_strategy="merge", + primary_key=["id"], + dry_run=True, + progress="log", + ) From 8ecdc2eaac4c6b8ce9d10057bc7236a248d19296 Mon Sep 17 00:00:00 2001 From: TiM Date: Fri, 23 Oct 2026 08:37:56 +1300 Subject: [PATCH 4/5] fix(analysis): Refuse upsert and cdc on a destination with no such strategy dlt checks a named merge strategy against the destination only at the load step, and does not refuse one where the destination has no merge strategy for the table at all: it appends instead (bucket destinations), or a custom sink merges its own way (MongoDB upserts), so cdc reported success and deleted nothing. upsert and cdc are now checked against the strategies the destination supports for this table, through its selector and the table format its run parameters set, before --dry-run returns. Docs: Delta Lake supports both; the unsupported list names MongoDB and the bucket destinations; csv, deltalake and the filesystem-incremental sentence list upsert and cdc as rejected; the primary-key line says --dry-run does not check it. Tests: Turso refuses upsert and cdc, and the partial-read cases restore dlt's table_rows so a regression cannot spread to unrelated tests. --- docs/changelog.md | 5 +- docs/commands/ingest.md | 2 +- docs/getting-started/incremental-loading.md | 11 ++-- docs/supported-sources/clickhouse.md | 2 +- docs/supported-sources/csv.md | 5 +- docs/supported-sources/deltalake.md | 5 +- src/omniload/api.py | 27 +++++++++ tests/main/test_incremental_strategy_cdc.py | 66 +++++++++++++++++++-- tests/main/test_turso.py | 8 ++- 9 files changed, 112 insertions(+), 19 deletions(-) diff --git a/docs/changelog.md b/docs/changelog.md index 084891e12..76871cd63 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -13,7 +13,10 @@ `--interval-end` on a `query:` table that binds them, which windowed the read the same way. A plain `merge` or `delete+insert` now names the destination's default merge strategy, so a strategy an earlier load stored on the table, such as `cdc`, - no longer carries over to it. See [Incremental loading]. + no longer carries over to it; a table loaded with `scd2` and then with a plain + `merge` is merged with the destination's default and keeps no history for the + keys that load reads. `upsert` and `cdc` are rejected up front on a destination + that has no such strategy for the table. See [Incremental loading]. ## 2026/10/21 v0.20.0 diff --git a/docs/commands/ingest.md b/docs/commands/ingest.md index 3f1403fdf..49218fb57 100644 --- a/docs/commands/ingest.md +++ b/docs/commands/ingest.md @@ -27,7 +27,7 @@ omniload ingest \ - `--filesystem-incremental`: For filesystem-family sources, filters the metadata-only file listing by modification time and appends rows only from files not recorded by the previous run. This opt-in mode is append-only. With the default temporary pipeline directory, the destination must support dlt state sync; otherwise, reuse an explicit persistent `--pipelines-dir`. Equal-boundary files are deduplicated by file URL, while an older-mtime backfill needs `--full-refresh`. Can also be set via the `FILESYSTEM_INCREMENTAL` / `OMNILOAD_FILESYSTEM_INCREMENTAL` environment variables. Defaults to `False`. See [Filesystem sources](../getting-started/incremental-loading.md#filesystem-sources). - `--interval-start`: Sets the start of the interval for the incremental key. Defaults to `None`. - `--interval-end`: Sets the end of the interval for the incremental key. Defaults to `None`. -- `--primary-key TEXT`: Specifies the primary key for the merge operation. Defaults to `None`. +- `--primary-key TEXT`: Specifies the primary key for the merge operation. Required for `upsert` and `cdc` on a table that declares no key of its own. Defaults to `None`. - `--columns :`: Specifies the columns to be ingested. Defaults to `None`. - `--mask :[:param]`: Applies data masking to specified columns. Can be used multiple times for different columns. See the [Data Masking](../getting-started/data-masking.md) documentation for available algorithms and usage examples. Defaults to `None`. - `--reshape :`: Restructures each source document before it is loaded: flattens nested objects, coerces types, and leaves arrays as lists so they normalize into child tables. `` is `python`, `jq` or `polars`. See the [Reshape transforms](../getting-started/reshape-transforms.md) documentation for the spec formats and the type-fidelity differences between engines. Can also be set via the `RESHAPE` / `OMNILOAD_RESHAPE` environment variables. Defaults to `None`. diff --git a/docs/getting-started/incremental-loading.md b/docs/getting-started/incremental-loading.md index e7ee9b597..292cb21a3 100644 --- a/docs/getting-started/incremental-loading.md +++ b/docs/getting-started/incremental-loading.md @@ -308,7 +308,7 @@ SCD2 compares rows by a hash that dlt computes only for row-oriented data, so it Upsert inserts the source rows the destination does not have yet and updates the ones it already holds, matching them on the primary key. It deletes nothing: a row that disappears from the source stays in the destination. Where `merge` deletes the matching rows and inserts them again on most SQL destinations, `upsert` updates them in place. -Upsert needs a primary key. A SQL table that declares one is used as it is; for any other table, name the key columns with `--primary-key` on every run. A run without either is rejected before anything is loaded. +Upsert needs a primary key. A SQL table that declares one is used as it is; for any other table, name the key columns with `--primary-key` on every run. A run without either is rejected before the load; `--dry-run` does not check the key. Sources that manage their own incrementality, such as most SaaS and streaming sources, keep their own write dispositions and ignore `upsert`, as they ignore the other strategies. @@ -349,13 +349,13 @@ omniload ingest \ ``` :::{caution} -- An empty source table is handled according to the SQL backend. With `sqlalchemy` and `pyarrow`, which is the default, a run that reads no rows loads nothing and deletes nothing, so the destination keeps the rows it had. With `connectorx`, the same run deletes every row in the destination. +- An empty source table is handled according to the SQL backend. With `sqlalchemy` and `pyarrow`, which is the default, a run that reads no rows loads nothing and deletes nothing, so the destination keeps the rows it had. With `connectorx`, the same run deletes every row in the destination. A `query:` table is always read with `sqlalchemy`, and `adbcbridge` is not verified. - A destination table loaded earlier with `delete+insert` keeps its incremental key as a merge key, and a later `cdc` run then deletes only among the values of that key it has read. Add `--full-refresh` to the first `cdc` run into such a table. - Masking a primary-key column with `--mask` changes the key rows are matched on. A mask that draws a fresh value per run deletes and inserts every row again on every run, and a mask that maps two keys to one value merges those rows. ::: :::{note} -Which destinations support `upsert` and `cdc` is up to dlt. With dlt 1.31, both work on BigQuery, Databricks, DuckDB 1.4 or later, MotherDuck, Microsoft SQL Server, Postgres, Snowflake, and Athena with Iceberg tables. ClickHouse, CrateDB, Redshift, Synapse, Turso, and the destinations written through SQLAlchemy, such as MySQL and SQLite, support neither. +Which destinations support `upsert` and `cdc` is up to dlt. With dlt 1.31, both work on BigQuery, Databricks, DuckDB 1.4 or later, MotherDuck, Microsoft SQL Server, Postgres, Snowflake, Athena with Iceberg tables, and Delta Lake (`file+delta://`, `s3+delta://` and the other `+delta` destinations). ClickHouse, CrateDB, Redshift, Synapse, Turso, MongoDB, the bucket destinations (`s3://`, `gs://`, `az://` and the like), and the destinations written through SQLAlchemy, such as MySQL and SQLite, support neither. omniload rejects either strategy before loading, `--dry-run` included, on a destination that has no such strategy for the table. CrateDB is the exception: it reports Postgres' strategies and fails when the load runs. ::: (incremental-loading-filesystem)= @@ -396,8 +396,9 @@ omniload ingest \ The mode is opt-in, so omitting the flag keeps the append-on-re-run behaviour described above. It is append-only: omit `--incremental-strategy` or set it to -`append`. Combining it with `replace`, `delete+insert`, `merge`, or `scd2` is -rejected before extraction. Python callers set `filesystem_incremental=True` on +`append`. Combining it with `replace`, `delete+insert`, `merge`, `scd2`, or +`upsert` is rejected before extraction, and `cdc` is rejected for every +filesystem source. Python callers set `filesystem_incremental=True` on `run_ingest()`. The cursor is stored in dlt pipeline state. With omniload's default temporary diff --git a/docs/supported-sources/clickhouse.md b/docs/supported-sources/clickhouse.md index 392d1fc15..a58f19da3 100644 --- a/docs/supported-sources/clickhouse.md +++ b/docs/supported-sources/clickhouse.md @@ -56,5 +56,5 @@ omniload ingest \ --> :::{warning} -Clickhouse currently doesn't support `cdc`, `delete+insert`, `merge`, `scd2` or `upsert` incremental strategies. +Clickhouse as a destination currently doesn't support `cdc`, `delete+insert`, `merge`, `scd2` or `upsert` incremental strategies. ::: diff --git a/docs/supported-sources/csv.md b/docs/supported-sources/csv.md index de19af1d3..b7a373fad 100644 --- a/docs/supported-sources/csv.md +++ b/docs/supported-sources/csv.md @@ -82,8 +82,9 @@ seven things: not implement append rejects the load outright rather than duplicating rows; the MongoDB destination is one, so a `csv://` to MongoDB command now needs the explicit `--incremental-strategy replace`. -- **`merge`, `delete+insert` and `scd2` are rejected.** They need an - incremental or merge key, which the shared reader does not expose. Use a +- **`merge`, `delete+insert`, `scd2` and `upsert` are rejected.** They need an + incremental or merge key, which the shared reader does not expose. `cdc` is + rejected too, as for every source that is not a SQL database. Use a source that does, or `--full-refresh` to reset the destination. - **`--incremental-key` is rejected**, with the same error `file://` gives. The row cursor it drove compared raw strings against parsed datetimes and crashed diff --git a/docs/supported-sources/deltalake.md b/docs/supported-sources/deltalake.md index 159b93ebc..738a43560 100644 --- a/docs/supported-sources/deltalake.md +++ b/docs/supported-sources/deltalake.md @@ -211,8 +211,9 @@ batch size. table in the URI, so `--source-table` only names the dlt resource. - The `lakefs+delta://` and `hdfs+delta://` destinations take no storage options from the URI query, because dlt models no credentials for those protocols. -- The merge, delete-and-insert, and SCD2 strategies need an incremental or merge - key, which a whole-table read does not expose. +- The merge, delete-and-insert, SCD2, and upsert strategies need an incremental or + merge key, which a whole-table read does not expose. As a source, Delta Lake also + rejects `cdc`, which is for SQL database sources only. [Delta Lake]: https://delta.io/ [`polars.scan_delta`]: https://docs.pola.rs/api/python/stable/reference/api/polars.scan_delta.html diff --git a/src/omniload/api.py b/src/omniload/api.py index 0f4a456b9..19d3e9d9a 100644 --- a/src/omniload/api.py +++ b/src/omniload/api.py @@ -975,6 +975,33 @@ def binds(placeholder: str) -> bool: f"{' and '.join(fixes)}." ) + # dlt checks a named merge strategy against the destination only at the load step, + # and a destination with no merge strategy for the table is not refused at all: + # dlt appends instead, or a custom sink merges its own way, so `cdc` would delete + # nothing. Check upsert and cdc against what this table supports, up front. + if merge_strategy in (IncrementalStrategy.upsert, IncrementalStrategy.cdc): + capabilities = pipeline.destination.capabilities() + strategies = list(capabilities.supported_merge_strategies or []) + if strategies and capabilities.merge_strategies_selector: + table_format = destination.dlt_run_params( + uri=jr.dest_uri, table=dest_table, staging_bucket=jr.staging_bucket + ).get("table_format") + strategies = list( + capabilities.merge_strategies_selector( + strategies, table_schema={"table_format": table_format} + ) + ) + if merge_strategy.value not in strategies: + raise ValidationError( + f"The '{factory.destination_scheme}' destination does not support " + f"incremental strategy '{merge_strategy.value}' for this table. " + + ( + f"Its merge strategies are: {', '.join(strategies)}." + if strategies + else "It has no merge strategy for it." + ) + ) + if jr.dry_run: logger.info("Skipping data transfer, because `--dry-run` was selected.") return None diff --git a/tests/main/test_incremental_strategy_cdc.py b/tests/main/test_incremental_strategy_cdc.py index 316b2a539..a34201fc1 100644 --- a/tests/main/test_incremental_strategy_cdc.py +++ b/tests/main/test_incremental_strategy_cdc.py @@ -135,11 +135,20 @@ def test_upsert_allows_an_incremental_key(tmp_path): ), ], ) -def test_rejects_a_partial_read(tmp_path, strategy, option, kwargs): +def test_rejects_a_partial_read(tmp_path, monkeypatch, strategy, option, kwargs): """scd2 retires and cdc deletes every record a run leaves out, so anything that reads only part of the table is refused before the read. A `query:` table binds the interval into its placeholders with no incremental key, which windows it all the - same. One check serves both strategies, so each option is tried with each.""" + same. One check serves both strategies, so each option is tried with each. + + Were the refusal to regress, a `query:` case would build a custom-query source, + which replaces `table_rows` in dlt for the rest of the process, so it is restored + to keep such a failure from spreading to unrelated tests.""" + import dlt.sources.sql_database + + monkeypatch.setattr( + dlt.sources.sql_database, "table_rows", dlt.sources.sql_database.table_rows + ) src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" _make_sqlite(src, THREE) @@ -440,9 +449,10 @@ def test_cdc_on_an_empty_source_clears_it_with_connectorx(tmp_path): ) def test_a_later_merge_does_not_inherit_cdc(tmp_path, strategy, kwargs): """dlt keeps a table's merge strategy in its stored schema when a run names none, so - after a cdc load a plain `merge` or `delete+insert` would still run cdc and delete - every row it did not read. Each run names its own strategy, so a limited read after - cdc leaves the unread rows in place.""" + after a cdc load a plain `merge` would still run cdc and delete every row it did not + read. Each run names its own strategy, so a limited read after cdc leaves the unread + rows in place. The `delete+insert` case is a regression assertion only: its own merge + key already scopes the stored cdc delete to the keys it loaded.""" src, dest = tmp_path / "src.db", tmp_path / "wh.duckdb" _make_sqlite(src, THREE) @@ -506,3 +516,49 @@ def test_a_plain_merge_into_a_factory_destination_still_runs(tmp_path): dry_run=True, progress="log", ) + + +@pytest.mark.parametrize("strategy", ["cdc", "upsert"]) +@pytest.mark.parametrize( + ("dest_uri", "message"), + [ + ("sqlite:///{tmp}/app.db", "Its merge strategies are: delete-insert, scd2"), + ("mongodb://localhost:1/db", "It has no merge strategy for it"), + ], + ids=["sqlalchemy", "mongodb"], +) +def test_rejects_a_destination_that_cannot_run_the_strategy( + tmp_path, strategy, dest_uri, message +): + """dlt refuses a strategy a SQLAlchemy destination lacks only at the load step, and + one with no merge strategy at all is not refused: MongoDB upserts, a bucket appends, + and `cdc` deletes nothing. Both are refused up front, on a dry run too.""" + src = tmp_path / "src.db" + _make_sqlite(src, THREE) + with pytest.raises(ValidationError, match=message): + run_ingest( + source_uri=f"sqlite:///{src}", + dest_uri=dest_uri.format(tmp=tmp_path), + source_table="main.people", + dest_table="people", + incremental_strategy=strategy, + dry_run=True, + progress="log", + ) + + +@pytest.mark.parametrize("strategy", ["cdc", "upsert"]) +def test_a_delta_table_runs_both_strategies(tmp_path, strategy): + """Delta Lake's table format carries every dlt merge strategy, so the check reads the + destination's selector for the table rather than the bare capability list.""" + src = tmp_path / "src.db" + _make_sqlite(src, THREE) + run_ingest( + source_uri=f"sqlite:///{src}", + dest_uri=f"file+delta://{tmp_path / 'lake'}", + source_table="main.people", + dest_table="out.people", + incremental_strategy=strategy, + dry_run=True, + progress="log", + ) diff --git a/tests/main/test_turso.py b/tests/main/test_turso.py index 722105931..17efc190b 100644 --- a/tests/main/test_turso.py +++ b/tests/main/test_turso.py @@ -152,8 +152,10 @@ def test_scd2_guard_covers_a_turso_source(tmp_path): {"incremental_strategy": "merge", "primary_key": ["id"]}, {"incremental_strategy": "delete+insert", "incremental_key": "id"}, {"incremental_strategy": "scd2"}, + {"incremental_strategy": "upsert", "primary_key": ["id"]}, + {"incremental_strategy": "cdc"}, ], - ids=["merge", "delete+insert", "scd2"], + ids=["merge", "delete+insert", "scd2", "upsert", "cdc"], ) def test_requested_merge_strategies_are_refused_before_loading(tmp_path, kwargs): src, dest = tmp_path / "src.db", tmp_path / "app.db" @@ -169,8 +171,10 @@ def test_requested_merge_strategies_are_refused_before_loading(tmp_path, kwargs) {"incremental_strategy": "merge", "primary_key": ["id"]}, {"incremental_strategy": "delete+insert", "incremental_key": "id"}, {"incremental_strategy": "scd2"}, + {"incremental_strategy": "upsert", "primary_key": ["id"]}, + {"incremental_strategy": "cdc"}, ], - ids=["merge", "delete+insert", "scd2"], + ids=["merge", "delete+insert", "scd2", "upsert", "cdc"], ) def test_a_dry_run_reports_a_requested_merge(tmp_path, kwargs): src, dest = tmp_path / "src.db", tmp_path / "app.db" From b606171a67dfa1bb5e5e7dc0a63484591a669c9c Mon Sep 17 00:00:00 2001 From: TiM Date: Fri, 23 Oct 2026 08:39:13 +1300 Subject: [PATCH 5/5] test(analysis): Pin the bucket destination's refusal of upsert and cdc A bucket destination lists merge strategies but has none for a table without a table format, which only the destination's selector reveals. --- tests/main/test_incremental_strategy_cdc.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/main/test_incremental_strategy_cdc.py b/tests/main/test_incremental_strategy_cdc.py index a34201fc1..f52433223 100644 --- a/tests/main/test_incremental_strategy_cdc.py +++ b/tests/main/test_incremental_strategy_cdc.py @@ -524,8 +524,12 @@ def test_a_plain_merge_into_a_factory_destination_still_runs(tmp_path): [ ("sqlite:///{tmp}/app.db", "Its merge strategies are: delete-insert, scd2"), ("mongodb://localhost:1/db", "It has no merge strategy for it"), + ( + "s3://bucket/prefix?access_key_id=x&secret_access_key=y", + "It has no merge strategy for it", + ), ], - ids=["sqlalchemy", "mongodb"], + ids=["sqlalchemy", "mongodb", "bucket"], ) def test_rejects_a_destination_that_cannot_run_the_strategy( tmp_path, strategy, dest_uri, message