feat: detect Iceberg V2 writes and emit fall-back reasons

[jordepic]

Which issue does this PR close?

(Partially, second of three) Closes #4322. Part one was #4658 (merged); part three — the native writer itself — is tracked in #5308.

Rationale for this change

The first part of this work (#4658) split the Iceberg V2 write into a "writer" operator (inside AQE) and a "committer" operator, with the writer still running through iceberg-java.

This PR's goal is to detect WHICH iceberg writes we should attempt to complete, falling back otherwise. Iceberg-rust is less mature than iceberg-java when it comes to data file writing, so we establish whitelist criteria for which we are willing to write data files (I initially had blacklist critieria, but it turns out there's a lot more of those). Reasoning is listed in the markdown file attached to this commit.

What changes are included in this PR?

1. A CometNativeIcebergWrite operator which will never be selected (reserved for part 3).
2. A detailed gate which checks whether we should accelerate a write. I believe that a vast majority of iceberg tables are just using parquet files on S3 with default settings (pareto principle), so we aim to hit those common use cases, while safely falling back for the less common ones.

There are certain metadata properties (mainly to do with column stats) that iceberg-rust doesn't currently support. We're still able to provide complete parity here by virtue of comitting metadata in the JVM (will happen in part 3, slight divergence from #4487 ). Other aspects of data file writes simply can't be equal between java and rust without a lot of forking on the rust side (one example would be checking the flushed row group sizes at a granularity of x bytes). All of these conditions are listed in the associated docs.

How are these changes tested?

A new CometIcebergWriteDetectionSuite (32 tests) pins every rule: one negative test per fall-back trigger asserting the reason string, positive tests for the boundary cases (defaults, explicit metrics modes, session-conf codec overrides, s3 data locations, partitioned tables), a smoke test that every reflection accessor resolves against the Iceberg runtime on the classpath, and a test that the planner registration records a fall-back reason on the write exec. The suite passes on the spark-3.4 (Iceberg 1.5.2, Scala 2.12) and spark-3.5 (Iceberg 1.8.1) profiles.

[jordepic]

Hey @andygrove , @comphead , @mbutrovich , @parthchandra ! Here's commit 2/3 in accelerating iceberg writes. I've had to deal with a lot of ambiguity in this one when deciding what we do and don't want to accelerate, but I think it's a pretty comprehensive list at this point and covers the edge cases. Feel free to reply here, iceberg's slack, or the datafusion discord if you want to discuss more or have a call. Thanks!

[jordepic]

@andygrove would you mind kicking CI off on this one? Sorry, I fixed a formatting issue

[andygrove]

Hey @andygrove , @comphead , @mbutrovich , @parthchandra ! Here's commit 2/3 in accelerating iceberg writes. I've had to deal with a lot of ambiguity in this one when deciding what we do and don't want to accelerate, but I think it's a pretty comprehensive list at this point and covers the edge cases. Feel free to reply here, iceberg's slack, or the datafusion discord if you want to discuss more or have a call. Thanks!

Thanks @jordepic. I appreciate the offer of the direct communications channels, but we prefer to keep discussions in issues/PRs so that the whole community can follow along. All issue/PR discussions get copied to a mailing list and archived as well. This is an important part of the Apache process.

[jordepic]

@andygrove sounds good on the communications front, good point!

[andygrove]

@andygrove sounds good on the communications front, good point!

It also helps the LLMs follow along!

[andygrove]

Thanks for this @jordepic. The staging makes sense, and a couple of things I checked and liked: the SparkWrite$ prefix test in isIcebergBatchWrite cleanly excludes merge-on-read deltas, since those arrive as SparkPositionDeltaWrite$PositionDeltaBatchWrite and friends. And the storage scheme set (file, memory, s3, s3a, gs, oss) matches what iceberg_scan.rs:256-270 actually supports, including the same contains("://") parsing, so the two sides will not drift apart in confusing ways.

A handful of things I would like to work through before this goes in.

1. PropertyKeys can turn an intended fallback into a planning crash

CometIcebergNativeWrite.scala:317 and :323 are strict private vals whose initializers read PropertyKeys.ParquetRowGroupCheckMinRecordCount and the matching ...Default. Because they are strict vals in an object body, this runs during CometIcebergNativeWrite$ initialization, which is triggered by the new case op: IcebergWriteExec arm in CometExecRule.scala:301, outside getSupportLevel's try. CometExecRule.apply has no safety net either. Could those be lazy val?

The broader version of the problem: getSupportLevel catches Exception, but any failure inside PropertyKeys surfaces as ExceptionInInitializerError on first access and NoClassDefFoundError after that, and both are Error. So the reflection-failure fallback this whole design rests on would not actually catch a missing TableProperties field. Would you mind widening the catch to NonFatal?

Reachability today is narrow, since an IcebergWriteExec in the plan implies IcebergWriteStrategy ran, which implies Iceberg is on the classpath. The realistic trigger is a future or forked Iceberg renaming one of these fields, which is exactly the case the docs promise falls back.

2. The doc describes an allowlist, the code is a denylist outside write.parquet.*

iceberg-writes.md says "Detection is an allowlist: a write is eligible only when its entire effective configuration matches the table below, and anything else, any other write-affecting property, any key added by a future Iceberg version ... falls back", then "Everything else must be absent."

That is stronger than what the rules enforce. requireOnlyVettedParquetWriteProperties only covers keys starting with write.parquet., and requireNoUnvettedParquetMrProperties only covers parquet.. Everything outside those prefixes has to be named explicitly. I checked TableProperties in the Iceberg 1.11 runtime jar and these all pass the gate today:

• write.spark.fanout.enabled
• write.upsert.enabled
• write.distribution-mode (and the write.delete/merge/update variants)
• write.target-file-size-bytes
• write.object-storage.path
• write.avro.compression-codec
• write.wap.enabled

Most of those look harmless for data-file bytes, and write.target-file-size-bytes is arguably covered by the file-rolling divergence you already document. write.spark.fanout.enabled=true is the one I would want gated explicitly, because it changes the writer structure rather than a parameter. A fanout writer keeps writers open across many partitions on unsorted input, so a clustered-only native writer would produce wrong output rather than merely different bytes. Does part 3 plan to support fanout? If not, could that get its own rule?

Either way, could the "Everything else must be absent" line be reworded to describe what is actually checked?

3. The metrics-mode rules are also a denylist

The doc table says write.metadata.metrics.default and write.metadata.metrics.column.<col> support "unset, truncate(N), or full". requireMetricsModeIsNotCounts and requireMetricsModeIsNotNone only reject the literal strings counts and none, so a typo like truncat(16) or any mode a future Iceberg adds would pass.

Would it be worth inverting these into a single allowlist rule that matches the doc? That would also let requireNoPerColumnStatsEnabled and requirePropertyAbsent(ParquetBloomFilterMaxBytes) go away, since requireOnlyVettedParquetWriteProperties already subsumes both. That rule table is going to keep growing, so trimming the redundancy now seems worthwhile.

4. requireFormatVersionAtMostTwo fails open

IcebergReflection.getFormatVersion(ctx.table) match {
  case Some(v) if v >= MinUnsupportedFormatVersion => Some(s"format-version=$v unsupported")
  case _ => None
}

getFormatVersion returns None on reflection failure after only a logError, and case _ => None reads that as eligible. So the one case where Comet does not know the format version gets treated as V1/V2. Should the None case return a reason instead?

5. getOuterSparkWrite fails open on a missing this$0

IcebergReflection.scala:174-175 returns Some(batchWrite) when this$0 is absent. Downstream, getSparkWriteField(batchWrite, "table") then calls SparkWrite.class.getDeclaredField("table").get(batchWrite) on an object that is not a SparkWrite. That raises IllegalArgumentException, gets caught, logged at ERROR, and reported as "SparkWrite.table is null", which is not what happened. Would returning None be better here, so checkTriggers reports "could not unwrap SparkWrite"?

6. Compression divergence is missing from the accepted-divergences list

The doc table says write.parquet.compression-codec and compression-level accept "any value (translated to the native writer)", and the closing paragraph says compression "must match iceberg-java exactly, or the write falls back." I think those are in tension for the default case.

From the 1.11 TableProperties, PARQUET_COMPRESSION_DEFAULT_SINCE_1_4_0 is zstd and PARQUET_COMPRESSION_LEVEL_DEFAULT is unset. SparkWriteConf.dataWriteProperties() only forwards the level when it is non-null, so in the default configuration neither writer gets an explicit level and parquet-mr and parquet-rs pick different zstd defaults. Same question for lz4 and the LZ4 versus LZ4_RAW codec id.

Nothing incorrect falls out of that, since the data reads back the same and file_size_in_bytes is measured from the real file. But it does mean byte-identical output is not achievable for the most common table setting, so it seems like it belongs in the divergences list rather than under "must match exactly".

7. The CometExecRule registration is dead code with a visible side effect

Since convert always returns None and createExec throws, the arm at CometExecRule.scala:301 cannot change a plan. What it does do is run isOperatorEnabled, which with the config off (the default) tags every IcebergWriteExec with "Native support for operator IcebergWriteExec is disabled. Set spark.comet.write.iceberg.nativeAcceleration.enabled=true to enable it."

IcebergWriteExec is Comet's own operator, inserted by IcebergWriteStrategy. So anyone on the split-operator path now sees extended EXPLAIN report a fall-back on a node Comet itself added, when nothing is falling back.

Would it be cleaner to hold the registration and the throwing createExec for part 3, and keep this PR to the detection logic plus CometIcebergWriteDetectionSuite? The suite already calls getSupportLevel directly, so it does not need the rule wired up. If you would rather keep the wiring so the explain plumbing gets exercised now, that is reasonable, but the spurious reason deserves a note in the docs.

8. captureWriteExec swallows every Throwable

In CometIcebergWriteDetectionSuite.scala:

try trigger
catch { case _: Throwable => () }

Combined with capturing from onFailure, this is what makes the s3://nonexistent-bucket and hdfs://nonexistent.invalid tests work, which makes sense. The side effect is that the positive tests cannot distinguish a successful write from one that blew up for an unrelated reason, so assertSupportLevelIs[Compatible]("ok") would keep passing if INSERT INTO regressed on some profile.

Could the tests that write to the real local warehouse assert the write actually succeeded, and only the deliberately-unreachable-location tests tolerate a failure?

9. Test coverage

A few cases worth pinning down, mostly around the doc-versus-code question above:

• write.spark.fanout.enabled=true and write.target-file-size-bytes, whichever way you decide they should go. Right now their behavior is a side effect of which prefixes the allowlist covers rather than a stated decision.
• The non-Iceberg batchWrite branch and the "could not unwrap SparkWrite" branch, which are the paths item 5 touches.
• Positive scheme tests for gs, oss, memory, and an explicit file:// data path. Only s3 is covered, and plain file is reached by the fall-through rather than by matching the set.
• The reason reported when nativeAcceleration.enabled=false.

Also, the description says the suite was verified on spark-3.4 (Iceberg 1.5.2) and spark-3.5 (Iceberg 1.8.1), but the workflow change puts it in the scans bucket, which runs on all five profiles up through Spark 4.2. Iceberg 1.9 through 1.11 add write.parquet.shred-variants and the write.parquet.bloom-filter-fpp.column. / bloom-filter-ndv.column. prefixes. The allowlist should handle all three correctly, but worth confirming the newer profiles are green rather than assuming.

[andygrove]

Talking about a planned follow-up here is confusing to me. Is this PR the planned follow-up, or should this link to an issue?

[jordepic]

3/3 is the planned follow up - how would you prefer I handle it?

[andygrove]

Perhaps you could link to an issue describing the work in 3/3?

[jordepic]

See #5308

[andygrove]

Sorry, I meant could you include the link in the docs.

[jordepic]

My apologies, I knew that and forgot, all done, may need to kick CI again

[jordepic]

@andygrove would you mind another CI kick? Also, if there's a way to grant me permissions for these, I promise not to mine crypto on the CI runners :)