chore: add a native Iceberg write benchmark

[0lai0]

Which issue does this PR close?

Closes #5647.

Rationale for this change

#5361 landed the native (iceberg-rust) writer with numbers measured by hand. Nothing in the tree exercises it repeatably, so a regression would go unnoticed, and enabling Comet turns the scan and the writer on at once, which makes their contributions easy to conflate.

What changes are included in this PR?

A new CometIcebergWriteBenchmark: four workloads (unpartitioned insert, partitioned insert clustered, partitioned insert fanout, copy-on-write DELETE) by three arms (stock Spark, Comet scan with the iceberg-java writer, Comet scan with the native writer). The first-to-second step is the scan speedup, the second-to-third the writer's own.

Details worth knowing when reading the diff. The fanout case turns the distribution off as well as the fanout writer on, or rows arrive clustered and the writer never fans out. The copy-on-write case fills the table under stock Spark for every arm, since the two writers roll files at different points. Each iteration rebuilds the table before the timer starts, as neither an insert nor a delete is idempotent.

Two base-class changes come along: CometBenchmarkBase.warn becomes protected with the duplicate in CometCodegenDispatchBenchmark removed, and configureIcebergHadoopCatalog is extracted there. The catalog settings were repeated in four benchmarks and had drifted.
CometOperatorSerdeBenchmark registers bench_cat where the rest use benchmark_cat. The two copies in files this PR already touches are migrated, and the rest is a follow-up.

benchmarks/micro/run.py needs no change: suites are discovered from the source directory and this one runs in 202 s.

How are these changes tested?

A benchmark has no assertions, so verifyArm runs each case once before any timing and throws if the case is not what its name claims: the resulting row count, a Comet operator in the Comet arms, no Comet operator in the baseline that divides every Relative figure, and an exchange present for the clustered case and absent for the fanout one. That last check is the only thing separating those two workloads.

Run with SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometIcebergWriteBenchmark. On -Pspark-4.1, Apple M5 laptop, 4M rows over ten mixed-type columns, the recorded run puts Comet scan + native write at 2.8X unpartitioned, 1.8X clustered, 2.0X fanout and 1.3X on copy-on-write, with the Comet scan arm between 1.1X and 1.4X. A laptop is not a measurement machine, and this corpus is not the one #5361 measured, so these show the benchmark works rather than settling a number.

The copy-on-write Comet scan row is the unstable one. It is 1.1X in the recorded run and 0.7X on a repeat of the same code, with a standard deviation many times the baseline's, while its best time stays within 8% of Spark's in both. The gap is in the tail rather than a systematically slower path. The plans explain the small margin there: in all three arms the main scan is BatchScan ... IcebergCopyOnWriteScan under a ColumnarToRow, with only the runtime group-filter subquery on Comet. The Comet scan arm therefore pays for a columnar shuffle with no scan speedup to offset it, while the native writer consumes that output directly.

spotless:check -Pspark-4.1 and test-compile -Pspark-4.1 are clean, and CometIcebergReadBenchmark was re-run as the only other caller of the rewired prepareIcebergTable.

[sunchao]

Correctness

Summary and behavior

This adds a repeatable benchmark for the existing native Iceberg writer, which previously relied on the ad hoc measurements from #5361. A deterministic ten-column Parquet corpus feeds three insert shapes and a copy-on-write delete. Each workload compares stock Spark, Comet execution with the JVM writer, and Comet execution with native writing. The PR also extracts the Hadoop catalog setup and warning output into CometBenchmarkBase.

The table is recreated before each timed statement. Delete inputs are populated with stock Spark for every arm, so their starting file layouts use the same writer. The plan checks reject a missing Comet engine or unexpected Spark-baseline Comet operators, validate final row counts, and label positive-arm native fallback. Those are useful controls, but I found two gaps in what the labels guarantee:

• [P2] The clustered workload selects fanout with the newer Iceberg dependencies. Iceberg 1.10.0 and 1.11.0 can request hash distribution while resolving useFanoutWriter=true. The exchange check passes, and both the JVM and native writers follow the fanout path.
• [P2] The middle arm inherits the caller's write flags. If both native-write flags arrive through JVM configuration, the two Comet arms have identical effective settings. The positive-only native check accepts that mislabeled JVM-writer arm. Both defaults are false, so this second finding is conditional on inherited configuration.

Spark compatibility and validation

I compared the command, configuration, timer, listener, and file-partition paths against the maintained Spark 3.5 and 4.0 branches. Their command execution is eager, so timing spark.sql includes the write execution. Their session initialization preserves the inherited flags behind the second finding. This PR changes benchmark code rather than expression or write semantics. The corpus includes nullable numeric, decimal, string, binary, date, and timestamp values with ANSI disabled. Row counts check cardinality, but do not establish column-value equality or ANSI/error compatibility.

Local validation consisted of 27 source assertions and configuration controls, a read-only Make argument check, git diff --check, and inspection of the cached Iceberg 1.10.0/1.11.0 dependency bytecode through the actual writer-selection chain. These checks did not execute a Spark job, native write, or benchmark. The required maintained Spark 3.4 and 4.1 branches were unavailable, so I am not claiming compatibility validation for those versions. Inspecting the Spark 4.1 profile's Iceberg artifact does not close that Spark-source gap.

At the current-head CI check, Comet CI, CodeQL, and the title workflow require approval and have executed zero jobs. Only the label job has passed. The generated merge commit has the expected base/head parents and the same tree as the reviewed head, but no product CI execution has validated that merge tree. The author's reported Spark 4.1 compilation, formatting, and laptop benchmark results remain author-supplied evidence.

Performance

Resetting tables outside the timer avoids measuring accumulating data or progressively empty deletes. Reusing the same corpus and stock-Spark delete population also makes the arms more comparable. The timed statement still includes planning, conversion, shuffle when present, writing, and driver commit. Since the middle arm enables Comet execution as well as scanning, the reported ratios describe the combined configurations. They do not isolate the scan kernel or writer kernel.

Spark's benchmark Relative column uses the best time in each case, and the local warehouse excludes object-store behavior. The author's reported results are therefore evidence for this laptop corpus and these combined plans. I did not collect an independent timing. Correcting the writer mode and pinning the arm settings are necessary before treating the resulting rows as the intended JVM/native and clustered/fanout comparisons. No production hot path or new native allocation is added by this PR.

Design

The workload/arm matrix makes the intended experiment easy to inspect. Keeping table reset, statement execution, and plan verification separate also gives each step a clear purpose. The two findings can be fixed within that design: pin the properties and flags that define each case, then validate the resolved writer choice in both directions. A shuffle is useful evidence about distribution, but it cannot stand in for writer identity.

The explicit fallback label is a useful behavior because it prevents a missing native writer from silently retaining a native label. The same guarantee should cover the JVM-writer arm and the clustered workload. These changes preserve the existing benchmark structure and do not require a new execution framework.

Abstraction & complexity

Arm holds configuration and engine expectations, while Workload holds table setup and the statement. Both abstractions correspond directly to the experiment's two dimensions. The shared catalog helper removes repeated configuration, and moving the existing warning helper into the base class lets both benchmarks use the same output path. I found no additional actionable abstraction issue in those changes. The needed improvements are local to case configuration and plan validation.

[sunchao]

Correctness

[P2] Could this case explicitly set write.spark.fanout.enabled=false and check the resolved writer mode? In the Iceberg 1.10.0 and 1.11.0 dependencies used by the Spark 4.0 and 4.1 profiles, SparkWriteConf.writeRequirements() defaults fanout to enabled. For this unsorted table, SparkWriteUtil then requests no ordering, and useFanoutWriter() resolves to true. Hash distribution still requests an exchange, so the check below passes even though the JVM writer is FanoutDataWriter. Comet also reads that resolved flag from SparkWrite and selects its native fanout writer. As written, this measures fanout with hash distribution and fanout without distribution, leaving the clustered writer unmeasured while labeling its result as clustered. Pinning fanout off for this workload, plus checking the actual mode rather than only the presence of an exchange, would keep the intended writer comparison valid.

[0lai0]

Thanks @sunchao for review. Fixed, Two things were needed:

• Pinned 'write.spark.fanout.enabled'='false', which selects the clustered writer.
• Added ALTER TABLE ... WRITE DISTRIBUTED BY PARTITION LOCALLY ORDERED BY <partition column> via a new Workload.orderBy. The local sort is what the clustered writer needs so it doesn't error on interleaved partitions, and it makes hasOrdering true so the writer choice no longer rests on the fanout default.
  verifyArm now checks for a Sort node through Workload.expectSort — the plan-visible, version-independent mark of that required ordering, which the exchange alone couldn't show.
  The clustered row measured fanout with hash distribution before this. Re-running and will update the description.

[sunchao]

Correctness

[P2] Could the Comet scan arm explicitly disable the native-write flags and reject an unexpected CometIcebergWriteExec? new SparkConf() inherits JVM spark.* properties, and the supported BENCH_MAVEN_OPTS invocation can supply both -Dspark.comet.write.iceberg.splitOperator.enabled=true and -Dspark.comet.iceberg.write.enabled=true. Neither session initialization nor this arm clears them. In that configuration the two Comet arms have identical effective flags, so a supported write runs natively in both, while verifyArm accepts the middle arm because it checks native presence only when expectNativeWrite is true. The table therefore reports a JVM-versus-native comparison that actually compares the native writer with itself. The defaults are false, so this requires inherited configuration. Explicit per-arm values and the negative plan check would make the reported engine independent of the caller's settings.

[0lai0]

Fixed. The Comet scan arm now pins spark.comet.write.iceberg.splitOperator.enabled=false and spark.comet.iceberg.write.enabled=false explicitly, and verifyArm rejects an unexpected CometIcebergWriteExec when expectNativeWrite=false, with an error pointing at BENCH_MAVEN_OPTS inheritance as the likely cause.

[sunchao]

Re-reviewed d1658d92 against b7f35b6a. Both previous P2 findings are addressed:

• Clustered/fanout selection: the workloads now explicitly pin fanout off/on. I traced those values through the Iceberg 1.10.0 and 1.11.0 resolver, JVM writer selection, and Comet's native writer-mode dispatch. The clustered case also declares hash distribution with local ordering and checks for the sort.
• Inherited native-write flags: the JVM arm now pins both flags off, while the native arm pins them on. The original inherited-configuration control now keeps the arms distinct for all four flag combinations, and the new negative check rejects an unexpected native writer.

I found no new or remaining verified P1/P2 issues in the increment. The additional table-order setup remains outside the timer.

Validation: 45 focused source/bytecode assertions and configuration controls passed, along with the supported Make argument check and git diff --check. These checks did not execute a Spark/native job or benchmark. The author's earlier timings are not measurements of this revised workload. Maintained Spark 3.5/4.0 sources were checked, while the 3.4/4.1 source gaps remain.

Current-head Comet CI and CodeQL still require approval and have run zero jobs. Only the label job has passed.

[andygrove]

@0lai0 This PR fails CI when added to the merge queue due to a linting error. Could you fix this?

--- /__w/datafusion-comet/datafusion-comet/spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergWriteBenchmark.scala
+++ <expected fix>
@@ -417,7 +417,7 @@
         throw new IllegalStateException(
           s"${arm.name}: '${workload.title}' expected a sort in the plan to be $expected but " +
             s"found $had, so it would measure the wrong writer. Iceberg reaches the clustered " +
-            s"writer only when the write has a required ordering; without one it uses the fanout " +
+            "writer only when the write has a required ordering; without one it uses the fanout " +
             s"writer. Plans:\n${plans.mkString("\n--\n")}")
       }
     }
@@ -430,9 +430,9 @@
     if (!arm.expectNativeWrite && nativeWrites.nonEmpty) {
       throw new IllegalStateException(
         s"${arm.name}: expected the iceberg-java writer but the plan contains " +
-          s"CometIcebergWriteExec, so this case would measure the native writer under a " +
-          s"JVM-writer label. This arm pins the native-write flags off, so they are leaking in " +
-          s"from the session defaults (e.g. `-Dspark.comet.iceberg.write.enabled=true` in " +
+          "CometIcebergWriteExec, so this case would measure the native writer under a " +
+          "JVM-writer label. This arm pins the native-write flags off, so they are leaking in " +
+          "from the session defaults (e.g. `-Dspark.comet.iceberg.write.enabled=true` in " +
           s"BENCH_MAVEN_OPTS). Plans:\n${plans.mkString("\n--\n")}")