Skip to content

[connectors] transaction support for Iceberg snapshots#6631

Merged
swanandx merged 1 commit into
iceberg-6165-modernize-snapshotfrom
iceberg-6165-transactions
Jul 15, 2026
Merged

[connectors] transaction support for Iceberg snapshots#6631
swanandx merged 1 commit into
iceberg-6165-modernize-snapshotfrom
iceberg-6165-transactions

Conversation

@swanandx

Copy link
Copy Markdown
Member

Add a transaction_mode option (none | snapshot) to the Iceberg connector. In snapshot mode the initial snapshot is ingested as Feldera transactions: the whole snapshot in one transaction when timestamp_column is unset, or one per lateness range when it is set.

Each execute_df call maps to one transaction. Parser entries carry the start label (the input queue starts the transaction on the first flushed entry) and a commit entry is pushed once the dataframe drains. This rides the existing non-fault-tolerant queue() path, which already honors the transaction fields on queue entries.

Input-buffer staging is deferred: it needs the flush-based FT queue path that follow mode will introduce.

Describe Manual Test Plan

Checklist

  • Unit tests added/updated
  • Integration tests added/updated
  • Documentation updated
  • Changelog updated

Breaking Changes?

not a breaking change

@mythical-fred mythical-fred left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

APPROVE with a few small suggestions.

Clean, well-scoped addition of transaction_mode = none | snapshot on top of #6623. Default is none (non-breaking); serde round-trip is covered by the two new unit tests; two new integration tests exercise the unordered and ordered transactional paths. Commit is signed off with no AI trailers, checklist notes explicitly that input-buffer staging is deferred to the FT queue work — nice, honest scoping.

A few things to consider (all non-blocking):

  • Option<Option<String>> is misleading. The outer Option means "is this call transactional", the inner Option<String> is the label — but you always construct it as Some(Some(format!("snapshot-{index}"))), so the inner Option never carries None. Either fold it to Option<String> (None = no transaction, Some = transactional with a label) or introduce a small enum (Transactional::No / Yes(Option<String>)) if you want to preserve the "labeled or unlabeled transaction" distinction for future modes. As written, readers will assume the inner None case exists.

  • Ordering::AcqRel on transaction_index is stronger than needed. The counter only feeds the human-readable snapshot-{index} label — no other state is synchronized through it. Ordering::Relaxed is the right default for pure counters; leaving AcqRel invites future readers to assume it establishes happens-before with something.

  • Docs. The PR checklist has "Documentation updated" unchecked. transaction_mode is now a public YAML option; please add a short entry to the Iceberg connector docs (behavior with/without timestamp_column, and that non-transactional is the default). Feldera's connector reference is where users will look for this.

  • Retry + commit interaction. In execute_df, if all max_retries attempts fail, we bail without pushing a commit entry — leaving the started transaction (from the first flushed entry of the failed attempts) dangling on the queue. And on the successful retry path, the previous attempts have already push_entry'd rows into the queue (that's pre-existing behavior from #6623), so a snapshot transaction spanning several retry attempts can include re-flushed rows from earlier failed attempts. Neither is new in this PR, but making the whole thing a transaction now makes the semantics user-visible. Worth a comment near the retry loop describing the intended behavior on partial failure — and, if feasible in a follow-up, aborting the started transaction on terminal failure rather than leaving it open.

@mihaibudiu mihaibudiu 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.

This looks surprisingly simple, but there may be subtleties which escape me

Comment thread crates/iceberg/src/input.rs
Add a transaction_mode option (none | snapshot) to the Iceberg
connector. In snapshot mode the initial snapshot is
ingested as Feldera transactions: the whole snapshot in one transaction
when timestamp_column is unset, or one per lateness range when it is set.

Each execute_df call maps to one transaction. Parser entries carry the
start label (the input queue starts the transaction on the first flushed
entry) and a commit entry is pushed once the dataframe drains. This rides
the existing non-fault-tolerant queue() path, which already honors the
transaction fields on queue entries.

Input-buffer staging is deferred: it needs the flush-based FT queue path
that follow mode will introduce.

Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com>
@swanandx
swanandx force-pushed the iceberg-6165-transactions branch from ce91266 to 86677ee Compare July 14, 2026 18:35
Comment thread crates/adapters/src/test/iceberg.rs

@ryzhyk ryzhyk 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.

Looks good. Before creating more PRs for the iceberg connector, let's make sure we have some integration tests. Also, let's start on iceberg QA workloads.

@swanandx
swanandx merged commit 327bfa2 into iceberg-6165-modernize-snapshot Jul 15, 2026
1 check passed
@swanandx
swanandx deleted the iceberg-6165-transactions branch July 15, 2026 10:06
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants