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

.github/workflows/pr_build_linux.yml

@@ -315,6 +315,7 @@ jobs:
               org.apache.comet.CometIcebergEncryptionSuite
               org.apache.comet.CometIcebergRewriteActionSuite
               org.apache.comet.CometIcebergWriteActionSuite
+              org.apache.comet.CometIcebergWriteDetectionSuite
               org.apache.comet.iceberg.IcebergReflectionSuite
               org.apache.comet.csv.CometCsvNativeReadSuite
               org.apache.comet.CometFuzzTestSuite

.github/workflows/pr_build_macos.yml

@@ -131,6 +131,7 @@ jobs:
               org.apache.comet.CometIcebergEncryptionSuite
               org.apache.comet.CometIcebergRewriteActionSuite
               org.apache.comet.CometIcebergWriteActionSuite
+              org.apache.comet.CometIcebergWriteDetectionSuite
               org.apache.comet.iceberg.IcebergReflectionSuite
               org.apache.comet.csv.CometCsvNativeReadSuite
               org.apache.comet.CometFuzzTestSuite

docs/source/user-guide/latest/iceberg-writes.md

@@ -41,7 +41,8 @@ writes into two operators:
 Data files are still written by iceberg-java; only the plan shape changes. The split makes the
 write's input visible to AQE and to Comet's columnar rules, and it is the groundwork for a
 planned follow-up in which Comet writes the data files natively via
-[iceberg-rust](https://github.com/apache/iceberg-rust).
+[iceberg-rust](https://github.com/apache/iceberg-rust), tracked in
+[#5308](https://github.com/apache/datafusion-comet/issues/5308).
 
 ## Configuration
 
@@ -57,6 +58,9 @@ spark.sql.catalog.<name>.warehouse=...
 
 # Split-operator plan (experimental, off by default)
 spark.comet.write.iceberg.splitOperator.enabled=true
+
+# Native-write eligibility detection (experimental, off by default; requires the split plan)
+spark.comet.write.iceberg.nativeAcceleration.enabled=true
 ```
 
 ## Supported operations
@@ -95,3 +99,102 @@ The rewrite is skipped — and the write runs through Spark's stock combined ope
 
 In every fallback case the write is planned as if Comet were absent; there is no correctness
 trade-off, only no plan change.
+
+## Native Parquet write eligibility
+
+A planned follow-up ([#5308](https://github.com/apache/datafusion-comet/issues/5308)) replaces
+the `IcebergWrite` operator's per-task Parquet write with
+[iceberg-rust](https://github.com/apache/iceberg-rust). The native writer must produce the same
+outcome as iceberg-java — the same Parquet features, statistics, and manifest metadata — so a
+write is only eligible when every table property it depends on is one the native path reproduces
+exactly. `spark.comet.write.iceberg.nativeAcceleration.enabled` enables this eligibility check;
+with the current release the native writer itself is not yet wired in, so every write still runs
+through iceberg-java and the check's outcome is reported as a fall-back reason in Comet's
+extended EXPLAIN output.
+
+**Most Iceberg write settings are not supported.** 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, any
+value outside the supported set, any reflection failure while inspecting the write — falls
+back to iceberg-java with a reason reported in extended EXPLAIN. Checks run on the effective
+configuration: table properties overlaid with `SparkWrite.writeProperties`, which is where
+iceberg-java resolves per-write options and `spark.sql.iceberg.*` session overrides.
+
+A write is eligible only when ALL of the following hold:
+
+| Setting                                                                                                                                     | Supported values                                                                  |
+| ------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------- |
+| resolved write format (`write-format` option overlaid on `write.format.default`)                                                            | `parquet`                                                                         |
+| `format-version`                                                                                                                            | `1` or `2`                                                                        |
+| `write.parquet.compression-codec` / `compression-level` / `row-group-size-bytes` / `page-size-bytes` / `page-row-limit` / `dict-size-bytes` | any value (translated to the native writer)                                       |
+| `write.parquet.row-group-check-min-record-count`                                                                                            | unset or `100` (the default)                                                      |
+| `write.parquet.row-group-check-max-record-count`                                                                                            | unset or `10000` (the default)                                                    |
+| `write.parquet.page-version`                                                                                                                | unset or `v1`                                                                     |
+| `write.parquet.shred-variants`                                                                                                              | unset or `false` (Spark 4.x / Iceberg 1.11 resolve this into every parquet write) |
+| `write.parquet.variant-inference-buffer-size`                                                                                               | any value (only meaningful when shredding, which is gated)                        |
+| `write.parquet.bloom-filter-enabled.column.<col>`                                                                                           | unset or `false`                                                                  |
+| `write.metadata.metrics.default`                                                                                                            | unset, `truncate(N)`, or `full`                                                   |
+| `write.metadata.metrics.column.<col>`                                                                                                       | unset, `truncate(N)`, or `full`                                                   |
+| `write.spark.fanout.enabled`                                                                                                                | any value (the native writer implements both clustered and fanout modes)          |
+| `write.target-file-size-bytes`                                                                                                              | any value (file rolling cadence differs; see accepted divergences)                |
+| data location URI scheme                                                                                                                    | `file`, `memory`, `s3`, `s3a`, `gs`, `oss`                                        |
+| partition spec                                                                                                                              | any (but see partition paths under accepted divergences)                          |
+
+Within the namespaces that shape data-file bytes — `write.parquet.*` and `parquet.*` —
+everything not listed above must be absent: unvetted `write.parquet.*` keys (e.g.
+`bloom-filter-max-bytes`, `stats-enabled.column.*`, keys added by future Iceberg versions),
+metrics modes outside the supported set (`counts`, `none`, or unparseable values), any
+`parquet.*` table property (including `parquet.enable.dictionary`), and any `parquet.*` key in
+the session Hadoop configuration (with `HadoopFileIO`-backed output those reach iceberg-java's
+writer but not the native one). Also gated explicitly: any `encryption.*` key,
+`write.object-storage.enabled=true`, `write.location-provider.impl`, and `io-impl`.
+
+Other `write.*` properties are intentionally not gated because they cannot make the native
+writer produce different data files: distribution and ordering settings shape the Spark plan
+identically on both paths, WAP / branch / snapshot properties act on the JVM committer,
+`write.avro.*` / `write.orc.*` apply only to formats already excluded, and merge-on-read
+settings route the write through `WriteDelta`, which the split plan never intercepts. Every
+rule is pinned by `CometIcebergWriteDetectionSuite`.
+
+Manifest `DataFile` metrics will be assembled on the JVM at commit time using Iceberg's own
+`MetricsConfig` logic, so iceberg-java's metadata decisions — metrics modes, the
+inferred-column cap (`write.metadata.metrics.max-inferred-column-defaults`), bound truncation,
+and list/map bounds suppression — are respected exactly regardless of what the native writer
+reports. The `counts`/`none` restrictions above remain only until that assembly lands.
+
+## Accepted divergences behind the toggle
+
+Some differences between parquet-mr and the pinned parquet-rs / iceberg-rust are unconditional —
+they apply to every native write and cannot be configured away. Enabling
+`nativeAcceleration.enabled` accepts them:
+
+- Footer key-value metadata differs: native files carry an `ARROW:schema` entry and no
+  `iceberg.schema` entry; iceberg-java files are the opposite.
+- The Parquet root schema element is named `arrow_schema` (iceberg-java: `table`).
+- `created_by` identifies parquet-rs, not parquet-mr.
+- No page CRC checksums and no page-header statistics (parquet-mr writes both by default).
+- Dictionary-encoded pages are labeled `RLE_DICTIONARY` (parquet-mr v1 files: `PLAIN_DICTIONARY`).
+- Fixed-length binary columns (`uuid`, `fixed`, decimals with precision > 18) are not
+  dictionary-encoded (parquet-mr dictionary-encodes them).
+- Row-group boundaries: parquet-mr flushes by byte size at a record-count check cadence,
+  parquet-rs buffers by row count. File rolling and file naming follow the same cadence-style
+  differences (iceberg-java checks the target file size every 1000 rows and names files
+  `<partition>-<task>-<operation>-<count>`; iceberg-rust checks per batch and uses a
+  process-local counter).
+- Partition paths are not URL-escaped: iceberg-java percent-encodes partition directory names
+  and values (`region=a%2Fb`), iceberg-rust writes them raw (`region=a/b`). Readers resolve
+  files through manifest metadata, not paths, so query results are unaffected — but the
+  directory layout differs from iceberg-java's, and partition values containing characters
+  that are invalid in a URI (`:`, `#`, newline) may produce paths that `HadoopFileIO`-based
+  readers cannot open.
+- Compressed page bytes are implementation-defined: the codec and any explicit level are
+  translated, but parquet-rs and parquet-mr embed different encoder implementations and
+  defaults (zstd default levels, LZ4 framing), so byte-identical output is not achievable even
+  for a default `zstd` table. The decompressed data is identical. For the same reason,
+  codec-level side channels (`zlib.compress.level`, `compression.brotli.quality`,
+  `io.compression.codec.zstd.level` — the last is present in every Hadoop configuration by
+  default) are not gated: they can only shift compressed bytes, which are already accepted as
+  divergent.
+
+All content not listed above — the logical data, encodings for non-FLBA columns, statistics
+values, and manifest metadata — must match iceberg-java exactly, or the write falls back.

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -131,6 +131,16 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
+  val COMET_ICEBERG_NATIVE_WRITE_ENABLED: ConfigEntry[Boolean] =
+    conf("spark.comet.write.iceberg.nativeAcceleration.enabled")
+      .category(CATEGORY_TESTING)
+      .doc(
+        "Whether to delegate the executor-side Parquet write to Comet's native (iceberg-rust) " +
+          "writer when the table's properties allow it. Requires " +
+          "`spark.comet.write.iceberg.splitOperator.enabled = true`. Off by default.")
+      .booleanConf
+      .createWithDefault(false)
+
   val COMET_ICEBERG_DATA_FILE_CONCURRENCY_LIMIT: ConfigEntry[Int] =
     conf("spark.comet.scan.icebergNative.dataFileConcurrencyLimit")
       .category(CATEGORY_SCAN)

spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala

@@ -58,6 +58,7 @@ object IcebergReflection extends Logging {
     val TABLE = "org.apache.iceberg.Table"
     val PARTITIONING = "org.apache.iceberg.Partitioning"
     val SPARK_WRITE = "org.apache.iceberg.spark.source.SparkWrite"
+    val TABLE_PROPERTIES = "org.apache.iceberg.TableProperties"
 
     // Iceberg 1.5.2 uses its own `ReplaceIcebergData` due to lack of `ReplaceData` in Spark 3.4.
     val REPLACE_ICEBERG_DATA = "org.apache.spark.sql.catalyst.plans.logical.ReplaceIcebergData"
@@ -146,6 +147,29 @@ object IcebergReflection extends Logging {
   def isIcebergSparkWrite(write: Any): Boolean =
     sparkWriteClassOpt.exists(_.isInstance(write))
 
+  def isIcebergBatchWrite(batchWrite: Any): Boolean = {
+    if (batchWrite == null) return false
+    batchWrite.getClass.getName.startsWith(ClassNames.SPARK_WRITE + "$")
+  }
+
+  def getOuterSparkWrite(batchWrite: Any): Option[Any] = {
+    if (batchWrite == null) None
+    else {
+      try {
+        val field = batchWrite.getClass.getDeclaredField("this$0")
+        field.setAccessible(true)
+        Option(field.get(batchWrite))
+      } catch {
+        case _: NoSuchFieldException =>
+          None
+        case e: Exception =>
+          logError(
+            s"Iceberg reflection failure: outer SparkWrite from BatchWrite: ${e.getMessage}")
+          None
+      }
+    }
+  }
+
   def isReplaceIcebergData(plan: Any): Boolean =
     plan != null && plan.getClass.getName == ClassNames.REPLACE_ICEBERG_DATA
 
@@ -1033,6 +1057,73 @@ object IcebergReflection extends Logging {
     getTableProperties(table).filter(_.containsKey("encryption.key-id")).map { props =>
       Option(props.get("encryption.data-key-length")).map(_.toInt).getOrElse(16)
     }
+
+  private def getSparkWriteField(sparkWrite: Any, fieldName: String): Option[Any] =
+    sparkWriteClassOpt.flatMap { cls =>
+      try {
+        val field = cls.getDeclaredField(fieldName)
+        field.setAccessible(true)
+        Option(field.get(sparkWrite))
+      } catch {
+        case _: NoSuchFieldException => None
+        case e: Exception =>
+          logError(
+            s"Iceberg reflection failure: Failed to read SparkWrite.$fieldName: ${e.getMessage}")
+          None
+      }
+    }
+
+  def getFormatFromSparkWrite(sparkWrite: Any): Option[String] =
+    getSparkWriteField(sparkWrite, "format")
+      .map(_.toString.toLowerCase(java.util.Locale.ROOT))
+
+  def getTableFromSparkWrite(sparkWrite: Any): Option[Any] =
+    getSparkWriteField(sparkWrite, "table")
+
+  def getWritePropertiesFromSparkWrite(sparkWrite: Any): Option[Map[String, String]] = {
+    import scala.jdk.CollectionConverters._
+    getSparkWriteField(sparkWrite, "writeProperties")
+      .map(_.asInstanceOf[java.util.Map[String, String]].asScala.toMap)
+  }
+
+  private lazy val tablePropertiesClassOpt: Option[Class[_]] =
+    tryLoadClass(ClassNames.TABLE_PROPERTIES)
+
+  def tablePropertyConstant(fieldName: String): String =
+    readTablePropertiesField(fieldName).asInstanceOf[String]
+
+  def tablePropertyIntConstant(fieldName: String): Int =
+    readTablePropertiesField(fieldName).asInstanceOf[Integer].intValue()
+
+  private def readTablePropertiesField(fieldName: String): Any = {
+    val cls = tablePropertiesClassOpt.getOrElse(
+      throw new IllegalStateException(s"${ClassNames.TABLE_PROPERTIES} is not on the classpath"))
+    try cls.getField(fieldName).get(null)
+    catch {
+      case e: NoSuchFieldException =>
+        throw new IllegalStateException(
+          s"${ClassNames.TABLE_PROPERTIES}.$fieldName not found " +
+            "(unsupported Iceberg version?)",
+          e)
+    }
+  }
+
+  def getDataLocation(table: Any): Option[String] =
+    try {
+      val locationProviderMethod =
+        findMethodInHierarchy(table.getClass, "locationProvider").getOrElse(
+          throw new NoSuchMethodException(
+            s"locationProvider() not found on ${table.getClass.getName}"))
+      val provider = locationProviderMethod.invoke(table)
+      val newDataLocMethod = provider.getClass.getMethod("newDataLocation", classOf[String])
+      newDataLocMethod.setAccessible(true)
+      val location = newDataLocMethod.invoke(provider, "").asInstanceOf[String]
+      Some(location.stripSuffix("/"))
+    } catch {
+      case e: Exception =>
+        logError(s"Iceberg reflection failure: Failed to get data location: ${e.getMessage}", e)
+        None
+    }
 }
 
 /**

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -298,6 +298,9 @@ case class CometExecRule(session: SparkSession)
       case op: DataWritingCommandExec =>
         convertToComet(op, CometDataWritingCommand).getOrElse(op)
 
+      case op: IcebergWriteExec if CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.get(op.conf) =>
+        convertToComet(op, CometIcebergNativeWrite).getOrElse(op)
+
       // For AQE broadcast stage on a Comet broadcast exchange
       case s @ BroadcastQueryStageExec(_, _: CometBroadcastExchangeExec, _) =>
         convertToComet(s, CometExchangeSink).getOrElse(s)