From 0eb9d0c3e321fb81d1861c5102effb4bfd668827 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 14 Oct 2026 12:44:49 -0600 Subject: [PATCH] fix: fall back for brotli Iceberg writes when Hadoop's BrotliCodec cannot be loaded parquet-mr compresses and decompresses brotli pages with Hadoop's org.apache.hadoop.io.compress.BrotliCodec, which neither Spark nor Iceberg ships. Without it, iceberg-java fails a brotli write, but the native writer wrote the files anyway, and Spark without Comet could not read them back. The eligibility gate now declines brotli unless that class loads. A gzip write with a numeric level, which iceberg-java also fails, stays native: its files are ordinary gzip pages that every reader can read. The user guide lists it as an accepted divergence. Closes #6960. --- docs/source/user-guide/latest/iceberg.md | 26 +++++++--- .../operator/CometIcebergNativeWrite.scala | 29 +++++++++++ .../IcebergWriteProtoTranslation.scala | 19 ++++--- .../CometIcebergWriteDetectionSuite.scala | 50 +++++++++++++++++++ 4 files changed, 111 insertions(+), 13 deletions(-) diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index b24bc258539..a993e1837a1 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/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 diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala index b385668291d..d91e609aa6c 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala +++ b/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, _) => diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala b/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala index 531f3c227dd..4698843c7df 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala +++ b/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 diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala index a102d142c83..073b4ba8c71 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala +++ b/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