[connectors] transaction support for Iceberg snapshots#6631
Conversation
mythical-fred
left a comment
There was a problem hiding this comment.
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 outerOptionmeans "is this call transactional", the innerOption<String>is the label — but you always construct it asSome(Some(format!("snapshot-{index}"))), so the innerOptionnever carriesNone. Either fold it toOption<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 innerNonecase exists. -
Ordering::AcqRelontransaction_indexis stronger than needed. The counter only feeds the human-readablesnapshot-{index}label — no other state is synchronized through it.Ordering::Relaxedis the right default for pure counters; leavingAcqRelinvites future readers to assume it establishes happens-before with something. -
Docs. The PR checklist has "Documentation updated" unchecked.
transaction_modeis now a public YAML option; please add a short entry to the Iceberg connector docs (behavior with/withouttimestamp_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 allmax_retriesattempts 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 alreadypush_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
left a comment
There was a problem hiding this comment.
This looks surprisingly simple, but there may be subtleties which escape me
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>
ce91266 to
86677ee
Compare
ryzhyk
left a comment
There was a problem hiding this comment.
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.
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
Breaking Changes?
not a breaking change