andygrove commented on code in PR #7027: URL: https://github.com/apache/datafusion-comet/pull/7027#discussion_r4541738845
##########
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:
Neither order needs agreeing now: #6996 detects brotli by upper-casing the
name itself (`47bb72e3c9` there), so `BROTLI` reaches the class check with or
without this PR's `codecName`. I merged the two branches locally and the
translation and detection suites pass (106). Whichever lands second has textual
conflicts in `resolveCompression`/`codecName` and in the `iceberg.md` table
row. The resolution is this PR's codec resolution plus that PR's brotli
detection. If this lands first, I'll rebase #6996 and define its detection
through `nativeCodec`, as you suggested.
##########
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:
Fixed in `2215217f5d`. The level note and the Scaladoc now say
`snappy/lz4/uncompressed`, and the codec row says that `lzo` and `lz4_raw` are
accepted by iceberg-java but fall back here. Prettier re-padded the table, so
the docs diff is longer than the wording change.
##########
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:
Folded in `2215217f5d`. The padded `" gzip"` and `none` cases now sit in
`unknown compression codec throws` next to `lzo`, with a note that the throw is
the defensive path behind the gate, and the separate no-trim test is gone.
--
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]
