This is an automated email from the ASF dual-hosted git repository.
mbutrovich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new bc53ad5c56 feat: Iceberg table format V3: native table decryption,
fall back for other V3 features (#4991)
bc53ad5c56 is described below
commit bc53ad5c56eff6e4e3c2d702a6869b7cf28352ae
Author: Matt Butrovich <[email protected]>
AuthorDate: Thu Jul 23 18:25:23 2026 -0400
feat: Iceberg table format V3: native table decryption, fall back for other
V3 features (#4991)
---
.github/actions/java-test/action.yaml | 14 +-
.github/workflows/pr_build_linux.yml | 1 +
.github/workflows/pr_build_macos.yml | 1 +
dev/diffs/iceberg/1.11.0.diff | 25 ++
docs/source/contributor-guide/roadmap.md | 6 +-
docs/source/user-guide/latest/iceberg.md | 22 +-
native/Cargo.lock | 76 +++---
native/Cargo.toml | 8 +-
.../core/src/execution/operators/iceberg_scan.rs | 57 +++++
native/core/src/execution/planner.rs | 8 +-
native/proto/src/proto/operator.proto | 14 ++
spark/pom.xml | 7 +
.../apache/comet/iceberg/IcebergReflection.scala | 63 +++++
.../org/apache/comet/rules/CometScanRule.scala | 86 ++++++-
.../serde/operator/CometIcebergNativeScan.scala | 29 +++
.../org/apache/comet/CometIcebergTestBase.scala | 2 +-
.../apache/comet/CometIcebergEncryptionSuite.scala | 261 +++++++++++++++++++++
.../org/apache/comet/iceberg/CometTestKMS.scala | 63 +++++
18 files changed, 682 insertions(+), 61 deletions(-)
diff --git a/.github/actions/java-test/action.yaml
b/.github/actions/java-test/action.yaml
index 7ec586bd10..a5ca166591 100644
--- a/.github/actions/java-test/action.yaml
+++ b/.github/actions/java-test/action.yaml
@@ -50,6 +50,18 @@ runs:
cd native
cargo build --release
+ # Evaluate hashFiles now, on the clean checkout, and reuse the result in
the
+ # cache key below. The cache step's inputs are re-evaluated when its post
+ # (cache-save) phase runs, and by then the test run has left trees under
the
+ # workspace (e.g. the Hive-catalog warehouse from
CometIcebergEncryptionSuite)
+ # that make a fresh '**/pom.xml' walk throw "Fail to hash files under
+ # directory". A precomputed string keeps the save phase from re-walking
them.
+ - name: Compute Maven cache key
+ id: maven-cache-key
+ if: ${{ runner.os != 'macOS' }}
+ shell: bash
+ run: echo "hash=${{ hashFiles('**/pom.xml') }}" >> "$GITHUB_OUTPUT"
+
- name: Cache Maven dependencies
# TODO: remove next line after working again
# temporarily work around
https://github.com/actions/runner-images/issues/13341
@@ -60,7 +72,7 @@ runs:
path: |
~/.m2/repository
/root/.m2/repository
- key: ${{ runner.os }}-java-maven-${{ hashFiles('**/pom.xml') }}
+ key: ${{ runner.os }}-java-maven-${{
steps.maven-cache-key.outputs.hash }}
restore-keys: |
${{ runner.os }}-java-maven-
diff --git a/.github/workflows/pr_build_linux.yml
b/.github/workflows/pr_build_linux.yml
index 6aaac53204..019ae27722 100644
--- a/.github/workflows/pr_build_linux.yml
+++ b/.github/workflows/pr_build_linux.yml
@@ -312,6 +312,7 @@ jobs:
org.apache.spark.sql.comet.ParquetEncryptionITCase
org.apache.comet.exec.CometNativeReaderSuite
org.apache.comet.CometIcebergNativeSuite
+ org.apache.comet.CometIcebergEncryptionSuite
org.apache.comet.CometIcebergRewriteActionSuite
org.apache.comet.iceberg.IcebergReflectionSuite
org.apache.comet.csv.CometCsvNativeReadSuite
diff --git a/.github/workflows/pr_build_macos.yml
b/.github/workflows/pr_build_macos.yml
index 36ed5f9405..0c02fdf8ce 100644
--- a/.github/workflows/pr_build_macos.yml
+++ b/.github/workflows/pr_build_macos.yml
@@ -128,6 +128,7 @@ jobs:
org.apache.spark.sql.comet.ParquetEncryptionITCase
org.apache.comet.exec.CometNativeReaderSuite
org.apache.comet.CometIcebergNativeSuite
+ org.apache.comet.CometIcebergEncryptionSuite
org.apache.comet.CometIcebergRewriteActionSuite
org.apache.comet.iceberg.IcebergReflectionSuite
org.apache.comet.csv.CometCsvNativeReadSuite
diff --git a/dev/diffs/iceberg/1.11.0.diff b/dev/diffs/iceberg/1.11.0.diff
index a404ded20a..ca56ab9b93 100644
--- a/dev/diffs/iceberg/1.11.0.diff
+++ b/dev/diffs/iceberg/1.11.0.diff
@@ -251,6 +251,31 @@ index 507d7b313b..3b73dcc014 100644
.config("spark.ui.enabled", "false")
.config(DISABLE_UI)
.enableHiveSupport()
+diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTestBase.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTestBase.java
+index 8ce60f6275..75112b4354 100644
+---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTestBase.java
++++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTestBase.java
+@@ -48,7 +48,6 @@ import org.apache.iceberg.types.Types.LongType;
+ import org.apache.iceberg.types.Types.MapType;
+ import org.apache.iceberg.types.Types.StructType;
+ import org.apache.iceberg.util.DateTimeUtil;
+-import org.assertj.core.api.Condition;
+ import org.junit.jupiter.api.Test;
+ import org.junit.jupiter.api.io.TempDir;
+ import org.junit.jupiter.params.ParameterizedTest;
+@@ -385,12 +384,6 @@ public abstract class AvroDataTestBase {
+ .build());
+
+ assertThatThrownBy(() -> writeAndValidate(writeSchema, expectedSchema))
+- .has(
+- new Condition<>(
+- t ->
+- IllegalArgumentException.class.isInstance(t)
+- ||
IllegalArgumentException.class.isInstance(t.getCause()),
+- "Expecting a throwable or cause that is an instance of
IllegalArgumentException"))
+ .hasMessageContaining("Missing required field: missing_str");
+ }
+
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java
index b61ecfa2f4..d696e85139 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java
diff --git a/docs/source/contributor-guide/roadmap.md
b/docs/source/contributor-guide/roadmap.md
index b8ae8a310f..59aba8e7c4 100644
--- a/docs/source/contributor-guide/roadmap.md
+++ b/docs/source/contributor-guide/roadmap.md
@@ -52,14 +52,14 @@ significant family of Spark expressions in one effort.
Native Parquet scans (`CometNativeScanExec`) support Dynamic Partition Pruning
(DPP) both with and without
Adaptive Query Execution. Non-AQE DPP landed in [#4011] and AQE DPP with
broadcast reuse landed in [#4112].
-Iceberg native scans currently support non-AQE DPP only ([#3349], [#3511]);
extending broadcast reuse to AQE
-DPP for Iceberg is tracked at [#3510].
+Iceberg native scans support non-AQE DPP ([#3349], [#3511]) and, on Spark
3.5+, AQE DPP with broadcast reuse
+([#4215]); on Spark 3.4 Iceberg AQE DPP falls back to Spark without reuse.
[#3349]: https://github.com/apache/datafusion-comet/pull/3349
-[#3510]: https://github.com/apache/datafusion-comet/issues/3510
[#3511]: https://github.com/apache/datafusion-comet/pull/3511
[#4011]: https://github.com/apache/datafusion-comet/pull/4011
[#4112]: https://github.com/apache/datafusion-comet/pull/4112
+[#4215]: https://github.com/apache/datafusion-comet/pull/4215
## TPC-H and TPC-DS Performance
diff --git a/docs/source/user-guide/latest/iceberg.md
b/docs/source/user-guide/latest/iceberg.md
index 80859ba998..a8ea6f30ce 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/docs/source/user-guide/latest/iceberg.md
@@ -26,7 +26,7 @@ then serialized to Comet's native execution engine (see
[PR #2528](https://github.com/apache/datafusion-comet/pull/2528)).
The example below uses Spark's package downloader to retrieve Comet
$COMET_VERSION and Iceberg
-1.8.1, but Comet has been tested with Iceberg 1.5, 1.7, 1.8, 1.9, and 1.10.
The native Iceberg
+1.8.1, but Comet has been tested with Iceberg 1.5, 1.7, 1.8, 1.9, 1.10, and
1.11. The native Iceberg
reader is enabled by default. To disable it, set
`spark.comet.scan.icebergNative.enabled=false`.
The example uses the Spark 3.5 / Scala 2.12 build of Comet; substitute the
Comet artifact
@@ -63,7 +63,15 @@ The native Iceberg reader supports the following features:
**Table specifications:**
-- Iceberg table spec v1 and v2 (v3 will fall back to Spark)
+- Iceberg table spec v1, v2, and v3
+
+**Encryption:**
+
+- Encrypted v3 tables using 128-bit or 256-bit AES-GCM data keys (requires
Iceberg 1.11 or newer).
+ Iceberg-Java unwraps the key envelope on the driver during planning and
stores the plaintext data
+ key in each file's `key_metadata`, which the native reader uses directly, so
no KMS integration is
+ needed on the native side. Iceberg-Java also permits 192-bit data keys;
those tables fall back to
+ Spark (no AES-192-GCM in the underlying crypto).
**Schema and data types:**
@@ -156,17 +164,19 @@ These `s3.*` storage properties are not specific to the
Hive catalog shown here.
### Current limitations
-The following scenarios will fall back to Spark's native Iceberg reader:
+The following scenarios will fall back to the JVM Iceberg reader:
-- Iceberg table spec v3 scans
+- Iceberg table spec v4 or newer
+- v3 tables with columns that declare an initial default value
+- v3 column types the native reader cannot read (`variant`, `geometry`,
`geography`, `unknown`)
+- Encrypted tables with 192-bit data keys (no AES-192-GCM in the underlying
crypto)
+- Deletion vectors (v3 Puffin deletes); positional and equality deletes in
Parquet are supported
- Iceberg writes (reads are accelerated, writes use Spark)
- Tables backed by Avro or ORC data files (only Parquet is accelerated)
- Tables partitioned on `BINARY` or `DECIMAL` (with precision >28) columns
- Scans with residual filters using `truncate`, `bucket`, `year`, `month`,
`day`, or `hour`
transform functions (partition pruning still works, but row-level filtering
of these
transforms falls back)
-- Dynamic Partition Pruning under Adaptive Query Execution (non-AQE DPP is
supported);
- see [#3510](https://github.com/apache/datafusion-comet/issues/3510)
### Iceberg UDFs
diff --git a/native/Cargo.lock b/native/Cargo.lock
index 29f13df281..f4431f8551 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -232,9 +232,9 @@ checksum =
"d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56"
[[package]]
name = "arrow"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "378530e55cd479eda3c14eb345310799717e6f76d0c332041e8487022166b471"
+checksum = "6cfdd0833e32a9874d2b55089333ad310c0be208aafa277385ce2461dec90be3"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -253,9 +253,9 @@ dependencies = [
[[package]]
name = "arrow-arith"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "a0ab212d2c1886e802f51c5212d78ebbcbb0bec980fff9dadc1eb8d45cd0b738"
+checksum = "0a41203398f0eaa6f7ec8e62c0da742a21abf282c148fc157f6c35c90e29981a"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -267,9 +267,9 @@ dependencies = [
[[package]]
name = "arrow-array"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "cfd33d3e92f207444098c75b42de99d329562be0cf686b307b097cc52b4e999e"
+checksum = "ae33dad492b7df00a217563a7b0ef2874df68a0deea1b1a3acf628152f7f7a69"
dependencies = [
"ahash",
"arrow-buffer",
@@ -286,9 +286,9 @@ dependencies = [
[[package]]
name = "arrow-buffer"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "0c6cd424c2693bcdbc150d843dc9d4d137dd2de4782ce6df491ad11a3a0416c0"
+checksum = "b9552f96391c005e6ab449fa941420935e7e062489b12b8b1b08879b2163f5b5"
dependencies = [
"bytes",
"half",
@@ -298,9 +298,9 @@ dependencies = [
[[package]]
name = "arrow-cast"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "4c5aefb56a2c02e9e2b30746241058b85f8983f0fcff2ba0c6d09006e1cded7f"
+checksum = "3a8a327c9649f30d8406995f27642b68df354713cca3baaaf100f076f18d5f34"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -320,9 +320,9 @@ dependencies = [
[[package]]
name = "arrow-csv"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "e94e8cf7e517657a52b91ea1263acf38c4ca62a84655d72458a3359b12ab97de"
+checksum = "af0dd6d90d1955e9f9a014c1e563ee8aeffc21909085d25623e1da44d96eca26"
dependencies = [
"arrow-array",
"arrow-cast",
@@ -335,9 +335,9 @@ dependencies = [
[[package]]
name = "arrow-data"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "3c88210023a2bfee1896af366309a3028fc3bcbd6515fa29a7990ee1baa08ee0"
+checksum = "2b24852db04738907e06c04ea61e42fe7fda962a34513022dc0d0e754fb7976b"
dependencies = [
"arrow-buffer",
"arrow-schema",
@@ -348,9 +348,9 @@ dependencies = [
[[package]]
name = "arrow-ipc"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "238438f0834483703d88896db6fe5a7138b2230debc31b34c0336c2996e3c64f"
+checksum = "29a908a11fcfb3fb2f6730f4ac15e367bc644e419155e96238f68cf3adde572b"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -364,9 +364,9 @@ dependencies = [
[[package]]
name = "arrow-json"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "205ca2119e6d679d5c133c6f30e68f027738d95ed948cf77677ea69c7800036b"
+checksum = "b8a96aed3931c076adee39ec2a40d8219fc7f09e79bcdaca1df16272993e1e14"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -389,9 +389,9 @@ dependencies = [
[[package]]
name = "arrow-ord"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "1bffd8fd2579286a5d63bac898159873e5094a79009940bcb42bbfce4f19f1d0"
+checksum = "63a083ec750f5c043f02946b4baf05fcdbb55f4560a3277055caca5cc99f3eb0"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -402,9 +402,9 @@ dependencies = [
[[package]]
name = "arrow-row"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "bab5994731204603c73ba69267616c50f80780774c6bb0476f1f830625115e0c"
+checksum = "514ba0ef0d4c5896202dae736251ce415abb43a950bed570fb7981b8716c0e4c"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -415,9 +415,9 @@ dependencies = [
[[package]]
name = "arrow-schema"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "f633dbfdf39c039ada1bf9e34c694816eb71fbb7dc78f613993b7245e078a1ed"
+checksum = "21ca356ad6425cecb6eb7b28e4f659f1ee7880fbb1a16127de7dd62901efee9e"
dependencies = [
"bitflags 2.13.1",
"serde_core",
@@ -426,9 +426,9 @@ dependencies = [
[[package]]
name = "arrow-select"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "8cd065c54172ac787cf3f2f8d4107e0d3fdc26edba76fdf4f4cc170258942222"
+checksum = "c58da39eb3d8350ad4a549e5c2bc49284dac554016c69829310350f1731b0aad"
dependencies = [
"ahash",
"arrow-array",
@@ -440,9 +440,9 @@ dependencies = [
[[package]]
name = "arrow-string"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "29dd7cda3ab9692f43a2e4acc444d760cc17b12bb6d8232ddf64e9bab7c06b42"
+checksum = "b6789b388467525e3271326b6b4915666ecfdf5142aef09779445c954b67543c"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -3489,7 +3489,7 @@ dependencies = [
[[package]]
name = "iceberg"
version = "0.10.0"
-source =
"git+https://github.com/apache/iceberg-rust?rev=e8460ee#e8460eee8725c7f66d53e26b89b3fef578f90ce5";
+source =
"git+https://github.com/apache/iceberg-rust?rev=4532da8#4532da80f3930cdbda1cf2effe05b50165da9875";
dependencies = [
"aes-gcm",
"anyhow",
@@ -3545,7 +3545,7 @@ dependencies = [
[[package]]
name = "iceberg-storage-opendal"
version = "0.10.0"
-source =
"git+https://github.com/apache/iceberg-rust?rev=e8460ee#e8460eee8725c7f66d53e26b89b3fef578f90ce5";
+source =
"git+https://github.com/apache/iceberg-rust?rev=4532da8#4532da80f3930cdbda1cf2effe05b50165da9875";
dependencies = [
"anyhow",
"async-trait",
@@ -4786,9 +4786,9 @@ dependencies = [
[[package]]
name = "parquet"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "5dafa7d01085b62a47dd0c1829550a0a36710ea9c4fe358a05a85477cec8a908"
+checksum = "d298093b2dec60289dce0684c986d0f7679e9dd15771c2c65406e1aaf604a704"
dependencies = [
"ahash",
"arrow-array",
@@ -4826,9 +4826,9 @@ dependencies = [
[[package]]
name = "parquet-variant"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "74c8db065291f088a2aad8ab831853eae1871c0d311c8d0b83bbc3b7e735d0fc"
+checksum = "3fc70e87931167a4a3fde2ee923023a0624367691e3fd503476a1084dda5a054"
dependencies = [
"arrow",
"arrow-schema",
@@ -4842,9 +4842,9 @@ dependencies = [
[[package]]
name = "parquet-variant-compute"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "a530e8d5b5e14efcb39c9a6ec55432ad11f6afb7dc4455a79be0dc615fe3cc31"
+checksum = "823a9ecee8fd83a68f7165ef13acc840c4d7ed995838ba99d4714b20a2b5e780"
dependencies = [
"arrow",
"arrow-schema",
@@ -4859,9 +4859,9 @@ dependencies = [
[[package]]
name = "parquet-variant-json"
-version = "58.3.0"
+version = "58.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "00ed89908289f67caa2ca078f9ff9aacd6229a313ec92b12bf4f48f613dc2b97"
+checksum = "6f37a91177e2dddb10333546952fcf2b7674b97ebd9449432f24b883d6ea8108"
dependencies = [
"arrow-schema",
"base64",
diff --git a/native/Cargo.toml b/native/Cargo.toml
index 313a0b43be..fd7d7d3a66 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -34,10 +34,10 @@ edition = "2021"
rust-version = "1.88"
[workspace.dependencies]
-arrow = { version = "58.3.0", features = ["prettyprint", "ffi", "chrono-tz"] }
+arrow = { version = "58.4.0", features = ["prettyprint", "ffi", "chrono-tz"] }
async-trait = { version = "0.1" }
bytes = { version = "1.11.1" }
-parquet = { version = "58.3.0", default-features = false, features =
["experimental"] }
+parquet = { version = "58.4.0", default-features = false, features =
["experimental"] }
datafusion = { version = "54.1.0", default-features = false, features =
["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] }
datafusion-datasource = { version = "54.1.0" }
datafusion-physical-expr-adapter = { version = "54.1.0" }
@@ -58,8 +58,8 @@ object_store = { version = "0.13.2", features = ["gcp",
"azure", "aws", "http"]
url = "2.2"
aws-config = "1.8.18"
aws-credential-types = "1.2.13"
-iceberg = { git = "https://github.com/apache/iceberg-rust";, rev = "e8460ee" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust";,
rev = "e8460ee", features = ["opendal-memory", "opendal-fs", "opendal-s3",
"opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust";, rev = "4532da8" }
+iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust";,
rev = "4532da8", features = ["opendal-memory", "opendal-fs", "opendal-s3",
"opendal-gcs", "opendal-oss", "opendal-azdls"] }
reqsign-core = "3"
[profile.release]
diff --git a/native/core/src/execution/operators/iceberg_scan.rs
b/native/core/src/execution/operators/iceberg_scan.rs
index 7c95c6a53b..d77b208964 100644
--- a/native/core/src/execution/operators/iceberg_scan.rs
+++ b/native/core/src/execution/operators/iceberg_scan.rs
@@ -633,6 +633,7 @@ mod tests {
use std::io::Write;
use std::sync::Arc;
+ use iceberg::encryption::StandardKeyMetadata;
use iceberg::io::{FileIO, FileIOBuilder};
use iceberg::scan::{FileScanTask, FileScanTaskDeleteFile};
use iceberg::spec::{DataContentType, DataFileFormat, Schema};
@@ -738,4 +739,60 @@ mod tests {
.await
.unwrap();
}
+
+ fn from_hex(s: &str) -> Vec<u8> {
+ assert!(s.len().is_multiple_of(2), "odd-length hex string");
+ (0..s.len())
+ .step_by(2)
+ .map(|i| u8::from_str_radix(&s[i..i + 2], 16).expect("invalid
hex"))
+ .collect()
+ }
+
+ // iceberg-rust encodes and decodes its own StandardKeyMetadata
identically. Confirms the API
+ // is present after the rev bump and the [version byte][Avro datum] format
round-trips.
+ #[test]
+ fn standard_key_metadata_roundtrips() {
+ let key: Vec<u8> = (0u8..16).collect();
+ let aad: &[u8] = b"comet-aad-prefix";
+ let km = StandardKeyMetadata::try_new(&key)
+ .unwrap()
+ .with_aad_prefix(aad);
+
+ let encoded = km.encode().unwrap();
+ assert_eq!(encoded[0], 0x01, "expected StandardKeyMetadata V1 marker");
+
+ let decoded = StandardKeyMetadata::decode(&encoded).unwrap();
+ assert_eq!(decoded.encryption_key().as_bytes(), key.as_slice());
+ assert_eq!(decoded.aad_prefix(), Some(aad));
+ }
+
+ // Cross-language gate for the encrypted-read passthrough plan:
iceberg-rust must decode the
+ // exact StandardKeyMetadata bytes that Iceberg-Java writes into a data
file's key_metadata,
+ // because Comet forwards those bytes verbatim (no re-encoding, no KMS)
into
+ // FileScanTask::key_metadata. This fixture is a real Iceberg-Java blob
produced by
+ // StandardEncryptionManager. To regenerate (e.g. after a wire-format
change), print the bytes
+ // from CometIcebergEncryptionSuite's plaintext-DEK test and paste them
here.
+ const JAVA_KEY_METADATA_HEX: &str =
+
"012084f49fba77f8ff1da0c115d1e46563cc0220f1d31d62b68808b469eb99fe9c57096000";
+ const JAVA_DEK_HEX: &str = "84f49fba77f8ff1da0c115d1e46563cc";
+
+ #[test]
+ fn decodes_java_produced_key_metadata() {
+ let blob = from_hex(JAVA_KEY_METADATA_HEX);
+ let expected_dek = from_hex(JAVA_DEK_HEX);
+
+ let decoded = StandardKeyMetadata::decode(&blob)
+ .expect("iceberg-rust failed to decode a Java-produced
StandardKeyMetadata blob");
+ assert_eq!(
+ decoded.encryption_key().as_bytes(),
+ expected_dek.as_slice(),
+ "DEK recovered by Rust differs from the Java plaintext DEK"
+ );
+ // The Java blob carries a 16-byte AAD prefix; confirm the
optional-field union decodes too.
+ assert_eq!(
+ decoded.aad_prefix().map(|a| a.len()),
+ Some(16),
+ "expected a 16-byte AAD prefix from the Java blob"
+ );
+ }
}
diff --git a/native/core/src/execution/planner.rs
b/native/core/src/execution/planner.rs
index e43daed9a5..a4aedfc991 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -3706,7 +3706,9 @@ fn parse_file_scan_tasks_from_common(
} else {
Some(del.equality_ids.clone())
},
- key_metadata: None,
+ // Plaintext StandardKeyMetadata forwarded verbatim from the
JVM; decoded by
+ // iceberg-rust with no KMS unwrap. None for unencrypted
delete files.
+ key_metadata:
del.key_metadata.clone().map(Vec::into_boxed_slice),
})
})
.collect::<Result<Vec<_>, ExecutionError>>()?;
@@ -3849,7 +3851,9 @@ fn parse_file_scan_tasks_from_common(
partition_spec,
name_mapping,
case_sensitive: false,
- key_metadata: None,
+ // Plaintext StandardKeyMetadata forwarded verbatim from the
JVM; decoded by
+ // iceberg-rust with no KMS unwrap. None for unencrypted data
files.
+ key_metadata:
proto_task.key_metadata.clone().map(Vec::into_boxed_slice),
})
})
.collect();
diff --git a/native/proto/src/proto/operator.proto
b/native/proto/src/proto/operator.proto
index 439c9613a4..5493d921b8 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -318,6 +318,11 @@ message IcebergFileScanTask {
// Total file size from the manifest entry, used to skip stat/HEAD calls
uint64 file_size_in_bytes = 5;
+ // Serialized StandardKeyMetadata (plaintext DEK + AAD prefix) for an
encrypted data file, taken
+ // verbatim from Iceberg's DataFile.keyMetadata(). Absent for unencrypted
tables. iceberg-rust
+ // decodes this directly; no KMS unwrap is needed on the native side.
+ optional bytes key_metadata = 6;
+
// Indices into IcebergScan deduplication pools
uint32 schema_idx = 15;
optional uint32 partition_type_idx = 16;
@@ -344,6 +349,15 @@ message IcebergDeleteFile {
// Equality field IDs (empty for positional deletes)
repeated int32 equality_ids = 4;
+
+ // Fields 5-7 are reserved for the deletion-vector coordinates on the
dv-read branch
+ // (referenced_data_file / content_offset / content_size_in_bytes) so the
two features can merge
+ // in either order without a tag collision.
+ reserved 5, 6, 7;
+
+ // Serialized StandardKeyMetadata for an encrypted delete file, from
DeleteFile.keyMetadata().
+ // Absent for unencrypted tables. Decoded directly by iceberg-rust; no KMS
unwrap needed.
+ optional bytes key_metadata = 8;
}
message Projection {
diff --git a/spark/pom.xml b/spark/pom.xml
index cc3f278ee3..fbe059c739 100644
--- a/spark/pom.xml
+++ b/spark/pom.xml
@@ -107,6 +107,13 @@ under the License.
<artifactId>spark-hadoop-cloud_${scala.binary.version}</artifactId>
<classifier>tests</classifier>
</dependency>
+ <dependency>
+ <!-- Test-only: provides an embedded Hive metastore so Iceberg's
type=hive catalog can
+ create encrypted tables (only Hive wires the KMS-backed
EncryptionManager). Version and
+ scope come from the parent pom's dependencyManagement. -->
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-hive_${scala.binary.version}</artifactId>
+ </dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
diff --git
a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index 1a5cea7c5c..72e885e834 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -51,6 +51,7 @@ object IcebergReflection extends Logging {
val UNBOUND_PREDICATE = "org.apache.iceberg.expressions.UnboundPredicate"
val SPARK_BATCH_QUERY_SCAN =
"org.apache.iceberg.spark.source.SparkBatchQueryScan"
val SPARK_STAGED_SCAN = "org.apache.iceberg.spark.source.SparkStagedScan"
+ val SPARK_SCHEMA_UTIL = "org.apache.iceberg.spark.SparkSchemaUtil"
}
/**
@@ -771,6 +772,68 @@ object IcebergReflection extends Logging {
unsupportedTypes.toList
}
+
+ /**
+ * Returns the names of schema columns (including nested struct/list/map
fields) that declare a
+ * V3 initial-default value. iceberg-rust does not synthesize default values
for columns absent
+ * from a data file, so reads projecting such columns must fall back. Throws
on reflection
+ * failure so the caller can fall back rather than risk a native crash.
+ */
+ def columnsWithInitialDefault(schema: Any): List[String] = {
+ import scala.jdk.CollectionConverters._
+ val columns =
+
schema.getClass.getMethod("columns").invoke(schema).asInstanceOf[java.util.List[_]]
+ columns.asScala.flatMap(walkFieldForDefault).toList
+ }
+
+ private def walkFieldForDefault(field: Any): List[String] = {
+ import scala.jdk.CollectionConverters._
+ val name =
field.getClass.getMethod("name").invoke(field).asInstanceOf[String]
+ val here =
+ if (field.getClass.getMethod("initialDefault").invoke(field) != null)
List(name) else Nil
+ val fieldType = field.getClass.getMethod("type").invoke(field)
+ val nested =
+ if
(fieldType.getClass.getMethod("isNestedType").invoke(fieldType).asInstanceOf[Boolean])
{
+ val nestedType =
fieldType.getClass.getMethod("asNestedType").invoke(fieldType)
+ val fields =
+ nestedType.getClass
+ .getMethod("fields")
+ .invoke(nestedType)
+ .asInstanceOf[java.util.List[_]]
+ fields.asScala.flatMap(walkFieldForDefault).toList
+ } else {
+ Nil
+ }
+ here ++ nested
+ }
+
+ /**
+ * Converts an Iceberg `Schema` to the Spark `StructType` it reads as, via
+ * `SparkSchemaUtil.convert`. Comet serializes the whole table/scan schema
to native (not just
+ * projected columns), so callers use this to run the schema through Comet's
existing type
+ * allow-list and fall back if any column is a type the native reader does
not support (e.g.
+ * variant). Throws on reflection failure so the caller can fall back.
+ */
+ def toSparkSchema(schema: Any): org.apache.spark.sql.types.StructType = {
+ val sparkSchemaUtil = loadClass(ClassNames.SPARK_SCHEMA_UTIL)
+ val schemaClass = loadClass(ClassNames.SCHEMA)
+ val convert = sparkSchemaUtil.getMethod("convert", schemaClass)
+ convert
+ .invoke(null, schema.asInstanceOf[AnyRef])
+ .asInstanceOf[org.apache.spark.sql.types.StructType]
+ }
+
+ /**
+ * The configured AES data-key length in bytes for an encrypted table
(Iceberg's
+ * `encryption.data-key-length`, default 16), or None if the table is not
encrypted. Comet's
+ * native Parquet reader supports 128-bit (16-byte) and 256-bit (32-byte)
keys but not 192-bit
+ * (the underlying crypto has no AES-192-GCM), so callers fall back for
anything else. Throws on
+ * reflection failure so the caller can fall back.
+ */
+ def encryptionDataKeyLength(table: Any): Option[Int] =
+ getTableProperties(table).filter(_.containsKey("encryption.key-id")).map {
props =>
+
Option(props.get("encryption.data-key-length")).map(_.toInt).getOrElse(16)
+ }
}
/**
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index 59331129d5..5e451b382a 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -111,7 +111,12 @@ case class CometScanRule(session: SparkSession)
"all_entries",
"all_manifests")
- metadataTableSuffix.exists(suffix =>
scanExec.table.name().endsWith(suffix))
+ // Match case-insensitively: metadata tables surface lowercase via the
path form
+ // (...metadata.json#all_manifests) but uppercase via the
catalog-identifier form
+ // (db.table.ALL_DATA_FILES), and the latter must hit this gate too
rather than fall through
+ // to reflection that fails on the metadata-table class.
+ val name = scanExec.table.name().toLowerCase(Locale.ROOT)
+ metadataTableSuffix.exists(name.endsWith)
}
val fullPlan = plan
@@ -462,12 +467,18 @@ case class CometScanRule(session: SparkSession)
// Check Iceberg table format version
- val formatVersionSupported =
IcebergReflection.getFormatVersion(metadata.table) match {
- case Some(formatVersion) =>
- if (formatVersion > 2) {
+ // V3 adds column types iceberg-rust cannot read (variant, geometry,
geography, unknown)
+ // and column default values; those are handled by the allow-list and
default-value checks
+ // below, which fall back per-table. This gate only bounds the format
version. Deletion
+ // vectors (a V3 delete feature) are handled by the delete-file gate,
which still falls back
+ // for non-Parquet (Puffin) deletes since native deletion-vector reads
are not yet wired.
+ val formatVersion = IcebergReflection.getFormatVersion(metadata.table)
+ val formatVersionSupported = formatVersion match {
+ case Some(v) =>
+ if (v > 3) {
fallbackReasons += "Iceberg table format version " +
- s"$formatVersion is not supported. " +
- "Comet only supports Iceberg table format V1 and V2"
+ s"$v is not supported. " +
+ "Comet supports Iceberg table format V1, V2, and V3"
false
} else {
true
@@ -477,6 +488,68 @@ case class CometScanRule(session: SparkSession)
false
}
+ // Column default values are a V3 feature, so V1/V2 tables cannot have
them and need no
+ // check. Only inspect V3 tables. iceberg-rust does not synthesize a
V3 initial-default for
+ // a column absent from a data file, so a scan projecting such a
column must fall back. A V3
+ // table implies an Iceberg version new enough to expose
initialDefault(), so a reflection
+ // failure here is unexpected and also falls back rather than risk a
crash.
+ val defaultValuesSupported =
+ if (!formatVersion.exists(_ >= 3)) {
+ true
+ } else {
+ try {
+ val defaulted =
IcebergReflection.columnsWithInitialDefault(metadata.scanSchema)
+ if (defaulted.nonEmpty) {
+ fallbackReasons += "Iceberg column(s) with V3 default values
are not yet " +
+ s"supported by Comet's native reader:
${defaulted.mkString(", ")}"
+ false
+ } else {
+ true
+ }
+ } catch {
+ case e: Exception =>
+ fallbackReasons += "Iceberg reflection failure: could not
verify V3 default " +
+ s"values: ${e.getMessage}"
+ false
+ }
+ }
+
+ // Comet serializes the whole table/scan schema to native, not just
projected columns, so a
+ // type the native reader does not support (e.g. variant) breaks the
scan even when that
+ // column is not projected. The readSchema allow-list only covers
projected columns, so run
+ // the same allow-list over the full schema Comet may serialize.
Reflection failure also
+ // falls back.
+ val schemaTypesSupported =
+ try {
+ val fullSchema =
IcebergReflection.toSparkSchema(metadata.tableSchema)
+ typeChecker.isSchemaSupported(fullSchema, fallbackReasons)
+ } catch {
+ case e: Exception =>
+ fallbackReasons += "Iceberg reflection failure: could not verify
column " +
+ s"types: ${e.getMessage}"
+ false
+ }
+
+ // The native Parquet reader decrypts 128-bit and 256-bit AES-GCM data
keys (16- or
+ // 32-byte). 192-bit is unsupported because the underlying crypto has
no AES-192-GCM. Fall
+ // back for anything else. None means the table is unencrypted.
Reflection failure also
+ // falls back.
+ val encryptionKeyLengthSupported =
+ try {
+ IcebergReflection.encryptionDataKeyLength(metadata.table) match {
+ case Some(len) if len != 16 && len != 32 =>
+ fallbackReasons += s"Iceberg table encryption with a ${len *
8}-bit data key is " +
+ "not yet supported by Comet's native reader (only 128-bit
and 256-bit AES-GCM)"
+ false
+ case _ => true
+ }
+ } catch {
+ case e: Exception =>
+ fallbackReasons += "Iceberg reflection failure: could not verify
encryption key " +
+ s"length: ${e.getMessage}"
+ false
+ }
+
// Single-pass validation of all FileScanTasks
val taskValidation =
try {
@@ -713,6 +786,7 @@ case class CometScanRule(session: SparkSession)
}
if (schemaSupported && fileIOCompatible && formatVersionSupported &&
+ defaultValuesSupported && schemaTypesSupported &&
encryptionKeyLengthSupported &&
taskValidation.allParquet && allSupportedFilesystems &&
partitionTypesSupported &&
complexTypePredicatesSupported && transformFunctionsSupported &&
deleteFileTypesSupported && dppSubqueriesSupported) {
diff --git
a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
index 247a2f9245..c2b792c36c 100644
---
a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
+++
b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
@@ -231,6 +231,22 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
builder.build()
}
+ /**
+ * An encrypted ContentFile's plaintext StandardKeyMetadata blob as a
protobuf ByteString, or
+ * None for an unencrypted file (keyMetadata() returns null). `method` is
the cached
+ * ContentFile.keyMetadata() accessor. A reflection failure here is
intentionally not caught: by
+ * serde time the plan is already committed to the native scan, so there is
no fallback, and
+ * silently dropping the key of an encrypted file would fail obscurely in
the native reader.
+ */
+ private def keyMetadataBytes(
+ method: java.lang.reflect.Method,
+ contentFile: Any): Option[com.google.protobuf.ByteString] =
+ method.invoke(contentFile) match {
+ case buf: java.nio.ByteBuffer if buf.remaining() > 0 =>
+ Some(com.google.protobuf.ByteString.copyFrom(buf.duplicate()))
+ case _ => None
+ }
+
/**
* Extracts delete files from an Iceberg FileScanTask as a list (for
deduplication).
*
@@ -244,6 +260,8 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
fileScanTaskClass: Class[_]): Seq[OperatorOuterClass.IcebergDeleteFile]
= {
try {
val deleteFileClass =
IcebergReflection.loadClass(IcebergReflection.ClassNames.DELETE_FILE)
+ // keyMetadata() is declared on ContentFile; present across all
supported Iceberg versions.
+ val keyMetadataMethod = contentFileClass.getMethod("keyMetadata")
val deletes = IcebergReflection.getDeleteFilesFromTask(task,
fileScanTaskClass)
@@ -295,6 +313,10 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
case _: Exception =>
}
+ // Encrypted delete files carry a plaintext StandardKeyMetadata blob;
forward it verbatim.
+ // Unencrypted delete files leave the field unset.
+ keyMetadataBytes(keyMetadataMethod,
deleteFile).foreach(deleteBuilder.setKeyMetadata)
+
deleteBuilder.build()
}.toSeq
} catch {
@@ -884,6 +906,9 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
val lengthMethod = contentScanTaskClass.getMethod("length")
val residualMethod = contentScanTaskClass.getMethod("residual")
val fileSizeInBytesMethod = contentFileClass.getMethod("fileSizeInBytes")
+ // keyMetadata() is declared on ContentFile (present across all supported
Iceberg versions).
+ // For encrypted tables it returns the plaintext StandardKeyMetadata blob;
null otherwise.
+ val keyMetadataMethod = contentFileClass.getMethod("keyMetadata")
val taskSchemaMethod = fileScanTaskClass.getMethod("schema")
val toJsonMethod = schemaParserClass.getMethod("toJson", schemaClass)
toJsonMethod.setAccessible(true)
@@ -941,6 +966,10 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
fileSizeInBytesMethod.invoke(dataFile).asInstanceOf[Long]
taskBuilder.setFileSizeInBytes(fileSizeInBytes)
+ // Encrypted data files carry a plaintext StandardKeyMetadata
blob; forward it
+ // verbatim for iceberg-rust to decode. Unencrypted files
leave the field unset.
+ keyMetadataBytes(keyMetadataMethod,
dataFile).foreach(taskBuilder.setKeyMetadata)
+
val taskSchema = taskSchemaMethod.invoke(task)
val deletes =
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala
b/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala
index 33de4f3705..07ac91d90c 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala
@@ -72,7 +72,7 @@ trait CometIcebergTestBase {
finally deleteRecursively(dir)
}
- private def deleteRecursively(file: File): Unit = {
+ protected def deleteRecursively(file: File): Unit = {
if (file.isDirectory) file.listFiles().foreach(deleteRecursively)
file.delete()
}
diff --git
a/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
new file mode 100644
index 0000000000..1b221bf454
--- /dev/null
+++
b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet
+
+import java.io.File
+import java.nio.ByteBuffer
+import java.util.Collections
+
+import org.apache.iceberg.Files
+import org.apache.iceberg.encryption.{EncryptedKey,
NativeEncryptionKeyMetadata, StandardEncryptionManager}
+import org.apache.iceberg.util.ByteBuffers
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.CometTestBase
+import org.apache.spark.sql.comet.CometIcebergNativeScanExec
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+
+import org.apache.comet.iceberg.CometTestKMS
+
+/**
+ * Encryption-specific Iceberg tests. Lives in the spark-4.1 source set
because Iceberg table
+ * encryption is a V3 feature and the encryption APIs (KeyManagementClient,
StandardEncryption
+ * manager) are only public from Iceberg 1.11, which Comet pairs with Spark
4.1.
+ */
+class CometIcebergEncryptionSuite
+ extends CometTestBase
+ with CometIcebergTestBase
+ with AdaptiveSparkPlanHelper {
+
+ // A Hive catalog is required for an actual encrypted table: it is the only
Iceberg catalog that
+ // wires the KMS-backed EncryptionManager in 1.11 (Hadoop/REST write
plaintext). Point Iceberg's
+ // type=hive catalog at an in-memory Derby metastore so no external HMS is
needed. Set at session
+ // creation because Iceberg builds its HiveConf from the Hadoop
configuration, which reflects
+ // spark.hadoop.* only at context startup, not from runtime withSQLConf.
+ //
+ // lazy so the temp dir is created at session init, not at construction: CI
runs tests with
+ // -DwildcardSuites, which puts ScalaTest in discovery mode and instantiates
every suite before
+ // any test executes. createTempDirectory does not create its parent
(java.io.tmpdir, pinned to
+ // target/tmp by the pom), which does not exist yet on a fresh checkout, so
an eager val aborts
+ // discovery with NoSuchFileException. createDirectories makes the parent up
front.
+ private lazy val hiveWarehouse = {
+ val base = java.nio.file.Paths.get(System.getProperty("java.io.tmpdir"))
+ java.nio.file.Files.createDirectories(base)
+ val dir = java.nio.file.Files.createTempDirectory(base,
"comet-hive-warehouse").toFile
+ createdWarehouse = dir
+ dir
+ }
+
+ // Captured on init rather than read from the lazy val so a run that never
touched it (all tests
+ // skipped) creates nothing to delete.
+ private var createdWarehouse: File = _
+
+ override protected def afterAll(): Unit = {
+ try if (createdWarehouse != null) deleteRecursively(createdWarehouse)
+ finally super.afterAll()
+ }
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set(
+ "spark.hadoop.javax.jdo.option.ConnectionURL",
+ "jdbc:derby:memory:comet_iceberg_encryption;create=true")
+ .set("spark.hadoop.datanucleus.schema.autoCreateAll", "true")
+ .set("spark.hadoop.hive.metastore.schema.verification", "false")
+ // Hive's default metastore connection pool is BoneCP, which Spark's
Hive fork does not ship
+ // (NoClassDefFoundError: com/jolbox/bonecp/BoneCPConfig). This config
drives both the
+ // DataNucleus ObjectStore and the TxnHandler pool, so DBCP steers the
embedded metastore off
+ // BoneCP. Mirrors Iceberg's TestHiveMetastore.initConf.
+ .set("spark.hadoop.datanucleus.connectionPoolingType", "DBCP")
+ .set("spark.hadoop.hive.in.test", "true")
+ // Skip Iceberg's HMS table lock. The embedded metastore's ACID
TxnHandler rejects Iceberg's
+ // lock request ("Bug: operationType=UNSET"); disabling it uses the
lock-free commit path
+ // (atomic alter-table with an expected-parameter check), which is all
this single-writer test
+ // needs.
+ .set("spark.hadoop.iceberg.engine.hive.lock-enabled", "false")
+ .set("spark.hadoop.hive.metastore.warehouse.dir",
hiveWarehouse.toURI.toString)
+ .set("spark.sql.catalog.hive_cat",
"org.apache.iceberg.spark.SparkCatalog")
+ .set("spark.sql.catalog.hive_cat.type", "hive")
+ .set("spark.sql.catalog.hive_cat.warehouse",
hiveWarehouse.toURI.toString)
+ .set("spark.sql.catalog.hive_cat.encryption.kms-impl",
classOf[CometTestKMS].getName)
+ }
+
+ // End-to-end: an encrypted V3 table is read through Comet's native Iceberg
scan and matches
+ // Spark. This is the guard that the encrypted path stays native; Iceberg's
own TestTableEncryption
+ // (run under Comet in CI) only asserts correctness, not that the native
scan was used.
+ test("encrypted V3 table is read natively and matches Spark") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ assume(icebergVersionAtLeast(1, 11), "Iceberg table encryption requires
Iceberg 1.11+")
+
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "hive_cat.db.encrypted"
+ // The Hive catalog does not auto-create namespaces the way the Hadoop
catalog does.
+ spark.sql("CREATE NAMESPACE IF NOT EXISTS hive_cat.db")
+ spark.sql(s"DROP TABLE IF EXISTS $table")
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, name STRING, value DOUBLE) USING iceberg
+ TBLPROPERTIES ('format-version' = '3', 'encryption.key-id' =
'${CometTestKMS.MasterKeyId}')
+ """)
+ spark.sql(s"""
+ INSERT INTO $table VALUES (1, 'Alice', 10.5), (2, 'Bob', 20.3), (3,
'Charlie', 30.7)
+ """)
+
+ // Guard against a silently-plaintext table (e.g. catalog not wiring
encryption): a genuinely
+ // encrypted table records a plaintext DEK blob per data file in
key_metadata.
+ val keyMetadatas = spark
+ .sql(s"SELECT key_metadata FROM $table.files WHERE content = 0")
+ .collect()
+ .flatMap(r => Option(r.getAs[Array[Byte]]("key_metadata")))
+ assert(
+ keyMetadatas.nonEmpty,
+ "expected encryption to populate key_metadata; the Hive catalog may
not have wired the KMS")
+
+ val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $table ORDER BY
id")
+ assert(
+ collect(cometPlan) { case s: CometIcebergNativeScanExec => s
}.nonEmpty,
+ "expected the encrypted read to run natively but found no
CometIcebergNativeScanExec:\n" +
+ cometPlan)
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+
+ // Confirms the assumption that native encrypted reads rely on: an Iceberg
data file's
+ // key_metadata carries a PLAINTEXT data encryption key, not a KMS-wrapped
one. Iceberg mints a
+ // fresh DEK per file in StandardEncryptionManager.encrypt() and records it
plaintext in
+ // key_metadata (the manifest that holds it is itself encrypted). The KMS
only wraps the KEK /
+ // manifest-list keys, never the per-file DEK. Runs without a catalog.
+ test("data file key_metadata carries a plaintext DEK that round-trips") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ assume(icebergVersionAtLeast(1, 11), "Iceberg table encryption requires
Iceberg 1.11+")
+
+ val dataKeyLength = 16
+ val mgr = new StandardEncryptionManager(
+ Collections.emptyList[EncryptedKey](),
+ CometTestKMS.MasterKeyId,
+ dataKeyLength,
+ new CometTestKMS)
+
+ withTempDir { dir =>
+ // encrypt() lazily generates the per-file DEK + AAD prefix; no bytes
are written.
+ val keyMetadata =
+ mgr.encrypt(Files.localOutput(new java.io.File(dir,
"data.parquet"))).keyMetadata()
+
+ val dek = keyMetadata.encryptionKey()
+ assert(
+ dek.remaining() == dataKeyLength,
+ s"expected a plaintext $dataKeyLength-byte DEK but got
${dek.remaining()} bytes")
+ val aad = keyMetadata.aadPrefix()
+ assert(aad != null && aad.remaining() > 0, "expected a non-empty AAD
prefix")
+
+ // buffer() is the serialized StandardKeyMetadata blob stored in the
manifest's key_metadata
+ // field, and the exact bytes Comet would forward to iceberg-rust. It
must start with the V1
+ // marker and round-trip back to the same DEK. StandardKeyMetadata.parse
is package-private,
+ // so reach it reflectively, but read the result through the public
NativeEncryptionKeyMetadata
+ // interface (the concrete class is not accessible). iceberg-rust
decodes this same format.
+ val blob = ByteBuffers.toByteArray(keyMetadata.buffer())
+ assert(blob(0) == 1, s"unexpected key_metadata version byte: ${blob(0)}")
+
+ val kmClass =
Class.forName("org.apache.iceberg.encryption.StandardKeyMetadata")
+ val parse = kmClass.getDeclaredMethod("parse", classOf[ByteBuffer])
+ parse.setAccessible(true)
+ val parsed =
+ parse.invoke(null,
ByteBuffer.wrap(blob)).asInstanceOf[NativeEncryptionKeyMetadata]
+ val roundTripped = ByteBuffers.toByteArray(parsed.encryptionKey())
+
+ assert(
+ java.util.Arrays.equals(ByteBuffers.toByteArray(dek), roundTripped),
+ "DEK did not survive the StandardKeyMetadata encode/parse round-trip")
+ }
+ }
+
+ // A 256-bit (32-byte) AES-GCM data key is supported natively, so the read
must run through Comet
+ // and match Spark.
+ test("encrypted V3 table with a 256-bit data key is read natively and
matches Spark") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ assume(icebergVersionAtLeast(1, 11), "Iceberg table encryption requires
Iceberg 1.11+")
+
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "hive_cat.db.encrypted_256"
+ spark.sql("CREATE NAMESPACE IF NOT EXISTS hive_cat.db")
+ spark.sql(s"DROP TABLE IF EXISTS $table")
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, name STRING) USING iceberg
+ TBLPROPERTIES ('format-version' = '3', 'encryption.key-id' =
'${CometTestKMS.MasterKeyId}',
+ 'encryption.data-key-length' = '32')
+ """)
+ spark.sql(s"INSERT INTO $table VALUES (1, 'Alice'), (2, 'Bob')")
+
+ val keyMetadatas = spark
+ .sql(s"SELECT key_metadata FROM $table.files WHERE content = 0")
+ .collect()
+ .flatMap(r => Option(r.getAs[Array[Byte]]("key_metadata")))
+ assert(
+ keyMetadatas.nonEmpty,
+ "expected encryption to populate key_metadata; the Hive catalog may
not have wired the KMS")
+
+ val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $table ORDER BY
id")
+ assert(
+ collect(cometPlan) { case s: CometIcebergNativeScanExec => s
}.nonEmpty,
+ "expected the 256-bit encrypted read to run natively but found no " +
+ s"CometIcebergNativeScanExec:\n$cometPlan")
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+
+ // A 192-bit data key is unsupported (no AES-192-GCM in the underlying
crypto), so the table must
+ // fall back to Spark while still returning correct results.
+ test("encrypted V3 table with a 192-bit data key falls back to Spark") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ assume(icebergVersionAtLeast(1, 11), "Iceberg table encryption requires
Iceberg 1.11+")
+
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "hive_cat.db.encrypted_192"
+ spark.sql("CREATE NAMESPACE IF NOT EXISTS hive_cat.db")
+ spark.sql(s"DROP TABLE IF EXISTS $table")
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, name STRING) USING iceberg
+ TBLPROPERTIES ('format-version' = '3', 'encryption.key-id' =
'${CometTestKMS.MasterKeyId}',
+ 'encryption.data-key-length' = '24')
+ """)
+ spark.sql(s"INSERT INTO $table VALUES (1, 'Alice'), (2, 'Bob')")
+
+ val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $table ORDER BY
id")
+ assert(
+ collect(cometPlan) { case s: CometIcebergNativeScanExec => s }.isEmpty,
+ "expected a 192-bit encrypted table to fall back to Spark but it ran
natively:\n" +
+ cometPlan)
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+}
diff --git
a/spark/src/test/spark-4.1/org/apache/comet/iceberg/CometTestKMS.scala
b/spark/src/test/spark-4.1/org/apache/comet/iceberg/CometTestKMS.scala
new file mode 100644
index 0000000000..b9317af6b2
--- /dev/null
+++ b/spark/src/test/spark-4.1/org/apache/comet/iceberg/CometTestKMS.scala
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.iceberg
+
+import java.nio.ByteBuffer
+import java.util
+
+import org.apache.iceberg.encryption.Ciphers
+import org.apache.iceberg.encryption.KeyManagementClient
+import org.apache.iceberg.util.ByteBuffers
+
+/**
+ * In-memory KMS for exercising Iceberg table encryption in tests. Iceberg
instantiates this by
+ * class name via the `encryption.kms-impl` catalog property, so it needs a
public no-arg
+ * constructor. Wrapping/unwrapping mirrors Iceberg's own MemoryMockKMS
(AES-GCM under a fixed
+ * master key); it is self-consistent, which is all Iceberg requires. Not for
production use.
+ */
+class CometTestKMS extends KeyManagementClient {
+
+ override def wrapKey(key: ByteBuffer, wrappingKeyId: String): ByteBuffer = {
+ val masterKey = CometTestKMS.masterKey(wrappingKeyId)
+ val encryptor = new Ciphers.AesGcmEncryptor(masterKey)
+ ByteBuffer.wrap(encryptor.encrypt(ByteBuffers.toByteArray(key), null))
+ }
+
+ override def unwrapKey(wrappedKey: ByteBuffer, wrappingKeyId: String):
ByteBuffer = {
+ val masterKey = CometTestKMS.masterKey(wrappingKeyId)
+ val decryptor = new Ciphers.AesGcmDecryptor(masterKey)
+ ByteBuffer.wrap(decryptor.decrypt(ByteBuffers.toByteArray(wrappedKey),
null))
+ }
+
+ override def initialize(properties: util.Map[String, String]): Unit = {}
+}
+
+object CometTestKMS {
+ // 16-byte master key; keyed by the id referenced in the table's
`encryption.key-id` property.
+ val MasterKeyId = "keyA"
+ private val MasterKey = "0123456789012345".getBytes("UTF-8")
+
+ private def masterKey(wrappingKeyId: String): Array[Byte] = {
+ if (wrappingKeyId != MasterKeyId) {
+ throw new RuntimeException(s"Unknown wrapping key id: $wrappingKeyId")
+ }
+ MasterKey
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
