fix: fall back for brotli Iceberg writes when Hadoop's BrotliCodec cannot be loaded

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

@@ -466,7 +466,7 @@ A write is eligible only when ALL of the following hold:
 | ------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | resolved write format (`write-format` option overlaid on `write.format.default`)                                                            | `parquet`                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
 | `format-version`                                                                                                                            | `1`, `2` or `3` (on `3`, writes that carry row lineage fall back; see below)                                                                                                                                                                                                                                                                                                                                                                                                    |
-| `write.parquet.compression-codec` / `compression-level` / `row-group-size-bytes` / `page-size-bytes` / `page-row-limit` / `dict-size-bytes` | the sizes/limit must be positive Java ints (`Integer.parseInt` semantics — no trimming, no values past `Int.MaxValue` — matching iceberg-java, whose writer fails on anything else); `compression-level` must be a Java int within the native writer's per-codec range (zstd 1–22, gzip 0–9, brotli 0–11; ignored by both writers for snappy/lz4/none) — iceberg-java never validates the level, so an out-of-range value falls back rather than becoming a native task failure |
+| `write.parquet.compression-codec` / `compression-level` / `row-group-size-bytes` / `page-size-bytes` / `page-row-limit` / `dict-size-bytes` | the sizes/limit must be positive Java ints (`Integer.parseInt` semantics — no trimming, no values past `Int.MaxValue` — matching iceberg-java, whose writer fails on anything else); `compression-level` must be a Java int within the native writer's per-codec range (zstd 1–22, gzip 0–9, brotli 0–11; ignored for snappy/lz4/none), which iceberg-java never validates, so other levels fall back; `brotli` needs Hadoop's `BrotliCodec` class on the classpath (see below) |
 | `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`                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
@@ -550,6 +550,15 @@ cannot write through it either, so the write fails with iceberg-java's own error
 the native writer ([#6141](https://github.com/apache/datafusion-comet/issues/6141)). A spec whose
 fields are all `void` writes unpartitioned and stays eligible.
 
+A write that compresses with `brotli` falls back unless Hadoop's
+`org.apache.hadoop.io.compress.BrotliCodec` class can be loaded. parquet-mr compresses and
+decompresses brotli pages with that class, and neither Spark nor Iceberg ships it. Without a brotli
+codec jar on the classpath, iceberg-java fails the write, and no JVM Parquet reader, Spark's or
+Iceberg's, can read a brotli file. parquet-rs compresses brotli without the class, so a native write
+would commit files that only a native reader could read. The gate looks the class up on the
+driver, through the class loader that Spark adds application jars to
+([#6960](https://github.com/apache/datafusion-comet/issues/6960)).
+
 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,
@@ -726,8 +735,16 @@ a data file but not what any reader computes from it:
   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.
+  default) are not gated: the native writer does not read them, and they can only shift
+  compressed bytes, which are already accepted as divergent. A numeric gzip level is the one
+  exception, because it changes whether iceberg-java can write at all. Hadoop reads
+  `zlib.compress.level` as the name of a `ZlibCompressor.CompressionLevel` constant, such as
+  `BEST_SPEED`, so iceberg-java fails a gzip write whose level is a number, set either as
+  `write.parquet.compression-level` or as `zlib.compress.level` in the Hadoop configuration. It
+  fails on Hadoop 3.4, and on Hadoop 3.3 when Hadoop's native zlib library is loaded. The native
+  writer writes such a table at its `write.parquet.compression-level`, or at gzip's default level,
+  and any reader can decompress the pages, whatever the level
+  ([#6960](https://github.com/apache/datafusion-comet/issues/6960)).
 
 #### Manifest metadata visible to later readers
 
@@ -800,9 +817,6 @@ Which entries apply depends on which native paths are enabled:
 
 The native Iceberg writer is disabled by default. With `spark.comet.write.iceberg.enabled=true`:
 
-- **Brotli tables.** The native writer commits brotli-compressed files. JVM Parquet readers can
-  read them only if Hadoop's `BrotliCodec` is on the classpath. iceberg-java fails the write
-  instead ([#6960](https://github.com/apache/datafusion-comet/issues/6960)).
 - **`write.parquet.compression-codec=none`.** The native writer writes the table uncompressed, and
   iceberg-java rejects the value. The native writer also trims or lowercases a few other settings
   that iceberg-java reads as written

spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala

@@ -35,6 +35,7 @@ import org.apache.comet.objectstore.NativeConfig
 import org.apache.comet.serde.{CometOperatorSerde, Compatible, OperatorOuterClass, SupportLevel, Unsupported}
 import org.apache.comet.serde.OperatorOuterClass.Operator
 import org.apache.comet.serde.QueryPlanSerde.exprToProto
+import org.apache.comet.util.ClassLoaders
 
 object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
 
@@ -99,6 +100,8 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
   private val ParquetMrPropertyPrefix = "parquet."
   private val CometS3CredentialProviderClassProperty =
     "s3.comet.credential.provider.class"
+  // The codec parquet-mr loads for brotli. No Spark or Iceberg distribution ships it.
+  private[comet] val BrotliCodecClass = "org.apache.hadoop.io.compress.BrotliCodec"
 
   // Hadoop S3A settings are not forwarded wholesale. Keep this allow-list in lockstep with
   // NativeConfig.s3aSuffixToIcebergGlobalKey: every admitted setting must be translated into the
@@ -423,6 +426,7 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
     requireParquetPageVersionDefault,
     requireShredVariantsDisabled,
     requireNativeSupportedCompressionLevel,
+    requireLoadableBrotliCodec,
     requireOnlyVettedParquetWriteProperties,
     requirePropertyAbsent(
       PropertyKeys.ParquetEnableDictionary,
@@ -594,6 +598,31 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
   private val requireNativeSupportedCompressionLevel: TriggerRule = ctx =>
     IcebergWriteProtoTranslation.compressionLevelRejection(ctx.properties)
 
+  // parquet-mr compresses brotli pages, and decompresses them for every JVM reader, with Hadoop's
+  // `BrotliCodec`, which it loads by name. Neither Spark nor Iceberg ships that class. Without a
+  // brotli codec jar, iceberg-java fails the write, and parquet-rs would write files that only a
+  // native reader can read, so decline and let the write fail the way iceberg-java fails it.
+  // https://github.com/apache/datafusion-comet/issues/6960
+  private val requireLoadableBrotliCodec: TriggerRule = ctx =>
+    if (IcebergWriteProtoTranslation.compressesWithBrotli(ctx.properties) &&
+      !isLoadable(BrotliCodecClass)) {
+      Some(
+        s"${PropertyKeys.ParquetCompressionCodec}=brotli needs $BrotliCodecClass, which is not " +
+          "on the classpath (iceberg-java fails the write and JVM readers cannot read the files)")
+    } else {
+      None
+    }
+
+  // Through the context class loader, which Spark adds application jars to, and without
+  // initializing the class, so the check runs none of the codec's code.
+  private def isLoadable(className: String): Boolean =
+    try {
+      ClassLoaders.contextOrDefault(getClass.getClassLoader).loadClass(className)
+      true
+    } catch {
+      case _: ClassNotFoundException | _: LinkageError => false
+    }
+
   private val requireOnlyVettedParquetWriteProperties: TriggerRule = ctx =>
     ctx.properties
       .find { case (k, _) =>

spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala

@@ -100,6 +100,9 @@ object IcebergWriteProtoTranslation {
       }
     }
 
+  /** True when the write compresses with brotli. Unlike `resolveCompression`, never throws. */
+  def compressesWithBrotli(props: Map[String, String]): Boolean = codecName(props) == "brotli"
+
   /** Builds the parquet settings message. Pure: no SparkWrite or Iceberg `Table` access. */
   def buildParquetSettings(
       props: Map[String, String],
@@ -176,12 +179,8 @@ object IcebergWriteProtoTranslation {
    * Iceberg accepts `uncompressed`, `none` (treated as uncompressed), `snappy`, `gzip`, `lz4`,
    * `zstd`, `brotli`. Defaults to `zstd` (since Iceberg 1.4).
    */
-  private def resolveCompression(props: Map[String, String]): CompressionCodec = {
-    val raw = props
-      .get(Keys.ParquetCompression)
-      .map(_.trim.toLowerCase(Locale.ROOT))
-      .getOrElse(Keys.ParquetCompressionDefaultSince14)
-    raw match {
+  private def resolveCompression(props: Map[String, String]): CompressionCodec =
+    codecName(props) match {
       case "uncompressed" | "none" => CompressionCodec.None
       case "snappy" => CompressionCodec.Snappy
       case "gzip" => CompressionCodec.Gzip
@@ -191,7 +190,13 @@ object IcebergWriteProtoTranslation {
       case other =>
         throw new IllegalArgumentException(s"Unsupported parquet codec '$other'")
     }
-  }
+
+  /** The codec name, trimmed and lowercased, or Iceberg's default when the write sets none. */
+  private def codecName(props: Map[String, String]): String =
+    props
+      .get(Keys.ParquetCompression)
+      .map(_.trim.toLowerCase(Locale.ROOT))
+      .getOrElse(Keys.ParquetCompressionDefaultSince14)
 
   /**
    * Iceberg leaves `write.parquet.compression-level` null by default and lets each parquet writer

spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala

@@ -377,6 +377,40 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
     }
   }
 
+  // parquet-mr writes and reads brotli pages with Hadoop's BrotliCodec, which neither Spark nor
+  // Iceberg ships. Without it iceberg-java fails the write, and a native write would commit files
+  // that no JVM reader can read (#6960). The writes are planned, not run, so that iceberg-java
+  // never tries to load the class.
+  test("fall-back: brotli compression without Hadoop's BrotliCodec on the classpath") {
+    withDetectionCatalog { _ =>
+      createTable(
+        "brotli_table",
+        partitionSpec = "",
+        properties = Some("'write.parquet.compression-codec'='brotli'"))
+      createTable("brotli_session", partitionSpec = "")
+      val tableWrite = planInsertWriteExec(s"$catalog.$ns.brotli_table")
+      // The session setting reaches the gate through SparkWrite.writeProperties. withSQLConf
+      // returns Unit on Spark 3.4/3.5, hence the var.
+      var sessionWrite: IcebergWriteExec = null
+      withSQLConf("spark.sql.iceberg.compression-codec" -> "brotli") {
+        sessionWrite = planInsertWriteExec(s"$catalog.$ns.brotli_session")
+      }
+      Seq("brotli_table" -> tableWrite, "brotli_session" -> sessionWrite).foreach {
+        case (table, writeExec) =>
+          assertUnsupportedContains(
+            writeExec,
+            table,
+            "write.parquet.compression-codec=brotli",
+            CometIcebergNativeWrite.BrotliCodecClass)
+          // With a brotli codec jar on the classpath, the write is native again.
+          withContextClassLoaderResolving(CometIcebergNativeWrite.BrotliCodecClass) {
+            val support = CometIcebergNativeWrite.getSupportLevel(writeExec)
+            assert(support.isInstanceOf[Compatible], s"expected Compatible for $table: $support")
+          }
+      }
+    }
+  }
+
   test("fall-back: write.parquet.bloom-filter-max-bytes set") {
     withDetectionCatalog { _ =>
       createTable(
@@ -1778,6 +1812,22 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
     }
   }
 
+  /**
+   * Runs `f` with a context class loader that resolves `className`, as one with the class's jar
+   * on it would. The gate asks only whether the name resolves, so an unrelated class stands in.
+   */
+  private def withContextClassLoaderResolving[T](className: String)(f: => T): T = {
+    val thread = Thread.currentThread()
+    val original = thread.getContextClassLoader
+    thread.setContextClassLoader(new ClassLoader(original) {
+      override protected def loadClass(name: String, resolve: Boolean): Class[_] =
+        if (name == className) classOf[org.apache.hadoop.io.compress.GzipCodec]
+        else super.loadClass(name, resolve)
+    })
+    try f
+    finally thread.setContextClassLoader(original)
+  }
+
   /**
    * Runs Spark's transition insertion followed by [[EliminateRedundantTransitions]] over a
    * hand-built `CometIcebergWriteExec -> CometSparkToColumnarExec -> source` plan and returns the