Repository navigation
Incremental: Add the upsert and cdc strategies - #422
Merged
Merged
Conversation
`--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.
… 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.
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.
…rategy 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.
A bucket destination lists merge strategies but has none for a table without a table format, which only the destination's selector reveals.
Contributor
|
Important Review skippedAuto reviews are limited based on label configuration. 🏷️ Required labels (at least one): 1
Please check the settings in the CodeRabbit UI or the ⚙️ Run configuration
You can disable this status message by setting the Use the checkbox below for a quick retry:
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #422 +/- ##
==========================================
+ Coverage 60.11% 60.24% +0.13%
==========================================
Files 201 201
Lines 10311 10344 +33
==========================================
+ Hits 6198 6232 +34
+ Misses 4113 4112 -1 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
5 tasks done
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Two new
--incremental-strategyvalues for #396, both dlt merge strategies keyed on the primary key.upsertinserts new rows and updates changed ones in place, and deletes nothing.cdckeeps the destination equal to the source: it also deletes the rows the source no longer holds. Until now onlyreplacecould drop a vanished row, by rewriting the whole table.mergewrite disposition that names the strategy, the pathscd2already took. Theruns_scd2special case becomes one named-merge-strategy path shared by all three.cdcdeletes whatever a run does not read, so it shares scd2's refusal of partial reads:--incremental-key,--sql-limit,--yield-limit, and--interval-start/--interval-endon aquery:table that binds them into its placeholders.cdcis accepted for SQL database sources only (SQL_SOURCE_SCHEMES). Other sources can keep a cursor or a default window of their own (Google Ads' rollingsegments_date, Kafka offsets, the saved-query windows in Allium and Dune), and on dlt 1.31 that is not visible before extraction:DltResource.incrementalis a wrapper on every resource whose signature takes an incremental, and the innerIncrementalonly exists once the generator runs. The refusal sits ahead of thehandles_incrementality()suppression, so a source that owns its disposition gets an error rather than a silently droppedcdc, on--dry-runtoo.--dry-runreturns. dlt refuses a missing strategy only at the load step, and where a destination has no merge strategy for the table it refuses nothing: bucket destinations fall back to append and MongoDB's sink upserts, socdcwould report success having deleted nothing.>=1.31, the first release with thecdcstrategy.Docs: updated
Behaviour changes
mergeordelete+insertnow names its merge strategy. dlt keeps a table's merge strategy in the stored schema when a run names none, so after acdcload a plainmergewith--sql-limitstill ran cdc and deleted the rows it did not read. The bare disposition now names the destination's first supported strategy, which is the one dlt picks by default. That replaces any stored strategy,scd2included: a table loaded withscd2and then with a plainmergekeeps no history for the keys that load reads. Destinations that declare no merge strategies keep the baremerge.--interval-start/--interval-endon aquery:table that uses the placeholders now fail the partial-read check. On a plain table they only take effect through--incremental-key, so they are not refused there.Limits
sqlalchemyandpyarrow(the default) a read that yields no rows produces no load job, so nothing is deleted. Withconnectorxan empty Arrow table reaches the merge and the destination is emptied, which matches the source. Both are pinned by tests (the connectorx one skips where connectorx is not installed, which includes CI).adbcbridgeis not verified.delete+insertkeeps its incremental key as a stored merge key, so a latercdcrun deletes only inside the key values it loaded. Documented with--full-refreshas the remedy and pinned by a test.--dry-runreturns, because it needs the built resource. A keyless dry run passes; the real run is refused. A table with no declared key needs--primary-keyon every run, even where dlt could take it from the stored schema.merge; the destination check cannot see that. The CrateDB page lists both strategies as unsupported.mergeitself is outside the strategy naming: a table it shares with an earliercdcload would still inherit cdc.Out of scope
source_filter,destination_scope),skip_unchanged_rows, and a--merge-keyfor partition-scoped cdc deletes.scd2, and the destination check forscd2. Both rules apply tocdc(and the destination check toupsert) only, so existingscd2runs are unchanged.Review
merge(a storedcdcstrategy deleted unread rows), reading capabilities from the resolved destination (MongoDB's factory destination raised), thequery:wording of the primary-key error.docs/getting-started/incremental-loading.md:297: the SCD2 section could repeat the changelog's note that a later plainmergedrops history for the keys it reads (declined: stated in the changelog).src/omniload/api.py:999: the destination refusal lists dlt's strategy names (delete-insert) rather than--incremental-strategyvalues (mention).src/omniload/api.py:982: the destination check does not coverscd2, which still appends into a bucket or upserts into MongoDB (mention; pre-existing).docs/supported-sources/csv.md:85: upsert's rejection reason repeats the error text, which names an incremental or merge key though upsert keys on the primary key (mention).tests/main/test_incremental_strategy_cdc.py:432: the connectorx empty-source pin needs connectorx installed to run (track).Changes
src/omniload/model.py:upsertandcdcmembers.src/omniload/api.py: the named-merge-strategy path, the SQL-only cdc refusal, the shared partial-read check with the interval options, the early primary-key check, the destination strategy check, and naming the default strategy on a plainmerge.pyproject.toml:dlt[deltalake]>=1.31,<1.32.tests/main/test_incremental_strategy_cdc.py(new),tests/main/filesystem/test_incremental_strategy.py,tests/main/test_turso.py.docs/getting-started/incremental-loading.md: Upsert and CDC sections;docs/commands/ingest.md, the ClickHouse, CrateDB, CSV, Delta Lake and Turso pages, the changelog, andupsertin the Vale vocabulary.Test plan
tests/main/test_incremental_strategy_cdc.py: 47 passed, 1 skipped (connectorx absent). Delete and update round trips for both strategies; the partial-read refusal forscd2andcdcover every option; the non-SQL refusal (own-cursor stub, managed source,mmap://, with and without--dry-run); primary-key refusal, reflection, thequery:wording and the dynamic-hints deferral; both Arrow backends; the empty-source and merge-key pins; the InfluxDB--primary-keyoverride withupsert; a limitedmerge/delete+insertaftercdckeeping the unread rows; amergedry run into MongoDB's factory destination; the destination refusal (SQLAlchemy, MongoDB, a bucket) and Delta Lake accepting both.tests/main/test_turso.py:upsertandcdcjoin the merge refusals, on a load and on a dry run.tests/main/filesystem/test_incremental_strategy.py:upsertjoins the key-strategy rejection, andcdcgets the non-SQL refusal.merge-after-cdccase (thedelete+insertcase passes either way, since its own merge key scopes the stored cdc delete, and stays as a regression assertion); reading capabilities from the unresolved destination fails the MongoDB test; dropping the destination check fails the SQLAlchemy, MongoDB and bucket cases; skipping the selector fails the bucket cases; dropping the table format fails the Delta cases.scd2test passes unchanged.poe lintclean;sphinx-build -Wclean; Docker-free unit lane 1,390 passed, 45 skipped.tests/warehouse/db/test_mongodb.py::test_mongodb_source[postgres], the MongoDB container closed the connection under parallel load); that file passes 20/20 on its own.b606171a.