feat: Optionally split the Iceberg V2 write operator into distinct writer and committer operations - #4658
feat: Optionally split the Iceberg V2 write operator into distinct writer and committer operations#4658jordepic wants to merge 5 commits into
Conversation
|
Let me know what you think @mbutrovich , @comphead ! |
9a32b67 to
77030ec
Compare
|
Thanks for splitting this out of #4487, @jordepic. This is a much better review target, and the split is clean. (Reviewed with LLM assistance.) I traced the commit protocol against Spark's own V2 write and it is a faithful reproduction. The writer does per-task A few things I would like to resolve before approving, none of them large: 1. The 2. AQE exactly-once confidence. The design wraps the writer child in 3. Docs scope. You flagged this yourself. Since this PR does not ship the native write path yet, could we trim One minor note: commit messages are serialized with Also a small process point: once these splits land, it is probably worth closing #4487 (it is currently conflicting and stale) so the splits become the source of truth. |
|
Thanks for your patience with this @jordepic. I posted my AI review above. My human review is that I am in favor of getting this merged since it is gated behind a config that is disabled by default. I would prefer that we describe the functionality as experimental in the documentation for now, until this is more widely tested by multiple users. |
|
@mbutrovich @comphead Do you have any additional thoughts on this? |
|
Thanks! Let me get to this now - I think barring some much more sluggish changes on the iceberg/spark side to either change the v2 writer operator this is as clean as it'll get. |
77030ec to
8126020
Compare
|
Went ahead and fixed that feedback @andygrove . Docs simplified to reflect just this PR. test added to ensure exactly once committing under AQE. Fall back to normal spark iceberg writing if commit coordination is required. |
Thanks @jordepic. I've triggered CI. |
| case class IcebergCommitExec( | ||
| // Neither of these fields are serialized, this is all run on the driver. | ||
| @transient batchWrite: BatchWrite, | ||
| @transient refreshCache: () => Unit, |
There was a problem hiding this comment.
ideally to introduce some type for that
type RefreshCache = () => Unit
@transient
refreshCache: RefreshCache,
There was a problem hiding this comment.
Do you think it reads that much better? Just wanted to emphasize that this is a simple lambda. Fair point!
There was a problem hiding this comment.
Done -- added IcebergCommitExec.RefreshCache (type RefreshCache = () => Unit) and used it for the constructor param.
|
Thanks @jordepic I haven't yet go through PR in details, however if PR is supposed to close original ticket stating We need to check in tests there is no fallback for iceberg writes and the entire plan is fully native, I didn't see this in tests. Maybe I just missed this |
@comphead sorry - to be clear, we will fallback to using the normal java writer if we can't support the write. I was just being idealistic in that commit message - iceberg-rs can't support everything. This particular PR doesn't invoke any of that logic yet since it is only one of three separate pieces of this change. I'll have two more to put up after this. I updated the issue text to emphasize that we're trying to run the full plan using native code.
I haven't added the native acceleration just yet so those tests aren't in this particular PR, but they will be in the final code! |
I see, thanks for the clarification, would be that correct to state the PR partially closes #4322 ? |
|
Modified to reflect that, thank you @comphead ! |
jordepic
changed the title
|
I want to see what happens if I enable this and run the Iceberg Java suites. I know we're discussing merging this as experimental and disabled by default, but I just want the smoke test first. I'll run that overnight. |
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
First pass, thanks @jordepic!
| override protected def doExecute(): RDD[InternalRow] = { | ||
| val rdd = child.execute() | ||
| val factory = batchWrite.createBatchWriterFactory(PhysicalWriteInfoImpl(rdd.getNumPartitions)) | ||
| require( |
There was a problem hiding this comment.
Is this require unreachable? buildTwoOp already returns None when useCommitCoordinator() is true (IcebergWriteStrategy.scala L107), and the AQE re-plan case derives from that same guarded path.
There was a problem hiding this comment.
Correct -- with the planning-time fallback it's unreachable through IcebergWriteStrategy. I kept it as a backstop for direct construction of the node and marked it as such with a short comment. Happy to drop it instead if you'd rather not carry it.
| def serializeMessage(message: WriterCommitMessage): Array[Byte] = { | ||
| val bos = new ByteArrayOutputStream() | ||
| val oos = new ObjectOutputStream(bos) | ||
| try oos.writeObject(message) | ||
| finally oos.close() | ||
| bos.toByteArray | ||
| } | ||
|
|
||
| def deserializeMessage(bytes: Array[Byte]): WriterCommitMessage = { | ||
| val bis = new ByteArrayInputStream(bytes) | ||
| val ois = new ObjectInputStream(bis) | ||
| try ois.readObject().asInstanceOf[WriterCommitMessage] | ||
| finally ois.close() | ||
| } |
There was a problem hiding this comment.
Would Spark's serializer bound to Spark's classloader be safer here than the raw ObjectOutputStream / ObjectInputStream round-trip? My concern is that raw ObjectInputStream resolves classes with the latest user-defined loader and may fail to find Iceberg classes under --packages, REPL, or child-classloader isolation. Comet already uses SparkEnv.get.serializer for this elsewhere (CometShuffleDependency.scala L58, CometBlockStoreShuffleReader.scala L50). What do you think about Utils.serialize / Utils.deserialize with Utils.getContextOrSparkClassLoader? It might also let us drop one of two serialization passes, since the message is Java-serialized here and then serialized again by Spark's result serializer during executeCollect.
There was a problem hiding this comment.
Good catch. Switched to Utils.serialize / Utils.deserialize(bytes, Utils.getContextOrSparkClassLoader), which resolves classes through Spark's context classloader and fixes the --packages/REPL case. I kept the explicit byte-array column rather than trying to drop one of the two passes: the writer's output has to cross an ordinary RDD[InternalRow] boundary (that's what makes it visible to AQE), so the row always goes through Spark's result serializer; the inner pass is what turns the opaque WriterCommitMessage into a stable binary column.
| def runWriter( | ||
| writer: DataWriter[InternalRow], | ||
| iter: Iterator[InternalRow], | ||
| rowsMetric: SQLMetric, | ||
| projection: UnsafeProjection, | ||
| replaceDataDispatch: Option[ReplaceDataDispatchInfo]): Iterator[InternalRow] = { | ||
| val message = Utils.tryWithSafeFinallyAndFailureCallbacks(block = { | ||
| if (replaceDataDispatch.isDefined) { | ||
| runReplaceDataWriter(writer, iter, replaceDataDispatch.get, rowsMetric) | ||
| } else { | ||
| while (iter.hasNext) { | ||
| writer.write(iter.next()) | ||
| rowsMetric.add(1L) | ||
| } | ||
| } | ||
| writer.commit() |
There was a problem hiding this comment.
Should we be flushing the writer's dataWriter.currentMetricsValues during and after iteration? Spark does this in WritingSparkTask via IteratorWithMetrics, and it looks like that is how Iceberg surfaces per-task write metrics. Since runWriter only increments numOutputRows, my read is that the Iceberg task write metrics are lost when the split is enabled. Am I missing where they get picked up?
There was a problem hiding this comment.
You're right, they were dropped. IcebergWriteExec now declares write.supportedCustomMetrics() as V2 custom SQL metrics and the writer updates them from dataWriter.currentMetricsValues every CustomMetrics.NUM_ROWS_PER_UPDATE rows plus a final flush before the task commit, mirroring WritingSparkTask. Note Iceberg only implements supportedCustomMetrics() on its Spark 4.0+ modules, so on 3.x the set is empty either way.
|
|
||
| override protected def doExecute(): RDD[InternalRow] = { | ||
| val rdd = child.execute() | ||
| val factory = batchWrite.createBatchWriterFactory(PhysicalWriteInfoImpl(rdd.getNumPartitions)) |
There was a problem hiding this comment.
How does this behave when the child RDD has zero partitions? Spark's SPARK-23271 handling runs a single-partition job so the empty commit is still produced deterministically, as writeWithV2 does. mapPartitionsInternal over a zero-partition RDD produces zero messages and relies on commit([]) behaving identically, which I'm not sure is guaranteed across write types. Could we add a test with a genuinely zero-partition child? The current empty-source test (L96) uses WHERE id < 0, which I believe yields one empty partition, not zero.
There was a problem hiding this comment.
Fixed -- doExecute now substitutes a dummy single-partition RDD when the child RDD has zero partitions, mirroring the SPARK-23271 handling in writeWithV2, so the commit always goes through the task-commit protocol. Added a test appending from a genuinely zero-partition RDD (sparkContext.emptyRDD, with getNumPartitions == 0 asserted) that checks exactly one commit message was collected. You're right that the existing empty-source test yields one empty partition; kept it since it covers a different shape.
| row.getInt(0) match { | ||
| case WRITE_OPERATION => | ||
| rowProjection.project(row) | ||
| writer.write(rowProjection) | ||
| case WRITE_WITH_METADATA_OPERATION => | ||
| rowProjection.project(row) | ||
| if (metadataProjection != null) metadataProjection.project(row) | ||
| val writeWithMetadata = dataWriterWriteWithMetadataMethod.getOrElse( | ||
| throw new UnsupportedOperationException( | ||
| "DataWriter.write(metadata, row) is not available in this Spark version but the " + | ||
| s"analyzer emitted operation code $WRITE_WITH_METADATA_OPERATION")) | ||
| writeWithMetadata.invoke(writer, metadataProjection, rowProjection) | ||
| case other => | ||
| throw new IllegalArgumentException( | ||
| s"Unexpected ReplaceData operation code $other; supported: " + |
There was a problem hiding this comment.
I believe codes 5 (WRITE) and 6 (WRITE_WITH_METADATA) are correct and complete for Spark 4.0+ copy-on-write ReplaceData (DELETE emits 6, UPDATE emits 6, MERGE emits 5 and 6), and that they do not exist before 4.0. This path looks safe only because IcebergReplaceDataShim (spark-3.x) returns None, routing 3.x through the plain write(row) loop. Could we state that invariant in a comment here? And would it be worth adding the suite to CI for all four Spark lines so the version split cannot regress unnoticed?
There was a problem hiding this comment.
Added the invariant comment (the codes are 4.0+ only; on 3.x IcebergReplaceDataShim returns None so rows take the plain write(row) loop and the dispatch is never reached). On CI: the suite is already in the scans shard of both pr_build_linux.yml and pr_build_macos.yml, and on Linux that shard runs across the full matrix (Spark 3.4/3.5/4.0/4.1/4.2), so the version split is exercised on every line.
| private def buildTwoOp( | ||
| write: Write, | ||
| rel: DataSourceV2Relation, | ||
| query: LogicalPlan, | ||
| replaceDataDispatch: Option[ReplaceDataDispatchInfo]): Option[SparkPlan] = { | ||
| val batchWrite = write.toBatch | ||
| if (batchWrite.useCommitCoordinator()) { | ||
| return None | ||
| } | ||
| // To mirror Spark ReplaceData semantics we invalidate our cache of the state of | ||
| // `originalTable`. | ||
| val refresh: () => Unit = () => IcebergRefreshCacheShim.recacheByPlan(rel) | ||
| Some( | ||
| IcebergCommitExec( | ||
| batchWrite, | ||
| refresh, | ||
| // `replaceDataDispatch` may project the data into the format the writer expects. | ||
| planLater(IcebergWriteLogical(query, batchWrite, replaceDataDispatch)))) |
There was a problem hiding this comment.
The writer sets requiredChildDistribution = UnspecifiedDistribution and depends on V2Writes having injected the repartition and local sort into .query. Iceberg uses the clustered writer for partitioned writes and throws on unclustered rows (ClusteredWriter.java L66-105). The AQE test (CometIcebergWriteActionSuite.scala L110) uses an unpartitioned table, so as far as I can tell coalesce and skew-split against a clustered partitioned write is never exercised. Could we add a partitioned write under AQE with a shuffle?
There was a problem hiding this comment.
Added a partitioned INSERT ... SELECT under AQE with an 8-partition shuffle (Iceberg's clustered distribution forces the exchange, AQE coalesces it), asserting the shuffle exists, exactly one commit, and all 500 rows land. Skew-split I couldn't force deterministically at unit-test scale, but the coalesce case covers the "AQE changed the shuffle output" class of risk for the clustered writer.
| private def assertExactlyOneCommit(snapshot: WriteSnapshot): Unit = { | ||
| assert( | ||
| snapshot.snapshotDelta == 1L, | ||
| s"expected exactly 1 new Iceberg snapshot, got ${snapshot.snapshotDelta}. Plans:\n" + | ||
| snapshot.plans.mkString("\n--\n")) | ||
| val (commits, writes) = collectIcebergWriteOps(snapshot.plans) | ||
| assert( | ||
| commits.nonEmpty, | ||
| s"expected >= 1 IcebergCommitExec in captured plans, got ${commits.size}. Plans:\n" + | ||
| snapshot.plans.mkString("\n--\n")) | ||
| assert( | ||
| writes.nonEmpty, | ||
| s"expected >= 1 IcebergWriteExec in captured plans, got ${writes.size}. Plans:\n" + | ||
| snapshot.plans.mkString("\n--\n")) |
There was a problem hiding this comment.
assertExactlyOneCommit proves the commit ran once, but does it prove files were written once? And I don't see a failure test. Could we add a case that injects a task failure and a case that injects a commit failure, asserting the table is unchanged and abort ran? The abort path feels like the entire risk surface of a split write, and right now it looks untested. A speculative-execution case (spark.speculation=true) asserting a single commit and documenting the orphan-file behavior would also be reassuring.
There was a problem hiding this comment.
This flushed out a real bug, thanks: on a failed write job, IcebergCommitExec previously threw out of executeCollect without ever calling batchWrite.abort. It now aborts (with empty messages -- per-task file cleanup already happened executor-side via DataWriter.abort) before rethrowing.
Tests added: (1) a task failure injected via a throwing UDF mid-write, asserting no new snapshot and the table unchanged; (2) a commit failure -- see the concurrent-append test on the other thread, where Iceberg's commit-time validation throws after the files are written and the job-level abort runs, asserting the failed DELETE left no snapshot.
Speculation I left untested: local-mode unit tests can't reliably trigger a speculative attempt. The exposure is the same as stock Spark+Iceberg though -- Iceberg opts out of the commit coordinator upstream too, so a straggler task attempt can at worst leave orphan data files that were never committed, on either path. Happy to note that in the guide if you'd like.
| try CometListenerBusUtils.waitUntilEmpty(spark.sparkContext) | ||
| catch { case _: java.util.concurrent.TimeoutException => () } |
There was a problem hiding this comment.
Should we stop swallowing TimeoutException from waitUntilEmpty? My worry is that a dropped or late listener event turns the commit-count assertion into a silent pass. Would it be better to fail on timeout, or to read the snapshot delta directly from the Iceberg metadata rather than depending on listener delivery?
There was a problem hiding this comment.
Done -- the helper no longer swallows TimeoutException; a dropped listener event now fails the test instead of passing silently. Worth noting the commit-count half of the assertion (snapshotDelta) was already reading Iceberg's snapshot metadata directly, not the listener -- only plan capture depends on listener delivery.
| test("AppendData unpartitioned INSERT INTO routes through two-op") { | ||
| assume(icebergAvailable, "Iceberg not available in classpath") | ||
| withIcebergCatalog { warehouseDir => | ||
| createTable(warehouseDir, "append_unpart", partitionSpec = "") | ||
| val snapshot = captureWrite("append_unpart") { | ||
| spark.sql( | ||
| "INSERT INTO cat.db.append_unpart VALUES " + | ||
| "(1, 'us-east', 10.5), (2, 'us-west', 20.3), (3, 'eu', 30.7)") | ||
| } | ||
| assertExactlyOneCommit(snapshot) | ||
| assertRows("append_unpart", expectedIds = Seq(1, 2, 3)) | ||
| } | ||
| } | ||
|
|
||
| test("AppendData partitioned INSERT INTO routes through two-op") { | ||
| assume(icebergAvailable, "Iceberg not available in classpath") | ||
| withIcebergCatalog { warehouseDir => | ||
| createTable(warehouseDir, "append_part", partitionSpec = "PARTITIONED BY (region)") | ||
| val snapshot = captureWrite("append_part") { | ||
| spark.sql( | ||
| "INSERT INTO cat.db.append_part VALUES " + | ||
| "(1, 'us-east', 10.5), (2, 'us-east', 20.3), (3, 'eu', 30.7)") | ||
| } | ||
| assertExactlyOneCommit(snapshot) | ||
| assertRows("append_part", expectedIds = Seq(1, 2, 3)) | ||
| } | ||
| } | ||
|
|
||
| test("AppendData INSERT FROM SELECT survives the intervening exchange/sort") { | ||
| assume(icebergAvailable, "Iceberg not available in classpath") | ||
| withIcebergCatalog { warehouseDir => | ||
| createTable(warehouseDir, "src", partitionSpec = "") | ||
| createTable(warehouseDir, "append_from_select", partitionSpec = "PARTITIONED BY (region)") | ||
| spark.sql( | ||
| "INSERT INTO cat.db.src VALUES " + | ||
| "(1, 'us-east', 10.5), (2, 'us-west', 20.3), (3, 'eu', 30.7)") | ||
|
|
||
| val snapshot = captureWrite("append_from_select") { | ||
| spark.sql( | ||
| "INSERT INTO cat.db.append_from_select " + | ||
| "SELECT id, region, amount FROM cat.db.src ORDER BY id") | ||
| } | ||
| assertExactlyOneCommit(snapshot) | ||
| assertRows("append_from_select", expectedIds = Seq(1, 2, 3)) | ||
| } | ||
| } | ||
|
|
||
| test("AppendData on an empty source still emits a single commit") { | ||
| assume(icebergAvailable, "Iceberg not available in classpath") | ||
| withIcebergCatalog { warehouseDir => | ||
| createTable(warehouseDir, "empty_target", partitionSpec = "") | ||
| val snapshot = captureWrite("empty_target") { | ||
| spark.sql( | ||
| "INSERT INTO cat.db.empty_target SELECT id, region, amount " + | ||
| "FROM (SELECT 1 AS id, 'r' AS region, 1.0 AS amount) WHERE id < 0") | ||
| } | ||
| assertExactlyOneCommit(snapshot) | ||
| assertRows("empty_target", expectedIds = Seq.empty) | ||
| } | ||
| } |
There was a problem hiding this comment.
Could we add a multi-partition partitioned write so message collection and clustering see more than one task (several tests use coalesce(1))? A concurrent-writer test that drives Iceberg's commit-time conflict validation (validateNoConflictingData) would also be valuable. My understanding is that the shared BatchWrite instance exists to make that validation see the writer's scan state, and I don't think anything currently tests it.
There was a problem hiding this comment.
Both added: (1) a partitioned write with 4 shuffle partitions and AQE coalescing disabled, asserting numCommittedMessages >= 2 on the committer's metric; (2) a concurrent-writer test -- a gate UDF blocks the CoW DELETE's write job after its scan snapshot is pinned, a conflicting append commits in between, and the DELETE's commit fails Iceberg's serializable-isolation validation, leaving only the append's snapshot. (Runtime group filtering is disabled in that test; otherwise Iceberg's runtime file filter detects the snapshot change and aborts before the write even starts, which is a different guard than the one we want to exercise.) The second test pins exactly the shared-BatchWrite property: the committer's validation sees the writer's pinned scan state.
| ## Supported operations | ||
|
|
||
| The split-operator plan is supported on every Spark version Comet supports, with identical | ||
| coverage on each: | ||
|
|
There was a problem hiding this comment.
Could we revisit the "identical coverage on every Spark version" claim? The copy-on-write ReplaceData path looks like it differs by version: 4.0+ uses operation-coded rows with projections, while 3.4/3.5 use a plain row stream. Would it be more accurate to state that the row-level DML mechanism differs by Spark version so the docs match the shim behavior?
There was a problem hiding this comment.
Reworded -- the docs now say the supported operations are the same on every version but the row-level DML mechanism differs (4.0+ operation-coded rows with projections, 3.4/3.5 a plain row stream).
8126020 to
4bf5940
Compare
|
@mbutrovich good morning! Did you ever get any feedback from the smoke tests? |
|
Thanks for your patience @jordepic. We've been busy with getting the Comet 1.0 release ready (we'll create the release candidate next week), and I'm keen to merge this PR soon after that. |
I kicked off the smoke test in CI |
|
I ran a CI audit of this change with Summary of the 4 root causes:
None of this blocks merging this PR, since the feature is off by default, but 2, 3, and 4 look like prerequisites before it can be enabled. |
4bf5940 to
6cf9902
Compare
|
@jordepic could you fix conflicts? |
Iceberg's V2 writes run as a single `V2ExistingTableWriteExec` command that both
writes files and commits. Spark's `InsertAdaptiveSparkPlan` treats it as a leaf,
so AQE never sees the data sub-query inside it and Comet's columnar rules can't
convert the scans / shuffles / sorts that feed the write.
This injects `IcebergWriteStrategy` ahead of Spark's `DataSourceV2Strategy` (via
`experimentalMethods.extraStrategies`, which is prepended to the planner's
strategy list). For Iceberg `AppendData`, `OverwriteByExpression`, and
`OverwritePartitionsDynamic` it rewrites the single command into two operators:
IcebergCommitExec(batchWrite, refreshCache) <- committer (driver-side)
└── AdaptiveSparkPlanExec <- AQE bubble
└── IcebergWriteExec(batchWrite) <- writer (UnaryExecNode)
└── <data sub-query> <- now visible to AQE / Comet
The writer emits one row per Spark task carrying the Java-serialised
`WriterCommitMessage`; the committer collects them and calls
`BatchWrite.commit(messages)` -- the same call Iceberg-Java makes internally --
then runs Spark's post-write cache refresh.
Two design points keep this stable under AQE, which re-runs the planner on each
materialised stage's `logicalLink`:
- The writer's child is wrapped in `IcebergWriteLogical`, a stable logical
anchor, so each re-plan re-emits only the writer rather than vanishing it or
re-firing the surrounding logical write and duplicating the commit.
- The `BatchWrite` is materialised once (`Write.toBatch()` mints a fresh
instance per call) and shared between committer and writer, so commit-time
validation sees the same scan / emitted-file state the writer used.
`WriteDelta` (merge-on-read) is intentionally not intercepted: its per-task
`DeltaWriter` is row-dispatched and the native writer can't emit position-delete
files, so the split plan would add planning complexity for no acceleration.
Copy-on-write DELETE / UPDATE / MERGE (`ReplaceData`) is handled in a follow-up
commit.
Guarded by `spark.comet.write.iceberg.splitOperator.enabled` (off by default);
when off, writes go straight through Iceberg-Java unchanged. File writing still
runs through Iceberg's JVM writer -- native Parquet write lands in a later
commit.
Adds copy-on-write row-level DML to the split-operator plan. Spark lowers a CoW `DELETE` / `UPDATE` / `MERGE` on a V2 table into a `ReplaceData` logical node; `IcebergWriteStrategy` now matches it (refreshing `originalTable`'s cache) and routes it through the same committer + writer pair as appends/overwrites. The writer gains a per-row dispatch path. On Spark 4.x a `ReplaceData` rewrite prefixes each row with an operation code and carries a `ReplaceDataProjections`; `IcebergWriteExec` applies the row / metadata projections per code (WRITE / WRITE_WITH_METADATA) before handing rows to the `DataWriter`. The two-arg `DataWriter.write(metadata, row)` is looked up reflectively since it is 4.x-only. On Spark 3.4 / 3.5 the rewritten stream is already post-projection, so the shim returns `None` and the writer keeps its plain `write(row)` loop. `ReplaceDataProjections` does not exist on 3.4 / 3.5, so `ReplaceDataDispatchInfo` mirrors it as a version-neutral carrier, populated by the per-version `IcebergReplaceDataShim`. Iceberg 1.5.2 (Spark 3.4) lacks native row-level operations and instead routes UPDATE / MERGE through its own `ReplaceIcebergData` logical node. It has the same field shape as Spark's `ReplaceData` and is matched by FQCN via reflection, so the main module takes no compile dependency on iceberg-spark-extensions.
6cf9902 to
99ca922
Compare
|
@andygrove back up |
|
@jordepic do you want to just enable |
|
@andygrove - your call, if you want we could use the other PR to run in parallel. |
4 participants
Which issue does this PR close?
(Partially, first of three) Closes #4322.
Rationale for this change
Iceberg spark writes are V2 operators and contain the functionality for writing data files, metadata files, and committing to the catalog. Ultimately, Comet is only well-positioned to just accelerate data file writing (assuming they're parquet files). It is also crucial to ensure that the actual data file writing piece of the spark plan for iceberg writing is included within the AQE block of a spark plan, thereby ensuring that we re-plan writes in response to runtime decisions regarding its upstream operators.
Our split is fairly simple - we write the data files like normal in the "writer" operator, serialize its output, and pass it back to the "committer" operator. In the future, we'll target just the "writer" operator for speedup with iceberg-rust.
What changes are included in this PR?
This PR contains 5 commits.
How are these changes tested?
We have unit tests for each operator that we're replacing that ensures that the plan shape is correct, we commit to our iceberg table the proper number of times, and our iceberg table end state is correct when we scan it after a write operation. I've been running with these changes locally now and they're all performing as expected as well.