From 5cde1d445941e28ea5501f368e64793f82ba3fcc Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 10:40:33 -0400
Subject: [PATCH 01/14] preliminary iceberg decryption support: bump
 iceberg-rest dep, feed key data through proto

---
 dev/diffs/iceberg/1.11.0.diff                 |  25 +++++
 native/Cargo.lock                             |   4 +-
 native/Cargo.toml                             |   4 +-
 .../src/execution/operators/iceberg_scan.rs   |  60 ++++++++++
 native/core/src/execution/planner.rs          |   6 +
 native/proto/src/proto/operator.proto         |  12 ++
 .../comet/iceberg/IcebergReflection.scala     |  51 +++++++++
 .../apache/comet/rules/CometScanRule.scala    |  59 +++++++++-
 .../operator/CometIcebergNativeScan.scala     |  25 +++++
 .../comet/CometIcebergEncryptionSuite.scala   | 105 ++++++++++++++++++
 .../apache/comet/iceberg/CometTestKMS.scala   |  69 ++++++++++++
 11 files changed, 411 insertions(+), 9 deletions(-)
 create mode 100644 spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
 create mode 100644 spark/src/test/spark-4.1/org/apache/comet/iceberg/CometTestKMS.scala

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/native/Cargo.lock b/native/Cargo.lock
index fc04499b51..b2127f8bb2 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -3487,7 +3487,7 @@ dependencies = [
 [[package]]
 name = "iceberg"
 version = "0.10.0"
-source = "git+https://github.com/apache/iceberg-rust?rev=ff678ca#ff678ca6d1123d9d82902b34379ccba2caf172b6"
+source = "git+https://github.com/apache/iceberg-rust?rev=3f163f25#3f163f25d5a0afeb8433481780da624f3103db89"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -3543,7 +3543,7 @@ dependencies = [
 [[package]]
 name = "iceberg-storage-opendal"
 version = "0.10.0"
-source = "git+https://github.com/apache/iceberg-rust?rev=ff678ca#ff678ca6d1123d9d82902b34379ccba2caf172b6"
+source = "git+https://github.com/apache/iceberg-rust?rev=3f163f25#3f163f25d5a0afeb8433481780da624f3103db89"
 dependencies = [
  "anyhow",
  "async-trait",
diff --git a/native/Cargo.toml b/native/Cargo.toml
index 7e52ec0282..2b05087876 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -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 = "ff678ca" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "ff678ca", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "3f163f25" }
+iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "3f163f25", 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 d89a25c879..9db4d2948a 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};
@@ -660,6 +661,7 @@ mod tests {
             partition_spec: None,
             name_mapping: None,
             case_sensitive: false,
+            key_metadata: None,
         }
     }
 
@@ -670,6 +672,7 @@ mod tests {
             file_size_in_bytes: 0,
             partition_spec_id: 0,
             equality_ids: None,
+            key_metadata: None,
         }
     }
 
@@ -736,4 +739,61 @@ mod tests {
             .await
             .unwrap();
     }
+
+    fn from_hex(s: &str) -> Vec<u8> {
+        assert!(s.len() % 2 == 0, "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. These two hex strings were captured from the "JAVA key_metadata
+    // blob hex" / "JAVA plaintext DEK hex" lines printed by CometIcebergEncryptionSuite (an
+    // Iceberg-Java StandardEncryptionManager producing a real V1 blob). Regenerate if the wire
+    // format changes.
+    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 a23a7b1f67..d5669f4751 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -3703,6 +3703,9 @@ fn parse_file_scan_tasks_from_common(
                         } else {
                             Some(del.equality_ids.clone())
                         },
+                        // 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>>()
@@ -3827,6 +3830,9 @@ fn parse_file_scan_tasks_from_common(
                 partition_spec,
                 name_mapping,
                 case_sensitive: false,
+                // 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 00158644d1..04f9315d1b 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -251,6 +251,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;
@@ -277,6 +282,13 @@ message IcebergDeleteFile {
 
   // Equality field IDs (empty for positional deletes)
   repeated int32 equality_ids = 4;
+
+  // Serialized StandardKeyMetadata for an encrypted delete file, from DeleteFile.keyMetadata().
+  // Absent for unencrypted tables. Decoded directly by iceberg-rust; no KMS unwrap needed.
+  // Field 8: 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.
+  optional bytes key_metadata = 8;
 }
 
 message Projection {
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 9f48ddfa9e..811a0462e9 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"
   }
 
   /**
@@ -735,6 +736,56 @@ 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]
+  }
 }
 
 /**
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..07db12635d 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -462,12 +462,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 +483,48 @@ 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
+          }
+
         // Single-pass validation of all FileScanTasks
         val taskValidation =
           try {
@@ -713,6 +761,7 @@ case class CometScanRule(session: SparkSession)
         }
 
         if (schemaSupported && fileIOCompatible && formatVersionSupported &&
+          defaultValuesSupported && schemaTypesSupported &&
           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 879762818f..8dd1a597a9 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
@@ -283,6 +283,19 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
           case _: Exception =>
         }
 
+        // Encrypted delete files carry a plaintext StandardKeyMetadata blob (keyMetadata() is on
+        // ContentFile); forward it verbatim. Null (unencrypted) leaves the field unset.
+        try {
+          contentFileClass.getMethod("keyMetadata").invoke(deleteFile) match {
+            case buf: java.nio.ByteBuffer if buf.remaining() > 0 =>
+              deleteBuilder.setKeyMetadata(
+                com.google.protobuf.ByteString.copyFrom(buf.duplicate()))
+            case _ =>
+          }
+        } catch {
+          case _: Exception =>
+        }
+
         deleteBuilder.build()
       }.toSeq
     } catch {
@@ -770,6 +783,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)
@@ -827,6 +843,15 @@ 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. Null (unencrypted) leaves the field unset.
+                keyMetadataMethod.invoke(dataFile) match {
+                  case buf: java.nio.ByteBuffer if buf.remaining() > 0 =>
+                    taskBuilder.setKeyMetadata(
+                      com.google.protobuf.ByteString.copyFrom(buf.duplicate()))
+                  case _ =>
+                }
+
                 val taskSchema = taskSchemaMethod.invoke(task)
 
                 val deletes =
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..aba41f3c98
--- /dev/null
+++ b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
@@ -0,0 +1,105 @@
+/*
+ * 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.nio.ByteBuffer
+import java.util.Collections
+
+import org.apache.iceberg.Files
+import org.apache.iceberg.encryption.{EncryptedKey, NativeEncryptionKeyMetadata, StandardEncryptionManager}
+import org.apache.spark.sql.CometTestBase
+
+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 {
+
+  // Confirms the assumption that native encrypted reads will 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 ever wraps the KEK
+  // / manifest-list keys, never the per-file DEK. So Comet can forward these bytes straight to
+  // iceberg-rust with no native KMS.
+  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 = toArray(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 = toArray(parsed.encryptionKey())
+
+      assert(
+        java.util.Arrays.equals(toArray(dek), roundTripped),
+        "DEK did not survive the StandardKeyMetadata encode/parse round-trip")
+
+      println(
+        s"key_metadata blob ${blob.length} bytes, plaintext DEK ${roundTripped.length} bytes; " +
+          "no KMS needed to recover the data-file key")
+      // Emit the raw bytes so the Rust side (iceberg-rust StandardKeyMetadata::decode) can be
+      // tested against a real Java-produced blob. Paste these into the Rust cross-compat test.
+      // println (not logInfo) so the values surface in the IntelliJ test console.
+      println(s"JAVA key_metadata blob hex: ${hex(blob)}")
+      println(s"JAVA plaintext DEK hex:     ${hex(roundTripped)}")
+    }
+  }
+
+  private def toArray(buffer: ByteBuffer): Array[Byte] = {
+    val dup = buffer.duplicate()
+    val bytes = new Array[Byte](dup.remaining())
+    dup.get(bytes)
+    bytes
+  }
+
+  private def hex(bytes: Array[Byte]): String = bytes.map(b => f"$b%02x").mkString
+}
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..85202c7fb4
--- /dev/null
+++ b/spark/src/test/spark-4.1/org/apache/comet/iceberg/CometTestKMS.scala
@@ -0,0 +1,69 @@
+/*
+ * 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
+
+/**
+ * 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(toArray(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(toArray(wrappedKey), null))
+  }
+
+  override def initialize(properties: util.Map[String, String]): Unit = {}
+
+  private def toArray(buffer: ByteBuffer): Array[Byte] = {
+    val dup = buffer.duplicate()
+    val bytes = new Array[Byte](dup.remaining())
+    dup.get(bytes)
+    bytes
+  }
+}
+
+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
+  }
+}

From ef554e2bab7cf30b68843cf968cd76af59ff317a Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 10:46:20 -0400
Subject: [PATCH 02/14] add encryption suite to CI

---
 .github/workflows/pr_build_linux.yml | 1 +
 .github/workflows/pr_build_macos.yml | 1 +
 2 files changed, 2 insertions(+)

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

From e13c7a445a82847472e22130206ae305735c7c91 Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 10:54:22 -0400
Subject: [PATCH 03/14] fix clippy

---
 native/core/src/execution/operators/iceberg_scan.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs
index 9db4d2948a..219b787299 100644
--- a/native/core/src/execution/operators/iceberg_scan.rs
+++ b/native/core/src/execution/operators/iceberg_scan.rs
@@ -741,7 +741,7 @@ mod tests {
     }
 
     fn from_hex(s: &str) -> Vec<u8> {
-        assert!(s.len() % 2 == 0, "odd-length hex string");
+        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"))

From 0131329cf171d9d5d274acbbfc2fe93dbae49efc Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 11:07:29 -0400
Subject: [PATCH 04/14] add end-to-end test using hive catalog

---
 .../src/execution/operators/iceberg_scan.rs   |   7 +-
 spark/pom.xml                                 |   7 ++
 .../comet/CometIcebergEncryptionSuite.scala   | 103 +++++++++++++++---
 3 files changed, 97 insertions(+), 20 deletions(-)

diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs
index 219b787299..d77b208964 100644
--- a/native/core/src/execution/operators/iceberg_scan.rs
+++ b/native/core/src/execution/operators/iceberg_scan.rs
@@ -769,10 +769,9 @@ mod tests {
     // 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. These two hex strings were captured from the "JAVA key_metadata
-    // blob hex" / "JAVA plaintext DEK hex" lines printed by CometIcebergEncryptionSuite (an
-    // Iceberg-Java StandardEncryptionManager producing a real V1 blob). Regenerate if the wire
-    // format changes.
+    // 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";
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/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
index aba41f3c98..3fffd56bf1 100644
--- a/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
+++ b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
@@ -24,7 +24,10 @@ import java.util.Collections
 
 import org.apache.iceberg.Files
 import org.apache.iceberg.encryption.{EncryptedKey, NativeEncryptionKeyMetadata, StandardEncryptionManager}
+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
 
@@ -33,14 +36,93 @@ import org.apache.comet.iceberg.CometTestKMS
  * 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 {
+class CometIcebergEncryptionSuite
+    extends CometTestBase
+    with CometIcebergTestBase
+    with AdaptiveSparkPlanHelper {
 
-  // Confirms the assumption that native encrypted reads will rely on: an Iceberg data file's
+  // 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.
+  private val hiveWarehouse =
+    java.nio.file.Files.createTempDirectory("comet-hive-warehouse").toFile
+
+  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,
+        s"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 ever wraps the KEK
-  // / manifest-list keys, never the per-file DEK. So Comet can forward these bytes straight to
-  // iceberg-rust with no native KMS.
+  // 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+")
@@ -82,15 +164,6 @@ class CometIcebergEncryptionSuite extends CometTestBase with CometIcebergTestBas
       assert(
         java.util.Arrays.equals(toArray(dek), roundTripped),
         "DEK did not survive the StandardKeyMetadata encode/parse round-trip")
-
-      println(
-        s"key_metadata blob ${blob.length} bytes, plaintext DEK ${roundTripped.length} bytes; " +
-          "no KMS needed to recover the data-file key")
-      // Emit the raw bytes so the Rust side (iceberg-rust StandardKeyMetadata::decode) can be
-      // tested against a real Java-produced blob. Paste these into the Rust cross-compat test.
-      // println (not logInfo) so the values surface in the IntelliJ test console.
-      println(s"JAVA key_metadata blob hex: ${hex(blob)}")
-      println(s"JAVA plaintext DEK hex:     ${hex(roundTripped)}")
     }
   }
 
@@ -100,6 +173,4 @@ class CometIcebergEncryptionSuite extends CometTestBase with CometIcebergTestBas
     dup.get(bytes)
     bytes
   }
-
-  private def hex(bytes: Array[Byte]): String = bytes.map(b => f"$b%02x").mkString
 }

From 93617ce9d35b79f4c50359a96f6ebbd2226b828b Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 11:30:48 -0400
Subject: [PATCH 05/14] add 256-bit fallback until arrow 58.4.0 lands

---
 .../comet/iceberg/IcebergReflection.scala     | 11 +++++
 .../apache/comet/rules/CometScanRule.scala    | 21 +++++++++-
 .../operator/CometIcebergNativeScan.scala     | 42 ++++++++++---------
 .../comet/CometIcebergEncryptionSuite.scala   | 42 +++++++++++++++----
 .../apache/comet/iceberg/CometTestKMS.scala   | 12 ++----
 5 files changed, 91 insertions(+), 37 deletions(-)

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 811a0462e9..29bf1920b4 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -786,6 +786,17 @@ object IcebergReflection extends Logging {
       .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 (arrow-rs 58.3) supports only 128-bit (16-byte) keys, 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 07db12635d..4b1e495dac 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -525,6 +525,25 @@ case class CometScanRule(session: SparkSession)
               false
           }
 
+        // The native Parquet reader (arrow-rs 58.3) decrypts only 128-bit AES-GCM data keys. Fall
+        // back for larger keys until arrow-rs 58.4 adds 256-bit support (arrow-rs#10349). 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 =>
+                fallbackReasons += s"Iceberg table encryption with a ${len * 8}-bit data key is " +
+                  "not yet supported by Comet's native reader (only 128-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 {
@@ -761,7 +780,7 @@ case class CometScanRule(session: SparkSession)
         }
 
         if (schemaSupported && fileIOCompatible && formatVersionSupported &&
-          defaultValuesSupported && schemaTypesSupported &&
+          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 8dd1a597a9..eaf8ae5f94 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
@@ -219,6 +219,22 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
     }
   }
 
+  /**
+   * 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).
    *
@@ -232,6 +248,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)
 
@@ -283,18 +301,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
           case _: Exception =>
         }
 
-        // Encrypted delete files carry a plaintext StandardKeyMetadata blob (keyMetadata() is on
-        // ContentFile); forward it verbatim. Null (unencrypted) leaves the field unset.
-        try {
-          contentFileClass.getMethod("keyMetadata").invoke(deleteFile) match {
-            case buf: java.nio.ByteBuffer if buf.remaining() > 0 =>
-              deleteBuilder.setKeyMetadata(
-                com.google.protobuf.ByteString.copyFrom(buf.duplicate()))
-            case _ =>
-          }
-        } catch {
-          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
@@ -844,13 +853,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                 taskBuilder.setFileSizeInBytes(fileSizeInBytes)
 
                 // Encrypted data files carry a plaintext StandardKeyMetadata blob; forward it
-                // verbatim for iceberg-rust to decode. Null (unencrypted) leaves the field unset.
-                keyMetadataMethod.invoke(dataFile) match {
-                  case buf: java.nio.ByteBuffer if buf.remaining() > 0 =>
-                    taskBuilder.setKeyMetadata(
-                      com.google.protobuf.ByteString.copyFrom(buf.duplicate()))
-                  case _ =>
-                }
+                // verbatim for iceberg-rust to decode. Unencrypted files leave the field unset.
+                keyMetadataBytes(keyMetadataMethod, dataFile).foreach(taskBuilder.setKeyMetadata)
 
                 val taskSchema = taskSchemaMethod.invoke(task)
 
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
index 3fffd56bf1..b5b4503213 100644
--- a/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
+++ b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
@@ -24,6 +24,7 @@ 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
@@ -151,7 +152,7 @@ class CometIcebergEncryptionSuite
       // 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 = toArray(keyMetadata.buffer())
+      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")
@@ -159,18 +160,43 @@ class CometIcebergEncryptionSuite
       parse.setAccessible(true)
       val parsed =
         parse.invoke(null, ByteBuffer.wrap(blob)).asInstanceOf[NativeEncryptionKeyMetadata]
-      val roundTripped = toArray(parsed.encryptionKey())
+      val roundTripped = ByteBuffers.toByteArray(parsed.encryptionKey())
 
       assert(
-        java.util.Arrays.equals(toArray(dek), roundTripped),
+        java.util.Arrays.equals(ByteBuffers.toByteArray(dek), roundTripped),
         "DEK did not survive the StandardKeyMetadata encode/parse round-trip")
     }
   }
 
-  private def toArray(buffer: ByteBuffer): Array[Byte] = {
-    val dup = buffer.duplicate()
-    val bytes = new Array[Byte](dup.remaining())
-    dup.get(bytes)
-    bytes
+  // The native Parquet reader (arrow-rs 58.3) decrypts only 128-bit AES-GCM keys, so a table with a
+  // larger data key must fall back to Spark and still return correct results. Flip this to a native
+  // assertion when arrow-rs 58.4 (256-bit support, arrow-rs#10349) is picked up.
+  test("encrypted V3 table with a 256-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_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 (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $table ORDER BY id")
+      assert(
+        collect(cometPlan) { case s: CometIcebergNativeScanExec => s }.isEmpty,
+        s"expected a 256-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
index 85202c7fb4..b9317af6b2 100644
--- 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
@@ -24,6 +24,7 @@ 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
@@ -36,23 +37,16 @@ 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(toArray(key), null))
+    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(toArray(wrappedKey), null))
+    ByteBuffer.wrap(decryptor.decrypt(ByteBuffers.toByteArray(wrappedKey), null))
   }
 
   override def initialize(properties: util.Map[String, String]): Unit = {}
-
-  private def toArray(buffer: ByteBuffer): Array[Byte] = {
-    val dup = buffer.duplicate()
-    val bytes = new Array[Byte](dup.remaining())
-    dup.get(bytes)
-    bytes
-  }
 }
 
 object CometTestKMS {

From e854424c9f73d5cf9f4769964675558a14c4af90 Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 12:05:39 -0400
Subject: [PATCH 06/14] format, again

---
 .../org/apache/comet/CometIcebergEncryptionSuite.scala        | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

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
index b5b4503213..0b3a5d0730 100644
--- a/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
+++ b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
@@ -112,7 +112,7 @@ class CometIcebergEncryptionSuite
       val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $table ORDER BY id")
       assert(
         collect(cometPlan) { case s: CometIcebergNativeScanExec => s }.nonEmpty,
-        s"expected the encrypted read to run natively but found no CometIcebergNativeScanExec:\n" +
+        "expected the encrypted read to run natively but found no CometIcebergNativeScanExec:\n" +
           cometPlan)
 
       spark.sql(s"DROP TABLE $table")
@@ -193,7 +193,7 @@ class CometIcebergEncryptionSuite
       val (_, cometPlan) = checkSparkAnswer(s"SELECT * FROM $table ORDER BY id")
       assert(
         collect(cometPlan) { case s: CometIcebergNativeScanExec => s }.isEmpty,
-        s"expected a 256-bit encrypted table to fall back to Spark but it ran natively:\n" +
+        "expected a 256-bit encrypted table to fall back to Spark but it ran natively:\n" +
           cometPlan)
 
       spark.sql(s"DROP TABLE $table")

From a0a23e53200220443243f92813d032f2b77bccf3 Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 14:18:24 -0400
Subject: [PATCH 07/14] fix hive warehouse initialization

---
 .../apache/comet/CometIcebergEncryptionSuite.scala  | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

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
index 0b3a5d0730..98f08b540d 100644
--- a/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
+++ b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
@@ -47,8 +47,17 @@ class CometIcebergEncryptionSuite
   // 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.
-  private val hiveWarehouse =
-    java.nio.file.Files.createTempDirectory("comet-hive-warehouse").toFile
+  //
+  // 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)
+    java.nio.file.Files.createTempDirectory(base, "comet-hive-warehouse").toFile
+  }
 
   override protected def sparkConf: SparkConf = {
     super.sparkConf

From 7cf63ebbe18aca23a5f31d780be51573456e940b Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 15:14:00 -0400
Subject: [PATCH 08/14] fix reflection issue on metadata tables spotted when
 running Iceberg Java's encryption suite locally

---
 .../main/scala/org/apache/comet/rules/CometScanRule.scala  | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

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 4b1e495dac..1d91ac7dad 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

From 8638d7b52984eb756794fc536985d57dfb7e0c39 Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 15:34:46 -0400
Subject: [PATCH 09/14] update docs.

---
 docs/source/contributor-guide/roadmap.md |  6 +++---
 docs/source/user-guide/latest/iceberg.md | 22 ++++++++++++++++------
 2 files changed, 19 insertions(+), 9 deletions(-)

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..2e7fc81079 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,14 @@ 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 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.
 
 **Schema and data types:**
 
@@ -156,17 +163,20 @@ 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 data keys larger than 128-bit (256-bit support needs arrow-rs 58.4,
+  [arrow-rs#10349](https://github.com/apache/arrow-rs/issues/10349))
+- 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
 

From 9a3a994aa101bb14c05175b6a4bfe1636f0c4aa1 Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Tue, 21 Jul 2026 16:16:07 -0400
Subject: [PATCH 10/14] Address PR feedback.

---
 docs/source/user-guide/latest/iceberg.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md
index 2e7fc81079..f1c40e198b 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/docs/source/user-guide/latest/iceberg.md
@@ -70,7 +70,8 @@ The native Iceberg reader supports the following features:
 - Encrypted v3 tables using 128-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.
+  the native side. Iceberg-Java also permits 192- and 256-bit data keys; those tables fall back to
+  Spark for now (the native Parquet reader decrypts only 128-bit AES-GCM until arrow-rs 58.4).
 
 **Schema and data types:**
 

From 15f71f83b9cc1857a47a034dbe14358acbb65532 Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Wed, 22 Jul 2026 08:05:59 -0400
Subject: [PATCH 11/14] clean up Hive warehouse files, and adjust Maven cache
 key computation

---
 .github/actions/java-test/action.yaml              | 14 +++++++++++++-
 .../org/apache/comet/CometIcebergTestBase.scala    |  2 +-
 .../apache/comet/CometIcebergEncryptionSuite.scala | 14 +++++++++++++-
 3 files changed, 27 insertions(+), 3 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/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
index 98f08b540d..3742841361 100644
--- a/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
+++ b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
@@ -19,6 +19,7 @@
 
 package org.apache.comet
 
+import java.io.File
 import java.nio.ByteBuffer
 import java.util.Collections
 
@@ -56,7 +57,18 @@ class CometIcebergEncryptionSuite
   private lazy val hiveWarehouse = {
     val base = java.nio.file.Paths.get(System.getProperty("java.io.tmpdir"))
     java.nio.file.Files.createDirectories(base)
-    java.nio.file.Files.createTempDirectory(base, "comet-hive-warehouse").toFile
+    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 = {

From df4d3219de036fd5dd531f96dc60b29018b63d48 Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Thu, 23 Jul 2026 07:54:52 -0400
Subject: [PATCH 12/14] add 256-bit data file key support, bump arrow-rs to
 58.4.0

---
 docs/source/user-guide/latest/iceberg.md      | 13 ++--
 native/Cargo.lock                             | 76 +++++++++----------
 native/Cargo.toml                             |  8 +-
 .../comet/iceberg/IcebergReflection.scala     |  5 +-
 .../apache/comet/rules/CometScanRule.scala    | 11 +--
 .../comet/CometIcebergEncryptionSuite.scala   | 48 ++++++++++--
 6 files changed, 100 insertions(+), 61 deletions(-)

diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md
index f1c40e198b..a8ea6f30ce 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/docs/source/user-guide/latest/iceberg.md
@@ -67,11 +67,11 @@ The native Iceberg reader supports the following features:
 
 **Encryption:**
 
-- Encrypted v3 tables using 128-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- and 256-bit data keys; those tables fall back to
-  Spark for now (the native Parquet reader decrypts only 128-bit AES-GCM until arrow-rs 58.4).
+- 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:**
 
@@ -169,8 +169,7 @@ The following scenarios will fall back to the JVM Iceberg reader:
 - 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 data keys larger than 128-bit (256-bit support needs arrow-rs 58.4,
-  [arrow-rs#10349](https://github.com/apache/arrow-rs/issues/10349))
+- 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)
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/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index 29bf1920b4..ff688d9ce6 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -790,8 +790,9 @@ object IcebergReflection extends Logging {
   /**
    * 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 (arrow-rs 58.3) supports only 128-bit (16-byte) keys, so callers fall
-   * back for anything else. Throws on reflection failure so the caller can fall back.
+   * 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 =>
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 1d91ac7dad..5e451b382a 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -530,15 +530,16 @@ case class CometScanRule(session: SparkSession)
               false
           }
 
-        // The native Parquet reader (arrow-rs 58.3) decrypts only 128-bit AES-GCM data keys. Fall
-        // back for larger keys until arrow-rs 58.4 adds 256-bit support (arrow-rs#10349). None
-        // means the table is unencrypted. Reflection failure also falls back.
+        // 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 =>
+              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 AES-GCM)"
+                  "not yet supported by Comet's native reader (only 128-bit and 256-bit AES-GCM)"
                 false
               case _ => true
             }
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
index 3742841361..1b221bf454 100644
--- a/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
+++ b/spark/src/test/spark-4.1/org/apache/comet/CometIcebergEncryptionSuite.scala
@@ -189,10 +189,9 @@ class CometIcebergEncryptionSuite
     }
   }
 
-  // The native Parquet reader (arrow-rs 58.3) decrypts only 128-bit AES-GCM keys, so a table with a
-  // larger data key must fall back to Spark and still return correct results. Flip this to a native
-  // assertion when arrow-rs 58.4 (256-bit support, arrow-rs#10349) is picked up.
-  test("encrypted V3 table with a 256-bit data key falls back to Spark") {
+  // 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+")
 
@@ -211,10 +210,49 @@ class CometIcebergEncryptionSuite
       """)
       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 256-bit encrypted table to fall back to Spark but it ran natively:\n" +
+        "expected a 192-bit encrypted table to fall back to Spark but it ran natively:\n" +
           cometPlan)
 
       spark.sql(s"DROP TABLE $table")

From 0868fba71c4653c16fb055eeddc6a02fd6228853 Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Thu, 23 Jul 2026 10:59:40 -0400
Subject: [PATCH 13/14] address PR feedback

---
 native/proto/src/proto/operator.proto | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto
index 235510c8b2..21480569f5 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -283,11 +283,13 @@ message IcebergDeleteFile {
   // Equality field IDs (empty for positional deletes)
   repeated int32 equality_ids = 4;
 
-  // Serialized StandardKeyMetadata for an encrypted delete file, from DeleteFile.keyMetadata().
-  // Absent for unencrypted tables. Decoded directly by iceberg-rust; no KMS unwrap needed.
-  // Field 8: fields 5-7 are reserved for the deletion-vector coordinates on the dv-read branch
+  // 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;
 }
 

From c57619cb56461854e15bc94b535e99666af83249 Mon Sep 17 00:00:00 2001
From: Matt Butrovich <mbutrovich@gmail.com>
Date: Thu, 23 Jul 2026 13:58:49 -0400
Subject: [PATCH 14/14] format

---
 .../apache/comet/serde/operator/CometIcebergNativeScan.scala  | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

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 c4d872e9c5..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
@@ -239,8 +239,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
    * 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: 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()))