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 { + 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 = (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::, 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. spark-hadoop-cloud_${scala.binary.version} tests + + + org.apache.spark + spark-hive_${scala.binary.version} + junit junit 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 + } +}