Skip to content

feat(iceberg): add upsert support to write_iceberg#7255

Open
DrishyaShah wants to merge 2 commits into
Eventual-Inc:mainfrom
DrishyaShah:feat/write-iceberg-upsert
Open

feat(iceberg): add upsert support to write_iceberg#7255
DrishyaShah wants to merge 2 commits into
Eventual-Inc:mainfrom
DrishyaShah:feat/write-iceberg-upsert

Conversation

@DrishyaShah

Copy link
Copy Markdown

Changes Made

Adds mode="upsert" to DataFrame.write_iceberg, delegating to PyIceberg's
Table.upsert() (requires pyiceberg>=0.9.0). A new join_cols parameter
specifies the column(s) used to match existing rows: matched rows are
updated, unmatched rows are inserted. Also adds a corresponding
IcebergTable.upsert(df, join_cols) convenience method on the catalog API,
mirroring the existing append/overwrite methods.

Validation added:

  • Raises if pyiceberg < 0.9.0 (when Table.upsert was introduced upstream)
  • Requires join_cols to be non-empty
  • Rejects checkpoint=... combined with mode="upsert" (not supported)

Return value differs from append/overwrite: instead of file-level
ADD/DELETE operations, upsert returns a 1-row summary DataFrame with
rows_updated and rows_inserted, taken from PyIceberg's UpsertResult.

Note: unlike append/overwrite, the upsert path materializes the full
DataFrame into a single local PyArrow table via to_arrow() before handing
it to PyIceberg. This isn't avoidable in this PR — PyIceberg's own
upsert() API only accepts one in-memory pa.Table, so there's no
distributed write path available for this mode yet.

Docs updated in docs/connectors/iceberg.md (new usage example, and the
FAQ entry that previously said upserts weren't supported).

AI Usage

Implementation was written with Claude Code assistance and manually
verified by the author. Additionally verified in review: full test suite
run (94 passed, 1 skipped, including 3 new upsert tests), PyIceberg's
Table.upsert() signature and UpsertResult fields cross-checked against
an installed pyiceberg 0.11, lint clean on changed lines, and the branch
rebased onto current upstream main with no residual conflicts.

Related Issues

Closes #3844

Adds mode=upsert to DataFrame.write_iceberg and IcebergTable.upsert(),
delegating to PyIceberg Table.upsert() (requires pyiceberg>=0.9.0).
Matched rows are updated, unmatched rows are inserted. Returns a summary
DataFrame with rows_updated and rows_inserted counts.

Closes Eventual-Inc#3844

Co-authored-by: Claude Code <noreply@anthropic.com>
@github-actions github-actions Bot added the feat label Jul 11, 2026
@greptile-apps

greptile-apps Bot commented Jul 11, 2026

Copy link
Copy Markdown
Contributor

Greptile Summary

This PR adds Iceberg upsert support to the write APIs. The main changes are:

  • DataFrame.write_iceberg accepts mode="upsert" and join_cols.
  • Upserts delegate to PyIceberg and return updated/inserted row counts.
  • IcebergTable.upsert mirrors the existing catalog write helpers.
  • Iceberg docs and write tests cover the new mode.

Confidence Score: 4/5

The upsert path needs a fix for partial update inputs before merging.

  • The all-columns upsert flow is covered by tests.
  • The changed coercion step can write nulls for omitted table columns.
  • That can produce incorrect row data for common keyed update inputs.

daft/dataframe/dataframe.py

Important Files Changed

Filename Overview
daft/dataframe/dataframe.py Adds upsert validation, PyIceberg delegation, and summary output; the schema coercion can turn omitted update columns into explicit nulls.
daft/catalog/__iceberg.py Adds a catalog-level upsert helper that delegates to DataFrame.write_iceberg.
tests/io/iceberg/test_iceberg_writes.py Adds basic read-after-write, version, and join column validation coverage for upsert.
docs/connectors/iceberg.md Documents the new upsert write mode and updates the FAQ.

Reviews (1): Last reviewed commit: "feat(iceberg): add upsert support to wri..." | Re-trigger Greptile

Comment thread daft/dataframe/dataframe.py
mode="upsert" previously null-padded any table column missing from the
input DataFrame before calling PyIceberg's upsert(), which replaces
matched rows in full -- so omitted columns were silently overwritten
with NULL on update. PyIceberg's upsert() has no partial-column update
path (it errors on schema mismatch internally), so require the full
column set upfront with a clear ValueError instead of padding.

Also validates join_cols reference real DataFrame columns, replacing a
raw KeyError from inside PyIceberg with an actionable error.

Addresses Greptile review feedback on the upsert PR.

@srilman srilman left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@DrishyaShah I would prefer not adding this to Daft itself because unlike the other modes in write_iceberg which works incrementally, upsert seems to materializes all outputs upfront (via the self.to_arrow()), which will likely cause an OOM. Rather than using the PyIceberg upsert API, it would be cool if we had some more native support instead. Ideally it:

  • Does the merging incrementally rather than materializing all at once
  • Works in both native and distributed

What do you think? Would love to see an issue is this is interesting to you

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Support upsert functionality in write_iceberg

2 participants