Repository navigation
Add icebergUpsert - #60
Conversation
Replace rows whose join columns match an incoming record and insert the rest, as a v2 position delete plus an append in one guarded commit. Matches are found by field id at physical parquet positions against the loaded snapshot; conflicts surface and are not retried, and staged files are kept after a failed commit. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
platypii
left a comment
There was a problem hiding this comment.
Thanks for this. The core approach (filter manifests by key values, scan candidate files for matches, stage position deletes, then append) is sound. A few things I'd like addressed before merging, roughly in priority order in the inline comments.
| const file = await resolver.reader(file_path, Number(file_size_in_bytes)) | ||
| const parquetMetadata = await readParquetMetadata(file) | ||
| // Resolve physical columns by field id, so renamed columns still match. | ||
| const fileSchema = fileIcebergSchema(parquetMetadata, schema, nameMappings) |
There was a problem hiding this comment.
This throws for any data file that predates the join column being added to the schema, or where the key is an identity-partition column not stored in the parquet. The read path handles both by falling back to the partition value or initial-default. Here we should do the same, or skip the file when no row can match, rather than failing the whole upsert.
There was a problem hiding this comment.
Fixed via readDataFile: key columns missing from a file now come from the identity partition value or initial-default, like reads. Added a test with the partition column absent from the parquet and a join column added after the file was written.
Note
Auto-replied by the GitHub Copilot app.
| { type: 'assert-default-sort-order-id', 'default-sort-order-id': metadata['default-sort-order-id'] }, | ||
| ] | ||
| if (!deletes.length) { | ||
| const appended = await icebergStageAppend({ tableUrl, metadata, records, resolver }) |
There was a problem hiding this comment.
Ordering problem: the delete parquet, delete manifest and manifest list are written before the new records are validated. If prepareAppend throws on a bad record, the thrown error carries no writtenFiles, so the delete files are orphaned with no way to clean up. Suggest running prepareAppend first (it doesn't depend on the delete snapshot), then the delete stage, then the snapshot. The existing "rejects invalid input before writing any file" test only covers key-column errors, so it misses this.
Also: manifests get fetched three to four times per upsert: in findMatchingRows, again inside findDataFilePartitions, and once per buildSnapshotUpdate. findMatchingRows already has every data entry with partition and spec id, so pass those through instead of re-reading.
There was a problem hiding this comment.
prepareAppend now runs before any delete file is written, and the test covers a non-key record the writer rejects after a key matched. Manifests are read once by the scan: the matched candidates' partition and spec ids are passed to the delete writer (extracted from icebergStagePositionDelete as writePositionDeletes), and with a single snapshot there's one buildSnapshotUpdate.
Note
Auto-replied by the GitHub Copilot app.
| const requirements = [ | ||
| { type: 'assert-table-uuid', uuid: metadata['table-uuid'] }, | ||
| { type: 'assert-ref-snapshot-id', ref: 'main', 'snapshot-id': normalizeCurrentSnapshotId(metadata['current-snapshot-id']) }, | ||
| { type: 'assert-current-schema-id', 'current-schema-id': metadata['current-schema-id'] }, |
There was a problem hiding this comment.
This commits two snapshots, a delete then an append. Spark MERGE and Java RowDelta commit a single overwrite snapshot holding both the delete manifest and the data manifest. The intermediate state here is visible to time travel and incremental readers as a half-applied upsert. buildSnapshotUpdate already takes arbitrary newManifests, so one snapshot with operation overwrite should work and would also remove the applyUpdates requirements-merging special case.
There was a problem hiding this comment.
Now one overwrite snapshot holding the data manifest and the delete manifest(s), built with a single buildSnapshotUpdate. The applyUpdates / requirements merging is gone; the schema, spec and sort-order assertions are appended to that snapshot's requirements.
Note
Auto-replied by the GitHub Copilot app.
| const columns = keyFields.map(field => { | ||
| const physical = fileSchema.fields.find(f => f.id === field.id) | ||
| if (!physical) throw new Error(`join column ${field.name} not found in data file ${file_path}`) | ||
| return sanitize(physical.name) |
There was a problem hiding this comment.
Rows already removed by an equality delete are treated as live, so we'd stage position deletes for invisible rows. fetchDeleteMaps already returns equalityDeleteGroups; filter the way read.js does with deleteFileAppliesToDataEntry and equalityMatch.
There was a problem hiding this comment.
Fixed: equality deletes now apply through readDataFile, so invisible rows aren't matched. The new test covers this with an equality delete committed the way another engine would.
Note
Auto-replied by the GitHub Copilot app.
| * @param {Set<string>} options.keys | ||
| * @returns {Promise<{file_path: string, pos: bigint}[]>} | ||
| */ | ||
| async function findMatchingRows({ metadata, resolver, schema, keyFields, records, keys }) { |
There was a problem hiding this comment.
This re-implements per-file column resolution and delete application from readDataFile, with less coverage (the missing-column and equality-delete comments below are both divergences). Worth considering a shared helper, or a positions option on readDataFile, so upsert gets read semantics for free.
There was a problem hiding this comment.
Went with a positions option on readDataFile (adds _pos to each row). The scan now just reads the key columns with a key filter through it, so it gets read semantics for column resolution, defaults and both delete types. Happy to switch to a separate helper if you'd prefer that.
Note
Auto-replied by the GitHub Copilot app.
| if (normalizeCurrentSnapshotId(metadata['current-snapshot-id']) === null || !records.length) return [] | ||
| /** @type {Record<string, {$in: any[]}>} */ | ||
| const filter = {} | ||
| for (const field of keyFields) { |
There was a problem hiding this comment.
This downloads every delete file in the snapshot, including equality deletes we never use and position deletes for partitions with no candidate. Filter to content === 1 and to entries that apply to at least one candidate first.
There was a problem hiding this comment.
Now only delete files that apply to at least one candidate are fetched (deleteFileAppliesToDataEntry). I kept applicable equality deletes rather than filtering to content === 1, since the equality-delete thread needs them.
Note
Auto-replied by the GitHub Copilot app.
| * @param {Field['type']} type | ||
| * @returns {boolean} | ||
| */ | ||
| function keyValueValid(value, type) { |
There was a problem hiding this comment.
Number.isInteger accepts out-of-range values for int columns, like 2**40. Those can never match a stored INT32 and get handed to the writer late. Add a 32-bit range check.
There was a problem hiding this comment.
Added the 32-bit range check, plus a test case with 2 ** 40.
Note
Auto-replied by the GitHub Copilot app.
- Commit the deletes and new data as one overwrite snapshot - Find matches with readDataFile (new positions option), so missing columns, partition values, defaults and equality deletes follow reads - Write data before deletes so a rejected record orphans no delete files - Reuse scanned entries instead of re-reading manifests; fetch only delete files that apply to a candidate - Range-check int join keys Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
platypii
left a comment
There was a problem hiding this comment.
Looks good. All the review points are addressed in the code, and I also checked an int key promoted to long; matching still works. One non-blocking nit inline.
| const addedSize = prepared.addedFilesSize + deleteSize | ||
| /** @type {Snapshot['summary']} */ | ||
| const summary = { | ||
| operation: 'overwrite', |
There was a problem hiding this comment.
Nit, non-blocking: the old delete snapshot carried changed-partition-count, and this overwrite summary dropped it. writePositionDeletes already returns changedPartitions, so it's a one-liner to keep it.
Thanks for Icebird. Being able to read and write Iceberg tables directly from JavaScript, with no JVM or Spark in the loop, has been very useful for us. This PR adds the one write operation we kept missing.
Icebird can append and delete rows by position, but it can't update rows by key. To replace a row today, callers have to scan the table, work out the physical row positions themselves, and stage a delete and an append. They also have to commit the two atomically, or a concurrent writer can cause lost updates or duplicate rows.
icebergUpsertdoes this in one call. The API is modelled on pyiceberg'sTable.upsert(join_cols, defaulting to the identifier fields, rejecting duplicate keys):joinColsmatch an incoming record is replaced. Unmatched records are inserted. Composite keys are supported.joinColsdefaults to the schema'sidentifier-field-ids.icebergRewrite. Staged files are kept after a failed commit, because the commit may have succeeded.Unlike pyiceberg, which rewrites whole data files (copy-on-write), this writes small position-delete files (merge-on-read). Writes stay cheap, and
icebergRewritecan compact the deletes later.Limitations: v2 tables only. Join columns must be top-level
int,longorstringfields. Matched rows are always rewritten, even if unchanged.Follow-ups: v3 deletion vectors, skipping unchanged rows (as pyiceberg does), and limiting how many files the scan reads at once.
Note: The code was written with AI assistance, based on an approach we use in an internal project. It was then reviewed and tested against Icebird's conventions and existing write helpers.