iceberg: add a copy-on-write merge strategy to the output - #4666
Draft
Jeffail wants to merge 12 commits into
Draft
iceberg: add a copy-on-write merge strategy to the output#4666Jeffail wants to merge 12 commits into
Jeffail wants to merge 12 commits into
Conversation
The row-level upsert/delete path writes merge-on-read equality-delete files, which several query engines can't read — notably Snowflake and the Databricks Unity Catalog, and UC also rejects tables carrying identifier-field-ids. This adds an opt-in `merge_strategy: copy-on-write` that rewrites whole data files so the table only ever holds plain data files any engine can read, and skips registering identifier-field-ids on auto-created tables. The default stays merge-on-read, so existing configs are unaffected. copy-on-write trades write amplification for read compatibility (each mutating batch rewrites every data file holding a touched key), so it's a batch/moderate-throughput mode. The prototype supports flat-primitive schemas with int/long/string/boolean merge keys on unpartitioned tables; other cases fail loudly or stay on merge-on-read. New columns in a mutating batch trigger schema evolution (parity with the append path), and files left behind by a failed copy-on-write commit are cleaned up.
Extends the opt-in copy-on-write merge strategy from the initial flat-primitive/unpartitioned prototype to the breadth merge-on-read already supports, and tightens a couple of edges: - Column types: adds binary/fixed and nested struct/list/map (whose leaves must be supported primitives), via a recursive value massager that applies the same canonicalisation at every depth. This fixes silent truncation of integers nested beyond 2^53 and reshapes maps to the Arrow map JSON encoding. - Merge-key types: adds date/time/timestamp/timestamptz/uuid, each with a round-trip test that proves the filter matched the intended row (not just "no error"). Decimal merge keys are rejected up front with a clear message — an upstream substrait conversion panics on a decimal filter literal — pointing at merge-on-read instead. - Partitioned tables: supported. Copy-on-write rewrites by filter and appends value-routed rows, so it needs no partition-vs-identifier constraint and can even move a key across partitions. - Format version: copy-on-write no longer forces the irreversible v1->v2 upgrade, since it only ever writes plain data files. Also adds scale and memory benchmarks and a docs pass: a merge-on-read vs copy-on-write decision guide, a worked Snowflake/Databricks example, a maintenance runbook, and the current limitations.
Retry idempotency: an ambiguous catalog response (ErrCommitStateUnknown, e.g. a 5xx/timeout after the write) previously failed the batch on the mutation paths, since blindly retrying a possibly-landed commit could duplicate it. Each mutation commit now stamps a per-call commit-id token into its snapshot summary; on a retry after a reload, if a snapshot already carries the token the prior attempt landed, so the commit returns success instead of re-applying. This makes both copy-on-write (Overwrite/Delete) and merge-on-read (RowDelta) safe to retry on an unknown state. The append path is unchanged — it is already idempotent via file-path dedupe. Also: - multi-writer concurrency test: two committers doing copy-on-write on one table converge via the optimistic-concurrency conflict + retry path with no lost updates (race-clean). - partition-transform integration coverage: truncate plus day/month temporal transforms round-trip under copy-on-write (identity and bucket were already covered), leaving zero delete files. - a one-time startup log when copy-on-write is configured with a mutating operation, noting the write-amplification trade-off and the sort-by-key / large-batch mitigations.
…, coverage Acts on a full review of the copy-on-write merge strategy. Two behavioural fixes and a large coverage expansion; no change to the merge-on-read or append defaults. Correctness: - The copy-on-write rewrite rejected numeric-epoch temporal values in data columns, diverging from the insert path and ignoring schema_metadata (so a CDC stream carrying epoch-number timestamps would create the table, ingest the first insert, then fail every upsert). The rewrite now runs numeric temporals through the same schema-metadata/unit-aware conversion the shredder uses, via a shared helper with an equivalence test that keeps the two paths from drifting. Temporal merge keys stay strict (a bare number is still rejected up front) so the filter literal can never disagree with the stored value. - No-timezone `timestamp` columns were written to parquet as UTC-adjusted (isAdjustedToUTC=true), so they read back as `timestamptz` and a copy-on-write file rewrite failed with "cannot promote timestamptz to timestamp". No-tz timestamps are now written with isAdjustedToUTC=false per the Iceberg spec; `timestamptz` is unchanged. (Tables whose data files were written before this fix keep the old annotation and would need their data rewritten — noted in the tests.) - Orphaned parquet files written by a copy-on-write attempt that hit a clean conflict and then succeeded on retry are now cleaned up (previously only the final-failure path cleaned up); cleanup is still skipped on an ambiguous (possibly-landed) commit. max_retries gained a lower bound. Boolean merge keys join decimal as gated (an upstream filter-rewrite limitation); boolean/decimal remain fine as ordinary columns. Coverage: numeric/production merge-key input shapes, case-insensitive matching, malformed nested input, the schema-evolution evolve-and-retry loop, the terminal-unknown cleanup-skip and lost-ack retry paths, runtime-warning wiring, and a set of copy-on-write integration tests that read back through DuckDB (an independent engine) — merge-key types, nested struct/list/map, schema evolution, v1 tables, delete-only and cross-partition key moves, and multi-file rewrites.
…ange The earlier spec fix (no-timezone `timestamp` columns written with isAdjustedToUTC=false) would silently change the on-disk encoding of existing tables, leaving them with mixed parquet annotations after an upgrade. This makes the choice per-table, permanent, and visible via the table property `redpanda-connect.timestamp-encoding`: - Tables created by this output are pinned to `spec` at creation. - An existing table without the property is bootstrapped on first write by probing a data file's parquet footer: files carrying the old UTC-adjusted annotation pin the table to `legacy`, so its encoding never changes and never mixes; an empty table, a schema without no-tz timestamp columns, or spec-encoded files pin `spec`. The resolved value is stamped onto the table so every future writer agrees; probe or stamp failures are hard errors rather than guesses. - `legacy` writes byte-identical output to pre-fix releases; `timestamptz` columns are unaffected in both modes. Mutating copy-on-write against a legacy-pinned table with no-tz timestamp columns now fails upfront with a migration message (compact the data files, then set the property to `spec`) instead of surfacing iceberg-go's "cannot promote timestamptz to timestamp" mid-commit. Insert-only copy-on-write and merge-on-read keep writing the table's own encoding and are unaffected.
A scripted setup/test/teardown harness for validating the iceberg output against a real Databricks Unity Catalog, following the existing e2e pattern (terraform + Taskfile + flag-driven tests driving the real router). Terraform provisions an isolated random-suffixed catalog, a pre-created schema, a 2X-Small serverless SQL warehouse with one-minute auto-stop, and the grants an external Iceberg REST client needs (including EXTERNAL USE SCHEMA); teardown is a terraform destroy. The tests cover the questions that only a real engine-backed catalog can answer: a copy-on-write insert/upsert/delete round-trip read back through Databricks' own SQL engine (via the statement-execution API, asserting exact final state, timestamp types, and zero delete files); a reproduction of Unity Catalog rejecting table creation that registers identifier-field-ids (which fails loudly in reverse if that behaviour ever changes); a non-fatal diagnostic recording how Unity Catalog actually handles a merge-on-read equality-delete commit; and a flag-gated commit-latency benchmark. The auth token travels via environment only — never terraform state, outputs, or flags. Everything skips cleanly when unconfigured, and the attributes that cannot be verified without live workspace access are marked in comments and the README.
Databricks workspaces on default (Databricks-managed) storage cannot vend credentials to external Iceberg REST clients, so the harness's tables must live in a catalog backed by customer-owned S3 — which is also the fix when a metastore has no storage root. A new opt-in `create_storage` variable provisions the full chain: S3 bucket, an IAM role built from the provider's Unity Catalog assume-role-policy data sources (the documented way around the credential/role trust-policy circular dependency), a storage credential, an external location, and the catalog's storage root resolving to it (an explicit storage_root variable still wins; otherwise the metastore root is inherited). The AWS provider is only truly configured when storage creation is enabled, so the default path still needs no AWS setup at all. The README gains a trial-account quick-start covering the express-trial signup, the metastore external-access toggle, and the default-storage limitation that makes customer-owned storage necessary; the provider lock file is now tracked, matching the sibling harnesses.
Unity Catalog rejects commits whose property updates carry keys it manages itself — iceberg-go's copy-on-write path defensively sets schema.name-mapping.default on tables without a name mapping, which UC prohibits external clients from writing, so every mutating copy-on-write commit against a UC table failed with "Table properties contain prohibited keys". The committer now learns prohibited keys from the catalog's own rejection (parsing the keys named in the error), warns once per key, and retries with those keys stripped from set-properties updates — via a CatalogIO wrapper that filters only property updates and passes every other update through untouched. Stripping the name mapping is safe: it exists as a read fallback for data files without parquet field IDs, which the Iceberg spec requires and this output always writes. The connector's own redpanda-connect.* properties are never stripped — a catalog prohibiting those fails loudly instead, since silently dropping the timestamp-encoding pin would change write semantics. NewCommitter now takes the table's catalog explicitly (the table type exposes no accessor for it), provided by catalogx so reloaded tables are rebound onto the wrapper too. Verified live against a Databricks Unity Catalog trial workspace: the first mutating commit is rejected, the key is learned and stripped, and the retried commit lands; the copy-on-write round-trip then reads back correctly through Databricks SQL with zero delete files.
Matches the short-description convention the other row-operation fields gained upstream.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
Comment on lines
+388
to
+393
| case iceberg.Int32Type: | ||
| n, err := cowValueToInt64(v) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("%s %q: %w", ioFieldIdentifierFields, name, err) | ||
| } | ||
| return iceberg.NewLiteral(int32(n)), nil |
There was a problem hiding this comment.
Silent int32 wrap can build a filter that deletes the wrong row.
cowValueToInt64 only guards against float64 precision loss (>= 1<<53); it never bounds the result to int32. For an int (Iceberg Int32Type) merge key, int32(n) then wraps silently: a message with {"id": 4294967297} (decoded as json.Number, so the float64 guard never fires) yields int32(4294967297) == 1, and buildCOWFilter produces id IN (1).
On a delete-only copy-on-write batch there is no NewReader, so nothing else validates the value — commitOverwrite runs txn.Delete with that filter and removes the row with id = 1. That is silent data loss on an unrelated row, and it's exactly the "filter must agree with the stored value" invariant the rest of this function is written to protect (the doc comment calls it the CON-490 hazard, and the sibling decimal/boolean cases fail loudly rather than risk it).
Suggested fix: range-check before narrowing — reject values outside [math.MinInt32, math.MaxInt32] for an Int32Type key with an actionable error naming identifier_fields and the offending value, in the same style as the other rejections here.
Ref: cow.go#L386-L399, and the no-silent-corruption bar in CONTRIBUTING.md §3.1.4 / §3.2.2.
| // | ||
| // Cleanup is best-effort: skipped when the filesystem can't be listed | ||
| // (before == nil). | ||
| if before != nil && !errors.Is(err, rest.ErrCommitStateUnknown) && (err != nil || retried) { |
There was a problem hiding this comment.
Diff-based orphan cleanup can delete a concurrent committer's in-flight files.
The safety argument in the comment above holds only for committed files. referencedDataFilePaths reads c.table's current snapshot, so it protects files that some snapshot already points at — but a second committer writing copy-on-write to the same table has, in the window between txn.Overwrite writing its parquet and its catalog commit landing, files that are (a) newer than our before snapshot and (b) referenced by no snapshot at all. Both cleanup triggers on this line reach them:
retriedsuccess: committer A conflicts, reloads, succeeds on attempt 2, then walks the data dir. Committer B is mid-Overwritefor its next batch → B's freshly written files are diffed out and removed → B's commit lands referencing deleted files.err != nil: same window, and this path cleans unconditionally.
The comment explicitly reasons that "by the time it succeeds c.table has been reloaded onto the latest committed state, so referencedDataFilePaths protects every live file" — that's true of committed files but not of another writer's uncommitted ones, which is the case the first-attempt guard was added to avoid. TestCOWConcurrentCommittersConverge doesn't cover it: there, the winner has fully committed before the loser's cleanup runs.
Note this is reachable with a single pipeline too (two Connect instances, or two iceberg outputs, against one table); max_in_flight: 1 only serialises within one output.
Suggested direction: scope cleanup to paths this committer actually authored rather than a directory diff (e.g. have the overwrite stage record the files it wrote, or restrict the diff to a per-commit path prefix), so cleanup can never touch a file another writer produced.
Ref: committer.go#L336-L352, cleanupOrphanedOverwriteFiles, CONTRIBUTING.md §3.1.4.
This change roughly triples the package's integration test count, so the runner's default five-minute go test timeout no longer fits — the run was killed mid-test in CI. Grant the package fifteen minutes, in line with the other larger suites. Also rewords a few code comments to describe their hazards directly rather than by internal reference.
Review follow-ups on the newest machinery: - The prohibited-keys error parse now requires the colon-delimited list form, stops at the first token that isn't shaped like a property key, and only learns keys the failed commit actually sent — so prose or unrelated errors mentioning the phrase can no longer poison the strip set. The reserved-prefix refusal is case-insensitive. - Learned prohibited keys now live on the router's table entry and seed each new committer, so writer recreation (schema evolution, transient errors) no longer costs a rejected commit per generation to re-learn them — which also removes a latent no-progress loop against catalogs that name one key per rejection. - An ambiguous commit response is handled by the idempotency check even when its text mentions prohibited keys, so a landed-but-unreported commit can never be retried without the duplicate guard. - NewCommitter rejects a nil reload function instead of panicking later. - The copy-on-write legacy-encoding guard and docs now tell operators to stop connector writers around an encoding migration, and comments on the stamp path describe the REST-catalog race semantics honestly. - Adds the previously-unexercised probe test for a table whose timestamptz column sorts ahead of its timestamp column.
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
1 participant
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.
The iceberg output's row-level
upsert/deletesupport writes merge-on-read equality-delete files, which several engine-backed catalogs can't read — Snowflake's engine ignores equality deletes entirely, and the Databricks Unity Catalog rejects table creation carrying identifier-field-ids and can't read v2 delete files at all. So the current mutation support only lands correctly on catalog-only targets (Polaris and friends), which came up in the field pretty quickly.This adds an opt-in
merge_strategy: copy-on-writethat materialises mutations by rewriting whole data files instead — the table only ever contains plain data files, which every engine can read. Under copy-on-write theidentifier_fieldsact as a connector-side merge key only and aren't registered as Iceberg identifier-field-ids, which is what lets Unity Catalog accept the auto-created tables. The default staysmerge-on-read; existing configs are untouched.Support matrix: all flat primitive column types plus nested struct/list/map; merge keys of int/long/string/date/time/timestamp/timestamptz/uuid (decimal and boolean keys are rejected with actionable errors — upstream filter-rewrite limitations); partitioned tables including transforms and cross-partition key moves; works on v1 or v2 tables without forcing the irreversible v1→v2 upgrade. The trade-off is write amplification (each mutating batch rewrites every file containing a touched key), so it's documented as a batch/moderate-throughput mode with a decision guide vs merge-on-read.
Behaviour changes reviewers should look at first:
timestampcolumns were written withisAdjustedToUTC=true, which is spec-incorrect and made them read back astimestamptz. New files are spec-correct — but to guarantee an existing table never changes or mixes encodings, the choice is pinned per table via aredpanda-connect.timestamp-encodingproperty (stamped at creation for new tables, footer-probed once and pinned for pre-existing ones). Copy-on-write against a legacy-pinned table fails upfront with a migration message.redpanda-connect.commit-idsummary property.schema.name-mapping.default, which the copy-on-write library path injects) are learned from the catalog's rejection and stripped from commits, with the connector's own properties exempt.Validation: full unit coverage (round-trip fidelity tests per column/key type, failure-path and idempotency tests incl. landed-but-unreported commits, race-checked concurrency); 13 integration tests against a REST catalog with DuckDB as an independent reader (zero-delete-files asserted throughout); and a live Databricks Unity Catalog run via the new
e2e/databricksharness — the copy-on-write round-trip reads back exactly through Databricks SQL (timestamp_ntz/timestamptyped correctly), the identifier-field rejection reproduces verbatim, and equality-delete commits are confirmed rejected loudly. Commit-latency benchmarks against the live catalog are included in the docs guidance.Draft while I collect review; the e2e harness needs live credentials so it skips by default. Happy to split anything out if the reviewers would rather take this in pieces.
🤖 Generated with Claude Code