[connectors] Implement Iceberg input follow mode - #6749
Conversation
Add `follow` and `snapshot_and_follow` modes to the Iceberg input connector. After the starting snapshot, the connector polls the catalog for new snapshots and, for each one, diffs it against its parent to find added and removed data files, ingesting added rows as inserts and removed rows as deletes. Only the manifests a snapshot added are read, so each step costs work proportional to the change, not to the table size. Follow mode requires a catalog: `metadata_location` points at a fixed snapshot and cannot observe new commits. currently handles copy-on-write only and rejects merge-on-read delete files with an actionable error; compaction snapshots (`operation = replace`) are skipped. Each followed snapshot forms one transaction and one checkpointable resume point, so a restart resumes after the last fully ingested snapshot. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com>
Signed-off-by: feldera-bot <feldera-bot@feldera.com>
mythical-fred
left a comment
There was a problem hiding this comment.
Nice, clean expansion of the Iceberg connector. Approach is right: diff each committed snapshot against its parent via manifest walk, ingest added files as inserts and removed files as deletes, one transaction and one checkpointable resume point per snapshot. Rejecting merge-on-read at the manifest level with an actionable error (pointing to #6165) is exactly the shape.
A few non-blocking observations:
timestamp_columnvalidation is now gated onself.config.snapshot(), sofollow-only mode silently accepts atimestamp_columnand never runsvalidate_timestamp_column. If the timestamp column is still meant to be applied to followed rows (forLATENESS), the validation should run in follow-only mode too; if it is ignored, please reject the combination at config-parse time with a clear message. Right now it's a quiet no-op.sleep(FOLLOW_POLL_INTERVAL)infollow_loop_inneris not cancellation-aware. At 1s it's harmless, but if the interval ever becomes configurable, wrap it in atokio::select!alongside the receiver so pause/stop don't have to wait it out.follow_records_totallumps inserts and deletes together. Splitting intofollow_inserts_total/follow_deletes_totalis cheap and would make follow-mode observability actually useful; you'd immediately see whether a table is churning or growing.snapshots_afterwalks the current-snapshot chain back toafter_id; correct, and the "no longer an ancestor" bail path is the right behavior. Worth adding a one-line comment that this is O(history), which is fine because it only runs on catalog refresh but could be pathological on very long histories.ci apply automatic fixesfollow-up commit is fine (bot-authored, per Gerd's 2026-04-10 rule).
CI is green. Signed off, no AI-attribution trailers.
| feature = "iceberg-tests-rest", | ||
| feature = "iceberg-tests-s3tables" | ||
| feature = "iceberg-tests-s3tables", | ||
| feature = "iceberg-tests-follow" |
There was a problem hiding this comment.
this feature isn't enabled in CI, so the tests won't run
There was a problem hiding this comment.
but it requires tabulario/iceberg-rest:1.6.0 for rest catalog
- catchup: group all commits seen in one poll into one transaction - always: one transaction per followed snapshot - end_snapshot_id: stop after ingesting the named snapshot Also recompute projection/name mapping per poll for schema evolution. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com>
mythical-fred
left a comment
There was a problem hiding this comment.
The follow-phase transaction modes and end_snapshot_id land cleanly on top of the snapshot-transaction machinery from the earlier commits.
Design is sound:
catchupgroups every commit visible in one poll into a single Feldera transaction; the window opens onbegin_catchup_windowkeyed on the batch's last snapshot id (correctly not sequence number — V1 tables number every snapshot 0, so a>=sequence check would commit immediately) and closes when the loop reaches that id.alwaysgives each followed snapshot its own transaction;snapshotandnonekeep their prior follow-phase semantics.- The Rollback path in
push_changed_filesnow abandons the committed catchup label and continues the window under a fresh one, so a mid-window retry's rows are not attributed to a transaction the queue already committed. Nice detail — easy to miss. finish_at_end_snapshotpushes aResumeInfo::eoientry withcommit_transaction: true, which closes any open catchup transaction on the way out;reset_catchup_windowclears the target metric. So anend_snapshot_idlanding mid-window still commits cleanly.- Projection and name mapping are recomputed per poll, so additive schema evolution is picked up between snapshots. Feldera-side relation is still fixed at pipeline creation, and the docs already say non-additive changes require a restart.
validate_end_snapshotrejectsend_snapshot_id == start_snapshot_idand any id not insnapshots_after(start)with clear messages. Snapshot ids are unordered so the exact-match-in-current-history check is exactly right — a not-yet-committed id would follow forever otherwise.
Tests cover: catchup grouping across a poll batch, always mode per-snapshot transactions, end_snapshot_id in follow-only and snapshot_and_follow, rejection paths for the misconfigured end bound. Config-side changes to IcebergTransactionMode are additive and #[serde(default)] at the parent keeps existing configs readable.
New metrics input_connector_iceberg_follow_transaction_starts (counter) and input_connector_iceberg_catchup_target_sequence_number (gauge, -1 when no window open) are documented and follow the connector's existing metric naming.
The single failing CI check is main at the self-hosted-runner container-startup step, with no PR-authored code in the failing steps. Unrelated to the diff; needs a rerun.
No AI trailers.
Approving. Prior APPROVE stands for the base of the PR; this new commit is a clean feature extension.
ryzhyk
left a comment
There was a problem hiding this comment.
Approving but please address the comments (especially re the ordering of inserts and deletes) before merging.
| // Inserts before deletes: a followed snapshot's added and | ||
| // removed files are disjoint, so ordering only affects | ||
| // intermediate state, not the final Z-set. | ||
| self.push_changed_files( |
There was a problem hiding this comment.
I believe the orders matters and deletes must be applied before inserts. Feldera must process a delete before an insert with the same PK, otherwise the insert can get lost.
A copy-on-write rewrite emits the same primary key in both the removed and added files. Applying the deletes first lets the insert win on a keyed relation, matching the Delta connector. - add a follow test for the copy-on-write delete path - reword the catchup and end_snapshot_id docs; drop the Delta comparison Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com>
6927aa1 to
11e072a
Compare
Add
followandsnapshot_and_followmodes to the Iceberg input connector. After the starting snapshot, the connector polls the catalog for new snapshots and, for each one, diffs it against its parent to find added and removed data files, ingesting added rows as inserts and removed rows as deletes. Only the manifests a snapshot added are read, so each step costs work proportional to the change, not to the table size.Follow mode requires a catalog:
metadata_locationpoints at a fixed snapshot and cannot observe new commits. currently handles copy-on-write only and rejects merge-on-read delete files with an actionable error; compaction snapshots (operation = replace) are skipped. Each followed snapshot forms one transaction and one checkpointable resume point, so a restart resumes after the last fully ingested snapshot.Ref: #6165
Describe Manual Test Plan
tested on staging that we are following the table correctly, even after 1h
Checklist
Breaking Changes?
should not be breaking change