comphead commented on code in PR #7027: URL: https://github.com/apache/datafusion-comet/pull/7027#discussion_r4361445658
##########
spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala:
##########
@@ -173,24 +190,30 @@ object IcebergWriteProtoTranslation {
// --- Internals ------------------------------------------------------------
/**
- * Iceberg accepts `uncompressed`, `none` (treated as uncompressed),
`snappy`, `gzip`, `lz4`,
- * `zstd`, `brotli`. Defaults to `zstd` (since Iceberg 1.4).
+ * The native codec for a codec name, or `None` when the native writer has
no equivalent. The
+ * name is resolved the way iceberg-java does: `Parquet.WriteBuilder`
upper-cases it with
+ * `Locale.ENGLISH` and calls `CompressionCodecName.valueOf`, so the match
is case-insensitive
+ * but never trims, and `none` (ORC's spelling) is not a Parquet codec.
Unset 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 {
- case "uncompressed" | "none" => CompressionCodec.None
- case "snappy" => CompressionCodec.Snappy
- case "gzip" => CompressionCodec.Gzip
- case "lz4" => CompressionCodec.Lz4
- case "zstd" => CompressionCodec.Zstd
- case "brotli" => CompressionCodec.Brotli
- case other =>
- throw new IllegalArgumentException(s"Unsupported parquet codec
'$other'")
+ private def nativeCodec(name: String): Option[CompressionCodec] =
+ name.toUpperCase(Locale.ENGLISH) match {
+ case "UNCOMPRESSED" => Some(CompressionCodec.None)
+ case "SNAPPY" => Some(CompressionCodec.Snappy)
+ case "GZIP" => Some(CompressionCodec.Gzip)
+ case "LZ4" => Some(CompressionCodec.Lz4)
+ case "ZSTD" => Some(CompressionCodec.Zstd)
+ case "BROTLI" => Some(CompressionCodec.Brotli)
+ case _ => None
}
+
+ private def codecName(props: Map[String, String]): String =
+ props.getOrElse(Keys.ParquetCompression,
Keys.ParquetCompressionDefaultSince14)
Review Comment:
This conflicts with #6996 (open), which also rewrites `resolveCompression`
and adds its own `codecName` that trims and lowercases, plus
`compressesWithBrotli(props) = codecName(props) == "brotli"`. If #6996 is
rebased onto this `codecName`, then `BROTLI` passes this gate but fails that
comparison and skips its brotli class check, so the native writer would commit
files that cannot be read. Whichever lands second could define brotli detection
through `nativeCodec`, for example
`nativeCodec(codecName(props)).contains(CompressionCodec.Brotli)`, with a
detection test using `BROTLI`. Could you agree an order with #6996?
##########
docs/source/user-guide/latest/iceberg.md:
##########
@@ -462,26 +462,26 @@ iceberg-java resolves per-write options and
`spark.sql.iceberg.*` session overri
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`, `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.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.*`
| any value
(manifest metrics are re-derived on the JVM with Iceberg's own logic)
|
-| `write.spark.fanout.enabled`
| any value (the
native writer implements both clustered and fanout modes)
|
-| `write.target-file-size-bytes`
| any value (the
two writers can choose different roll points; see accepted divergences)
|
-| data location URI scheme
| `file`,
`memory`, `s3`, `s3a`, `gs`, matched case-sensitively (`S3://` falls back).
`s3`, `s3a` and `gs` need a bucket in the authority (`s3://bucket/...`), so a
hostless form such as `s3:/bucket/key` falls back. `gs` only when the `FileIO`
opening the data location is a `GCSFileIO`; see below
|
-| resolved `table.locationProvider()`
| Iceberg's
built-in `DefaultLocationProvider`
|
-| Hadoop S3A settings for an `s3` / `s3a` data location
| only
`fs.s3a.access.key`, `secret.key`, `session.token`, `endpoint`,
`endpoint.region`, and `path.style.access`, including their
`fs.s3a.bucket.<data-bucket>.*` forms; any other effective `fs.s3a.*` setting
falls back, apart from the ignored settings listed below
|
-| Iceberg `FileIO` S3 settings for an `s3` / `s3a` data location
| the S3
endpoint, region, static/session credentials, path-style, SSE (`none`, `s3`,
`kms`, or `custom`; not `dsse-kms`), assume-role, anonymous/config-chain
settings parsed by the pinned iceberg-rust version, plus Comet's
credential-provider class and built-in web-identity properties. When a custom
provider is configured, its vendor-owned `s3.*` / `client.*` properties are
also forwarded; unsupported Iceberg-defined S3 settings still fall back
|
-| partition spec
| any, except a
`void` field whose source column was dropped beside a live field (see below)
|
-| column types
| any except
`uuid` (Spark plans it as a string; no Arrow cast reaches `fixed(16)`) and the
v3 types `variant`, `unknown`, `timestamp_ns`, `geometry` and `geography`
|
+| Setting
| Supported
values
|
+|
-------------------------------------------------------------------------------------------------------------------------------------------
|
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
+| 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 codec must be a name iceberg-java accepts
(`uncompressed`, `snappy`, `gzip`, `lz4`, `zstd`, `brotli`, matched
case-insensitively and without trimming; `none` is ORC's spelling and falls
back); 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 |
Review Comment:
The level note still lists `none` (`ignored by both writers for
snappy/lz4/none`), but `none` now falls back, and the Scaladoc in
`IcebergWriteProtoTranslation.scala` line 99 says the same. The row could also
say that `lzo` and `lz4_raw` are accepted by iceberg-java but fall back here.
##########
spark/src/test/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslationSuite.scala:
##########
@@ -64,6 +63,32 @@ class IcebergWriteProtoTranslationSuite extends AnyFunSuite {
assert(settings.getCompression == CompressionCodec.Zstd)
}
+ test("compression codec rejection follows iceberg-java's name resolution") {
+ // iceberg-java upper-cases the name and calls
`CompressionCodecName.valueOf`: no trimming,
+ // and `none` is ORC's spelling, not a Parquet codec.
+ Seq("uncompressed", "UNCOMPRESSED", "Snappy", "gzip", "LZ4", "zstd",
"brotli").foreach {
+ codec =>
+ assert(
+ compressionCodecRejection(Map(Keys.ParquetCompression ->
codec)).isEmpty,
+ s"codec=$codec")
+ }
+ assert(compressionCodecRejection(Map.empty).isEmpty)
+ Seq("none", " gzip", "gzip ", "", "lzo", "lz4_raw", "bogus").foreach {
codec =>
+ assert(
+ compressionCodecRejection(Map(Keys.ParquetCompression ->
codec)).isDefined,
+ s"codec='$codec'")
+ }
+ }
+
+ test("compression codec parsing does not trim") {
Review Comment:
`compression codec parsing does not trim` asserts that
`buildParquetSettings` throws for `" gzip"` and `"none"`, which overlaps
`compression codec rejection follows iceberg-java's name resolution` and the
existing `unknown compression codec throws`. The throw in `resolveCompression`
is now only a defensive path behind the gate. Could these cases fold into the
existing test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
