diff --git a/.github/workflows/README.md b/.github/workflows/README.md
index 6addfd923b9..849b1224074 100644
--- a/.github/workflows/README.md
+++ b/.github/workflows/README.md
@@ -29,7 +29,7 @@ ruleset in `.asf.yaml`. That splits CI into three tiers:
- **Nightly tier** (`nightly`): the regression sweep of everything else, once
a day against `main` as it stands. The Comet test suites against the other
three Spark profiles, Spark SQL on Spark 3.5, 4.0 and 4.2, Iceberg
- 1.9/1.10, and the Delta contrib suites on Spark 3.5, 4.0 and 4.1. See
+ 1.9/1.10, and the Delta contrib suites on Spark 3.5, 4.0, 4.1 and 4.2. See
[Nightly tier](#nightly-tier) below for how a failure surfaces.
Every queue-only and nightly job has a `run-*` label that opts a pull request
@@ -122,6 +122,7 @@ tiers partition the list and that the `pr` tier is exactly the default profile.
delta_3_5 run-delta-tests
delta_4_0 run-delta-tests
delta_4_1 run-delta-tests
+ delta_4_2 run-delta-tests
| | |
+-----------------------------------+-----------------------------------+
@@ -161,6 +162,7 @@ tiers partition the list and that the `pr` tier is exactly the default profile.
| `delta_3_5` | nightly, **or** PR with `run-delta-tests` | Delta contrib and native sources |
| `delta_4_0` | nightly, **or** PR with `run-delta-tests` | Delta contrib and native sources |
| `delta_4_1` | nightly, **or** PR with `run-delta-tests` | Delta contrib and native sources |
+| `delta_4_2` | nightly, **or** PR with `run-delta-tests` | Delta contrib and native sources |
| `docs` | push to main, paths matched, **or** dispatch on `main` | `.asf.yaml`, `docs/**`, `docs.yaml` |
| `spark_3_5` | nightly, **or** PR with `run-spark-3.5-tests` | Spark 3.5 sources |
| `spark_4_1` | PR or merge group for catalyst and `sql_core`; the `sql_hive` shards in the merge group, **or** PR with `run-spark-4.1-hive-tests` | Spark 4.1 sources |
@@ -379,16 +381,16 @@ umbrella doesn't watch, or operate independently of the rest of CI:
## Reusable workflows (called by `ci.yml`)
-| File | Called from `ci.yml` job(s) |
-| ----------------------------- | ------------------------------------- |
-| `pr_build_linux.yml` | `pr_build_linux` |
-| `pr_build_macos.yml` | `pr_build_macos` |
-| `pr_benchmark_check.yml` | `pr_benchmark_check` |
-| `delta_build_gate.yml` | `delta_build_gate` |
-| `delta_contrib_test.yml` | `delta_3_5`, `delta_4_0`, `delta_4_1` |
-| `pyarrow_udf_test.yml` | `pyarrow_udf_test` |
-| `docs.yaml` | `docs` |
-| `spark_sql_test_reusable.yml` | `iceberg_spark_test_reusable.yml` |
+| File | Called from `ci.yml` job(s) |
+| ----------------------------- | -------------------------------------------------- |
+| `pr_build_linux.yml` | `pr_build_linux` |
+| `pr_build_macos.yml` | `pr_build_macos` |
+| `pr_benchmark_check.yml` | `pr_benchmark_check` |
+| `delta_build_gate.yml` | `delta_build_gate` |
+| `delta_contrib_test.yml` | `delta_3_5`, `delta_4_0`, `delta_4_1`, `delta_4_2` |
+| `pyarrow_udf_test.yml` | `pyarrow_udf_test` |
+| `docs.yaml` | `docs` |
+| `spark_sql_test_reusable.yml` | `iceberg_spark_test_reusable.yml` |
## Changing what runs when
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f644c8f7fa4..47f094dda40 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -302,6 +302,7 @@ jobs:
delta_3_5: ${{ steps.compute.outputs.delta_3_5 }}
delta_4_0: ${{ steps.compute.outputs.delta_4_0 }}
delta_4_1: ${{ steps.compute.outputs.delta_4_1 }}
+ delta_4_2: ${{ steps.compute.outputs.delta_4_2 }}
steps:
- uses: actions/checkout@v7
with:
@@ -524,14 +525,25 @@ jobs:
needs: changes
permissions:
contents: read
- # Nightly by default; PRs need the `run-delta-tests` label. No Spark 4.2
- # caller: the contrib profile is dormant until a Delta release supports it.
+ # Nightly by default; PRs need the `run-delta-tests` label.
if: needs.changes.outputs.delta_4_1 == 'true'
uses: ./.github/workflows/delta_contrib_test.yml
with:
spark: '4.1'
java: '17'
+ delta_4_2:
+ name: Delta Contrib Tests (Spark 4.2)
+ needs: changes
+ permissions:
+ contents: read
+ # Nightly by default; PRs need the `run-delta-tests` label.
+ if: needs.changes.outputs.delta_4_2 == 'true'
+ uses: ./.github/workflows/delta_contrib_test.yml
+ with:
+ spark: '4.2'
+ java: '17'
+
spark_4_0:
name: Spark SQL Tests (Spark 4.0)
needs: changes
@@ -673,6 +685,7 @@ jobs:
- delta_3_5
- delta_4_0
- delta_4_1
+ - delta_4_2
- spark_4_0
- spark_4_1
- spark_4_2
diff --git a/.github/workflows/delta_contrib_test.yml b/.github/workflows/delta_contrib_test.yml
index 10e6b9f1029..7ce8b20d21b 100644
--- a/.github/workflows/delta_contrib_test.yml
+++ b/.github/workflows/delta_contrib_test.yml
@@ -135,9 +135,9 @@ jobs:
# The container job above has no Docker socket, so CometDeltaS3Suite cancels itself
# there. This job runs it on the plain runner, where Testcontainers can start MinIO.
contrib-delta-s3:
- name: Delta contrib S3 suite (Spark 3.5, MinIO)
- # Pinned to Spark 3.5, so only that caller runs it: once per ci.yml run.
- if: inputs.spark == '3.5'
+ name: Delta contrib S3 suite (Spark ${{ inputs.spark }}, MinIO)
+ # Cover the oldest pairing and Hadoop 3.5's encryption-context initialization.
+ if: inputs.spark == '3.5' || inputs.spark == '4.2'
runs-on: ubuntu-24.04
steps:
- uses: actions/checkout@v7
@@ -167,7 +167,7 @@ jobs:
with:
path: |
~/.m2/repository
- key: ${{ runner.os }}-delta-maven-${{ hashFiles('**/pom.xml') }}-3.5
+ key: ${{ runner.os }}-delta-maven-${{ hashFiles('**/pom.xml') }}-${{ inputs.spark }}
# Its own prefix: the Spark SQL and Comet test jobs fall back to the
# newest `java-maven-` entry, and this repository (built with
# `-Pdelta`) broke Spark's Arrow tests when they restored it.
@@ -200,12 +200,12 @@ jobs:
- name: Install Comet core jars
run: |
- ./mvnw -B -q -Prelease -Pspark-3.5,delta install -pl common,spark -DskipTests -Dspotless.check.skip=true
+ ./mvnw -B -q -Prelease -Pspark-${{ inputs.spark }},delta install -pl common,spark -DskipTests -Dspotless.check.skip=true
- name: Run the MinIO-backed S3 suite
run: |
docker info > /dev/null
- SPARK_HOME=$(pwd) COMET_CONF_DIR=$(pwd)/conf ./mvnw -B -Prelease -Pspark-3.5,delta test -pl contrib/delta-spark -DwildcardSuites=org.apache.comet.contrib.delta.CometDeltaS3Suite
+ SPARK_HOME=$(pwd) COMET_CONF_DIR=$(pwd)/conf ./mvnw -B -Prelease -Pspark-${{ inputs.spark }},delta test -pl contrib/delta-spark -DwildcardSuites=org.apache.comet.contrib.delta.CometDeltaS3Suite
env:
# This job exists only to run the suite, so a Docker or MinIO failure must fail
# it instead of cancelling every test, which the container job above relies on.
@@ -218,7 +218,7 @@ jobs:
with:
path: |
~/.m2/repository
- key: ${{ runner.os }}-delta-maven-${{ hashFiles('**/pom.xml') }}-3.5
+ key: ${{ runner.os }}-delta-maven-${{ hashFiles('**/pom.xml') }}-${{ inputs.spark }}
# `delta` is in the default feature set, so no regular job builds without it;
# this keeps the feature-off build and its "built without the delta feature"
diff --git a/contrib/delta-spark/README.md b/contrib/delta-spark/README.md
index 89664d3d61a..42de38f1afb 100644
--- a/contrib/delta-spark/README.md
+++ b/contrib/delta-spark/README.md
@@ -43,7 +43,11 @@ for configuration details.
| 4.0 | 4.0.x | supported |
| 4.1 | 4.3.x | supported |
| 3.4 | delta-core 2.4 | not supported (older Delta, would need shims) |
-| 4.2 | none released | inert until Delta ships a Spark 4.2 release |
+| 4.2 | 4.4.x | supported |
+
+Spark 4.2 uses the `delta-spark_4.2_2.13` artifact, pinned to Delta 4.4.1 by
+the `spark-4.2` Maven profile. Earlier Spark profiles keep `delta-spark` with
+their existing version pins.
## Building and testing
@@ -60,7 +64,7 @@ install is the trap the contributor guide warns about:
Run the test suites the same way (`test` instead of `install` on the second
line). CI runs them via `.github/workflows/delta_contrib_test.yml`: nightly on
-Spark 3.5, 4.0 and 4.1, or on a pull request carrying the `run-delta-tests`
+Spark 3.5, 4.0, 4.1 and 4.2, or on a pull request carrying the `run-delta-tests`
label.
`CometDeltaS3Suite` starts MinIO through Testcontainers and cancels itself when
no Docker daemon is reachable; setting `COMET_DELTA_S3_REQUIRED=1`, as the
@@ -68,3 +72,9 @@ no Docker daemon is reachable; setting `COMET_DELTA_S3_REQUIRED=1`, as the
`dev/` contains a benchmark script (`bench_delta_comet.py`) and a harness for
running Delta's own test suites against Comet (`run-delta-regression.sh`).
+
+### S3 encryption context
+
+A nonempty effective `fs.s3a.encryption.context` makes the Delta scan fall back to Spark. The fallback also applies on older Hadoop versions that ignore the key. The context is attached to KMS write requests, but Hadoop 3.5 also parses it while initializing an S3 filesystem for reads. A local Delta log or shallow clone can reference S3 data files without initializing that data bucket during metadata reads. Comet therefore declines configured contexts, including valid ones, so Spark performs the same initialization checks before opening those files. Bucket aliases and credential-provider values are included; decline reasons do not print context values. Existing SSE-C and client-side encryption declines still apply.
+
+The MinIO CI jobs run on Spark 3.5 and 4.2. The Spark 4.2 case covers a local shallow clone whose files remain in S3, with valid and malformed global, short-bucket and long-bucket context settings. The unconfigured control must use the native Delta scan; configured reads must fall back, returning rows for valid context and failing like Spark for malformed context.
diff --git a/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala b/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala
index 96c18a85de0..c6733091dec 100644
--- a/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala
+++ b/contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala
@@ -388,6 +388,12 @@ object DeltaScanSupport {
return encryptionReason
}
+ val contextReason =
+ unsupportedEncryptionContextReason(hadoopConf, dataFileUris ++ dvUris, propagatedConfCache)
+ if (contextReason.isDefined) {
+ return contextReason
+ }
+
// Always zero-I/O (plain propagated-conf read, no keystore): native's S3 client has no
// HTTP proxy support at all (no fs.s3a.proxy.* key is read anywhere in s3.rs), so a bucket
// requiring a proxy for S3 egress must decline here rather than claim and then connect
@@ -1155,6 +1161,42 @@ object DeltaScanSupport {
}
}
+ /**
+ * Decline S3 encryption context because native does not reproduce Hadoop's setup-time parsing.
+ * A local Delta log can reference S3 data or DVs, so metadata reads need not initialize S3A.
+ * Resolve the same bucket aliases and credential providers as Hadoop, without exposing values.
+ */
+ private[delta] def unsupportedEncryptionContextReason(
+ hadoopConf: Configuration,
+ uris: Seq[URI],
+ propagatedConfCache: MutableMap[String, Configuration] = MutableMap.empty)
+ : Option[String] = {
+ val key = "fs.s3a.encryption.context"
+ uris
+ .flatMap(s3Bucket)
+ .distinct
+ .iterator
+ .map { bucket =>
+ try {
+ val propagatedConf = propagatedConfCache.getOrElseUpdate(
+ bucket,
+ propagateBucketOptions(hadoopConf, bucket))
+ hadoopLookupPasswordEffective(propagatedConf, bucket, key) match {
+ case Left(reason) => Some(reason)
+ case Right(Some(_)) =>
+ Some(
+ s"Native Delta scan does not support $key for $bucket " +
+ "(Hadoop validates the context during S3A initialization)")
+ case Right(None) => None
+ }
+ } catch {
+ case e @ (_: IOException | _: RuntimeException) =>
+ Some(unverifiableValueReason(bucket, key, e))
+ }
+ }
+ .collectFirst { case Some(reason) => reason }
+ }
+
private def encryptionAlgorithmReason(conf: Configuration, bucket: String): Option[String] =
effectiveEncryptionAlgorithm(conf, bucket) match {
case Left(reason) => Some(reason)
@@ -1522,22 +1564,22 @@ object DeltaScanSupport {
* for the Arm A/B/C dispatch this delegates to) -- or `Left(reason)` when the value cannot be
* safely verified.
*
- * USED ONLY for keys whose real consumer IS `lookupPassword` -- the [[LookupPasswordConsumer]]
- * entries of [[S3ConfigKeyConsumers]], via [[s3KeyDivergenceReason]]. An earlier version ran
- * EVERY compared key through this function, reasoning that a wider read could only ever
- * over-decline; for a value-EQUALITY comparator that reasoning is half-true: the long-form
- * alias this function consults FIRST can hold exactly the value native resolves while Hadoop's
- * true propagate-then-plain-get value differs (e.g. a `${...}` reference whose referent
- * propagation redirects), producing a false EQUALITY that admits a really-diverging scan. A
- * [[PropagatedOptionConsumer]] key must therefore resolve like its actual consumer instead --
- * see [[s3KeyDivergenceReason]].
+ * Used for keys whose real consumer is `lookupPassword`: the [[LookupPasswordConsumer]] entries
+ * of [[S3ConfigKeyConsumers]] and the dedicated [[unsupportedEncryptionContextReason]] gate. An
+ * earlier version ran EVERY compared key through this function, reasoning that a wider read
+ * could only ever over-decline; for a value-EQUALITY comparator that reasoning is half-true:
+ * the long-form alias this function consults FIRST can hold exactly the value native resolves
+ * while Hadoop's true propagate-then-plain-get value differs (e.g. a `${...}` reference whose
+ * referent propagation redirects), producing a false EQUALITY that admits a really-diverging
+ * scan. A [[PropagatedOptionConsumer]] key must therefore resolve like its actual consumer
+ * instead -- see [[s3KeyDivergenceReason]].
*
* Honors [[ClearTextFallbackKey]] (via [[resolveViaCredentialAliases]]'s [[NoProvider]] arm),
* matching `getPassword`'s real refusal to read plaintext conf when the flag is off.
*
- * `hadoopConf` must be a [[propagateBucketOptions]] result (the caller,
- * [[s3KeyDivergenceReason]], always passes one) so that `${...}` references embedded in any
- * alias resolve exactly like `S3AFileSystem#initialize`'s real propagate-then-resolve order.
+ * `hadoopConf` must be a [[propagateBucketOptions]] result (both callers pass one) so that
+ * `${...}` references embedded in any alias resolve exactly like `S3AFileSystem#initialize`'s
+ * real propagate-then-resolve order.
*/
private def hadoopLookupPasswordEffective(
hadoopConf: Configuration,
diff --git a/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/CometDeltaS3Suite.scala b/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/CometDeltaS3Suite.scala
index 4f758e2430e..e7b3e4b12b3 100644
--- a/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/CometDeltaS3Suite.scala
+++ b/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/CometDeltaS3Suite.scala
@@ -30,7 +30,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.delta.DeltaLog
import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
-import org.apache.comet.CometS3TestBase
+import org.apache.comet.{CometConf, CometS3TestBase, ExtendedExplainInfo}
/**
* MinIO-backed integration coverage for multi-bucket Delta shapes: a real two-bucket shallow
@@ -142,6 +142,68 @@ class CometDeltaS3Suite extends CometDeltaTestBase with CometS3TestBase with Log
private def tablePath(bucket: String, relPath: String): String = s"s3a://$bucket/$relPath"
+ if (supportsS3EncryptionContext) {
+ test("encryption context on external S3 data files falls back before native reads") {
+ assume(dockerAvailable, "Docker is not available; skipping MinIO-backed Delta test")
+ withTempPath { dir =>
+ val sourcePath = tablePath(testBucketName, "context-source")
+ val clonePath = dir.getAbsolutePath
+ spark.range(0, 4).coalesce(1).write.format("delta").save(sourcePath)
+ spark.sql(s"CREATE TABLE delta.`$clonePath` SHALLOW CLONE delta.`$sourcePath`")
+ val files = DeltaLog.forTable(spark, clonePath).update().allFiles.collect()
+ assert(files.nonEmpty && files.forall(_.path.startsWith(s"s3a://$testBucketName/")))
+ checkDeltaNativeScanAnswer(spark.read.format("delta").load(clonePath))
+ val conf = spark.sparkContext.hadoopConfiguration
+ val contextKeys = Seq(
+ "fs.s3a.encryption.context",
+ s"fs.s3a.bucket.$testBucketName.encryption.context",
+ s"fs.s3a.bucket.$testBucketName.fs.s3a.encryption.context")
+ val keys = contextKeys :+ "fs.s3a.impl.disable.cache"
+ val original = keys.map(key => key -> Option(conf.getRaw(key))).toMap
+ try {
+ // A filesystem cached by the clone would hide Hadoop's malformed-context failure.
+ conf.set("fs.s3a.impl.disable.cache", "true")
+ for (key <- contextKeys;
+ value <- Seq("synthetic-malformed-pair", "purpose=synthetic")) {
+ contextKeys.foreach(conf.unset)
+ conf.set(key, value)
+ Seq(false, true).foreach { enabled =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> enabled.toString) {
+ val frame = spark.read.format("delta").load(clonePath)
+ assert(
+ deltaNativeScans(frame).isEmpty,
+ frame.queryExecution.executedPlan.toString)
+ if (enabled) {
+ val explain =
+ new ExtendedExplainInfo().generateExtendedInfo(
+ frame.queryExecution.executedPlan)
+ assert(explain.contains("fs.s3a.encryption.context"), explain)
+ assert(!explain.contains(value), explain)
+ }
+ if (value == "purpose=synthetic") {
+ assert(frame.collect().map(_.getLong(0)).sorted.toSeq == Seq(0L, 1L, 2L, 3L))
+ } else {
+ val error = intercept[Exception] { frame.collect() }
+ var root: Throwable = error
+ while (root.getCause != null) root = root.getCause
+ assert(root.isInstanceOf[IllegalArgumentException], error.toString)
+ assert(
+ root.getStackTrace.exists(
+ _.toString.contains("S3AEncryption.getS3EncryptionContext")))
+ }
+ }
+ }
+ }
+ } finally {
+ original.foreach {
+ case (key, Some(value)) => conf.set(key, value)
+ case (key, None) => conf.unset(key)
+ }
+ }
+ }
+ }
+ }
+
test("shallow clone across buckets + append declines with the multi-store reason") {
assume(dockerAvailable, "Docker is not available; skipping MinIO-backed Delta test")
diff --git a/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/CometDeltaTestBase.scala b/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/CometDeltaTestBase.scala
index 50fbcc23fe2..f57cd54b4db 100644
--- a/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/CometDeltaTestBase.scala
+++ b/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/CometDeltaTestBase.scala
@@ -19,6 +19,7 @@
package org.apache.comet.contrib.delta
+import org.apache.hadoop.fs.s3a.Constants
import org.apache.spark.SparkConf
import org.apache.spark.sql.{CometTestBase, DataFrame}
import org.apache.spark.sql.execution.SparkPlan
@@ -29,6 +30,9 @@ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
*/
abstract class CometDeltaTestBase extends CometTestBase with AdaptiveSparkPlanHelper {
+ protected def supportsS3EncryptionContext: Boolean =
+ classOf[Constants].getFields.exists(_.getName == "S3_ENCRYPTION_CONTEXT")
+
override protected def sparkConf: SparkConf = {
val conf = super.sparkConf
conf.set("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
diff --git a/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala b/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala
index 1d240ba3e22..0e4231504a6 100644
--- a/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala
+++ b/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala
@@ -21,6 +21,7 @@ package org.apache.comet.contrib.delta
import java.io.{File, IOException}
import java.net.URI
+import java.nio.charset.StandardCharsets.UTF_8
import java.nio.file.Files
import java.util.{Locale, UUID}
import java.util.concurrent.atomic.AtomicBoolean
@@ -32,6 +33,7 @@ import org.apache.hadoop.fs.{Path, RawLocalFileSystem}
import org.apache.hadoop.fs.s3a.{S3AFileSystem, S3AUtils}
import org.apache.hadoop.security.alias.CredentialProviderFactory
import org.apache.spark.sql.comet.CometDeltaNativeScanExec
+import org.apache.spark.sql.delta.DeltaLog
import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
import org.apache.spark.sql.execution.SparkPlan
@@ -46,6 +48,72 @@ import org.apache.comet.rules.CometScanRule
*/
class DeltaScanContribSuite extends CometDeltaTestBase {
+ if (supportsS3EncryptionContext) {
+ // This pins metadata-setup parity; the MinIO local-clone test exercises the admission gate.
+ test("malformed S3 encryption context fails on the driver before Delta scan planning") {
+ val conf = spark.sparkContext.hadoopConfiguration
+ val settings = Map(
+ "fs.s3a.impl" -> classOf[S3AFileSystem].getName,
+ "fs.s3a.impl.disable.cache" -> "true",
+ "fs.s3a.endpoint" -> "http://127.0.0.1:1",
+ "fs.s3a.endpoint.region" -> "us-east-1",
+ "fs.s3a.path.style.access" -> "true",
+ "fs.s3a.connection.ssl.enabled" -> "false",
+ "fs.s3a.bucket.probe" -> "0",
+ "fs.s3a.aws.credentials.provider" ->
+ "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider",
+ "fs.s3a.access.key" -> "synthetic-access",
+ "fs.s3a.secret.key" -> "synthetic-secret",
+ "fs.s3a.encryption.algorithm" -> "",
+ "fs.s3a.encryption.context" -> "synthetic-malformed-pair")
+ val original = settings.keys.map(key => key -> Option(conf.getRaw(key))).toMap
+ try {
+ settings.foreach { case (key, value) => conf.set(key, value) }
+ for (algorithm <- Seq("", "AES256", "SSE-KMS", "DSSE-KMS")) {
+ conf.set("fs.s3a.encryption.algorithm", algorithm)
+ val valid = new Configuration(conf)
+ valid.set("fs.s3a.encryption.context", "purpose=synthetic")
+ val fs = new S3AFileSystem()
+ try {
+ fs.initialize(new java.net.URI("s3a://valid-context/"), valid)
+ assert(fs.getUri.getHost == "valid-context")
+ } finally {
+ fs.close()
+ }
+ val errors = scala.collection.mutable.ArrayBuffer.empty[(String, String, Seq[String])]
+ Seq(false, true).foreach { enabled =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> enabled.toString) {
+ val group = s"malformed-context-${UUID.randomUUID()}"
+ spark.sparkContext.setJobGroup(group, group)
+ try {
+ val error = intercept[IllegalArgumentException] {
+ spark.read.format("delta").load("s3a://malformed-context/table").collect()
+ }
+ val frames = error.getStackTrace.map(_.toString).toSeq
+ assert(frames.exists(_.contains("S3AEncryption.getS3EncryptionContext")))
+ assert(frames.exists(_.contains("S3AFileSystem.initialize")))
+ assert(spark.sparkContext.statusTracker.getJobIdsForGroup(group).isEmpty)
+ errors += ((
+ error.getClass.getName,
+ error.getMessage,
+ frames.filter(_.startsWith("org.apache.hadoop."))))
+ } finally {
+ spark.sparkContext.clearJobGroup()
+ }
+ }
+ }
+ assert(errors.size == 2)
+ assert(errors.head == errors.last)
+ }
+ } finally {
+ original.foreach {
+ case (key, Some(value)) => conf.set(key, value)
+ case (key, None) => conf.unset(key)
+ }
+ }
+ }
+ }
+
test("contrib is inert when comet exec is disabled") {
// The COMET_EXEC_ENABLED gate lives in DeltaScanContrib.tryTransformV1; this pins it there.
withTempPath { dir =>
@@ -2050,6 +2118,168 @@ class DeltaScanContribSuite extends CometDeltaTestBase {
assert(!reason.get.contains("realSecretValue"))
}
+ test("S3 deletion-vector context gates the real claim path after encryption algorithms") {
+ withTempPath { dir =>
+ val path = dir.getAbsolutePath
+ spark
+ .range(4)
+ .coalesce(1)
+ .write
+ .format("delta")
+ .option("delta.enableDeletionVectors", "true")
+ .save(path)
+ val log = DeltaLog.forTable(spark, path)
+ val snapshot = log.update()
+ val files = snapshot.allFiles.collect()
+ assert(files.length == 1)
+ val dv = DeletionVectorDescriptor
+ .onDiskWithAbsolutePath("s3a://context-dv-bucket/deletion_vector.bin", 40, 1)
+ val replaced = files.head.copy(deletionVector = dv).withoutTightBoundStats
+ val commit = new File(dir, f"_delta_log/${snapshot.version + 1}%020d.json")
+ Files.write(
+ commit.toPath,
+ (files.head.remove.json + "\n" + replaced.json + "\n").getBytes(UTF_8))
+ val selected = log.update().allFiles.collect()
+ assert(selected.length == 1)
+ assert(selected.head.absolutePath(log).toUri.getScheme == "file")
+ assert(
+ selected.head.deletionVector
+ .absolutePath(new Path(path))
+ .toUri
+ .getHost == "context-dv-bucket")
+
+ val conf = spark.sparkContext.hadoopConfiguration
+ val contextKey = "fs.s3a.bucket.context-dv-bucket.encryption.context"
+ val algorithmKey = "fs.s3a.bucket.context-dv-bucket.encryption.algorithm"
+ val original = Seq(contextKey, algorithmKey).map(key => key -> Option(conf.getRaw(key)))
+ try {
+ conf.unset(contextKey)
+ conf.unset(algorithmKey)
+ // Plan only: the descriptor intentionally names no real sidecar and must never be read.
+ val baseline = spark.read.format("delta").load(path)
+ assert(deltaNativeScans(baseline).nonEmpty, baseline.queryExecution.executedPlan.toString)
+
+ conf.set(contextKey, "synthetic-malformed-pair")
+ val declined = spark.read.format("delta").load(path)
+ assert(deltaNativeScans(declined).isEmpty)
+ val reasons =
+ new ExtendedExplainInfo().getFallbackReasons(declined.queryExecution.executedPlan)
+ assert(
+ reasons.exists(_.contains("does not support fs.s3a.encryption.context")),
+ reasons.toString)
+ Seq("SSE-C", "CSE-KMS", "CSE-CUSTOM").foreach { algorithm =>
+ conf.set(algorithmKey, algorithm)
+ val encrypted = spark.read.format("delta").load(path)
+ assert(deltaNativeScans(encrypted).isEmpty)
+ val encryptionReasons =
+ new ExtendedExplainInfo().getFallbackReasons(encrypted.queryExecution.executedPlan)
+ assert(
+ encryptionReasons.exists(_.contains(s"fs.s3a.encryption.algorithm=$algorithm")),
+ encryptionReasons.toString)
+ assert(
+ !encryptionReasons.exists(_.contains("does not support fs.s3a.encryption.context")))
+ }
+ } finally {
+ original.foreach {
+ case (key, Some(value)) => conf.set(key, value)
+ case (key, None) => conf.unset(key)
+ }
+ }
+ }
+ }
+
+ test("unsupportedEncryptionContextReason passes absent/empty and non-S3 contexts") {
+ val conf = new Configuration(false)
+ assert(DeltaScanSupport.unsupportedEncryptionContextReason(conf, DataBucketUris).isEmpty)
+ conf.set("fs.s3a.encryption.context", "")
+ assert(DeltaScanSupport.unsupportedEncryptionContextReason(conf, DataBucketUris).isEmpty)
+ conf.set("fs.s3a.encryption.context", "synthetic-malformed-pair")
+ assert(
+ DeltaScanSupport
+ .unsupportedEncryptionContextReason(conf, Seq(new URI("file:///tmp/data.parquet")))
+ .isEmpty)
+ }
+
+ test(
+ "unsupportedEncryptionContextReason declines global/short/long values without printing them") {
+ val keys = Seq(
+ "fs.s3a.encryption.context",
+ "fs.s3a.bucket.data-bucket.encryption.context",
+ "fs.s3a.bucket.data-bucket.fs.s3a.encryption.context")
+ for (key <- keys; value <- Seq("purpose=synthetic", "synthetic-malformed-pair")) {
+ val conf = new Configuration(false)
+ conf.set(key, value)
+ val reason = DeltaScanSupport.unsupportedEncryptionContextReason(conf, DataBucketUris)
+ assert(reason.exists(_.contains("fs.s3a.encryption.context")))
+ assert(reason.exists(_.contains("data-bucket")))
+ assert(!reason.get.contains(value))
+ }
+ val unrelated = new Configuration(false)
+ unrelated.set("fs.s3a.bucket.other-bucket.encryption.context", "synthetic-malformed-pair")
+ assert(DeltaScanSupport.unsupportedEncryptionContextReason(unrelated, DataBucketUris).isEmpty)
+ }
+
+ test(
+ "unsupportedEncryptionContextReason resolves propagated references and preserves fallback") {
+ val conf = new Configuration(false)
+ conf.set("fs.s3a.encryption.context", s"$${fs.s3a.context.reference}")
+ conf.set("fs.s3a.context.reference", "")
+ conf.set("fs.s3a.bucket.data-bucket.context.reference", "synthetic-malformed-pair")
+ val reason = DeltaScanSupport.unsupportedEncryptionContextReason(conf, DataBucketUris)
+ assert(reason.isDefined)
+ assert(!reason.get.contains("synthetic-malformed-pair"))
+ conf.setBoolean("hadoop.security.credential.clear-text-fallback", false)
+ assert(DeltaScanSupport.unsupportedEncryptionContextReason(conf, DataBucketUris).isEmpty)
+ val unresolved = new Configuration(false)
+ unresolved.set("fs.s3a.encryption.context", s"$${unresolved.synthetic.context}")
+ val unresolvedReason =
+ DeltaScanSupport.unsupportedEncryptionContextReason(unresolved, DataBucketUris)
+ assert(unresolvedReason.isDefined)
+ assert(!unresolvedReason.get.contains("unresolved.synthetic.context"))
+ }
+
+ test("unsupportedEncryptionContextReason finds context stored only in JCEKS bucket aliases") {
+ val keys = Seq(
+ "fs.s3a.encryption.context",
+ "fs.s3a.bucket.data-bucket.encryption.context",
+ "fs.s3a.bucket.data-bucket.fs.s3a.encryption.context")
+ keys.foreach { key =>
+ withJceks(Map(key -> "purpose=synthetic")) { conf =>
+ assert(conf.get(key) == null)
+ conf.setBoolean("hadoop.security.credential.clear-text-fallback", false)
+ val reason = DeltaScanSupport.unsupportedEncryptionContextReason(conf, DataBucketUris)
+ assert(reason.isDefined)
+ assert(!reason.get.contains("purpose=synthetic"))
+ }
+ }
+ }
+
+ test(
+ "unsupportedEncryptionContextReason declines unverifiable providers without exposing paths") {
+ for ((key, value) <- Seq(
+ "fs.s3a.security.credential.provider.path" -> "jceks://file/no/synthetic-store.jceks",
+ "hadoop.security.credential.provider.path" -> "jceks://s3a@secret-bucket/secret.jceks")) {
+ val conf = new Configuration(false)
+ conf.set(key, value)
+ val reason = DeltaScanSupport.unsupportedEncryptionContextReason(conf, DataBucketUris)
+ assert(reason.isDefined)
+ assert(!reason.get.contains(value))
+ }
+ val corruptFile = File.createTempFile("comet-context-corrupt", ".jceks")
+ try {
+ Files.write(corruptFile.toPath, Array[Byte](1, 2, 3))
+ val conf = new Configuration(false)
+ conf.set(
+ "hadoop.security.credential.provider.path",
+ "jceks://file" + corruptFile.getAbsolutePath)
+ val reason = DeltaScanSupport.unsupportedEncryptionContextReason(conf, DataBucketUris)
+ assert(reason.isDefined)
+ assert(!reason.get.contains(corruptFile.getAbsolutePath))
+ } finally {
+ corruptFile.delete()
+ }
+ }
+
test(
"unsupportedEncryptionAlgorithmReason declines a bucket configured with global SSE-C, " +
"naming the algorithm key and the algorithm but never the customer-provided key value") {
@@ -2894,6 +3124,10 @@ class DeltaScanContribSuite extends CometDeltaTestBase {
"effectiveEncryptionAlgorithm's doc)"),
"fs.s3a.server-side-encryption.key" ->
"deprecated alias of fs.s3a.encryption.key, same reasoning",
+ "fs.s3a.encryption.context" ->
+ ("handled by unsupportedEncryptionContextReason: declines non-empty or unverifiable " +
+ "Hadoop-effective context for every S3 data/DV bucket, including when a local " +
+ "Delta log did not initialize that bucket's S3A filesystem"),
"fs.s3a.encryption.cse.kms.region" ->
("CSE tuning, newer Hadoop only: consulted solely when the algorithm resolves to " +
"a CSE variant, and the allowlist gate declines every CSE algorithm outright, " +
diff --git a/dev/ci/check-ci-config.py b/dev/ci/check-ci-config.py
index dfdbb406427..45c55f50c77 100644
--- a/dev/ci/check-ci-config.py
+++ b/dev/ci/check-ci-config.py
@@ -185,6 +185,7 @@
"delta_3_5",
"delta_4_0",
"delta_4_1",
+ "delta_4_2",
},
),
(["contrib/delta-spark/README.md"], set()),
@@ -233,13 +234,14 @@
"delta_3_5",
"delta_4_0",
"delta_4_1",
+ "delta_4_2",
"build_linux_all_profiles",
}
# One label opts a pull request into every nightly Iceberg version. Iceberg
# 1.11 is in the PR tier, so it needs no label.
ICEBERG_OPT_IN = {"iceberg_1_9", "iceberg_1_10"}
# Likewise one label for every Delta profile.
-DELTA_OPT_IN = {"delta_3_5", "delta_4_0", "delta_4_1"}
+DELTA_OPT_IN = {"delta_3_5", "delta_4_0", "delta_4_1", "delta_4_2"}
ALL_JOBS = QUEUE_TIER | NIGHTLY_TIER | {"docs"}
assert not QUEUE_TIER & NIGHTLY_TIER, "a job is queue or nightly, never both"
diff --git a/dev/ci/compute-changes.py b/dev/ci/compute-changes.py
index 42d4b859a37..0e37b231cd9 100644
--- a/dev/ci/compute-changes.py
+++ b/dev/ci/compute-changes.py
@@ -406,10 +406,11 @@
".github/actions/setup-builder/**",
],
# Same inputs as delta_3_5: the Delta contrib workflow is called once per
- # Spark profile, and the three calls share one path filter. Populated
+ # Spark profile, and the four calls share one path filter. Populated
# below, after the dict, so the lists cannot drift.
"delta_4_0": [],
"delta_4_1": [],
+ "delta_4_2": [],
}
FILTERS["spark_4_1_hive"] = FILTERS["spark_4_1"]
FILTERS["build_linux_full"] = FILTERS["build_linux"]
@@ -417,6 +418,7 @@
FILTERS["build_linux_all_profiles"] = FILTERS["build_linux"]
FILTERS["delta_4_0"] = FILTERS["delta_3_5"]
FILTERS["delta_4_1"] = FILTERS["delta_3_5"]
+FILTERS["delta_4_2"] = FILTERS["delta_3_5"]
# Which events may run each job, independent of the path filters above.
#
@@ -555,10 +557,11 @@
# The Delta contrib suites all run nightly. The PR tier's strict warnings
# job compiles the contrib against Spark 3.5, so a change that breaks its
# build on Spark 3.5 fails on its own pull request. One label opts a pull
- # request into all three.
+ # request into all four.
"delta_3_5": ["nightly", "label:run-delta-tests"],
"delta_4_0": ["nightly", "label:run-delta-tests"],
"delta_4_1": ["nightly", "label:run-delta-tests"],
+ "delta_4_2": ["nightly", "label:run-delta-tests"],
}
diff --git a/docs/source/contributor-guide/ci.md b/docs/source/contributor-guide/ci.md
index ca9b6d28dd6..774e8e783af 100644
--- a/docs/source/contributor-guide/ci.md
+++ b/docs/source/contributor-guide/ci.md
@@ -44,7 +44,7 @@ flowchart LR
PRTIER["PR tier
Linux build, lint, Rust tests
TPC-H / TPC-DS
Comet suites, Spark 4.1
Spark SQL, Spark 4.1 catalyst and sql_core
Iceberg 1.11"]
QUEUE["Queue tier, on top of the PR tier
Spark SQL, Spark 4.1 sql_hive
macOS build and Comet suites
Benchmark check, Delta gate
PyArrow UDF, Spark 4.0 / 4.1 / 4.2"]
- NIGHTLY["Nightly tier
Comet suites, Spark 3.5 / 4.0 / 4.2
Spark SQL, Spark 3.5 / 4.0 / 4.2
Iceberg 1.9 / 1.10
Delta contrib, Spark 3.5 / 4.0 / 4.1"]
+ NIGHTLY["Nightly tier
Comet suites, Spark 3.5 / 4.0 / 4.2
Spark SQL, Spark 3.5 / 4.0 / 4.2
Iceberg 1.9 / 1.10
Delta contrib, Spark 3.5 / 4.0 / 4.1 / 4.2"]
CACHE["Cache-refresh-only mode
the three cache-writing jobs, plus Lint"]
```
@@ -64,7 +64,7 @@ Suite by suite:
| Comet test suites, Spark 3.5 / 4.0 / 4.2 | with label | no | yes |
| Spark SQL tests, Spark 3.5 / 4.0 / 4.2 | with label | no | yes |
| Iceberg Spark SQL tests, Iceberg 1.9 / 1.10 | with label | no | yes |
-| Delta contrib tests, Spark 3.5 / 4.0 / 4.1 | with label | no | yes |
+| Delta contrib tests, Spark 3.5 / 4.0 / 4.1 / 4.2 | with label | no | yes |
The **PR tier** is the fast feedback loop while a change is being iterated on. The **queue
tier** is the authoritative gate: everything the PR tier runs plus the suites that must pass before
@@ -173,19 +173,19 @@ tier for you before the queue does, so a maintainer may ask you to reproduce a f
Each suite outside the PR tier has a label that runs it on a pull request:
-| Label | Runs |
-| -------------------------- | -------------------------------------------------- |
-| `run-macos-tests` | macOS build and Comet test suites |
-| `run-all-spark-profiles` | Comet test suites against Spark 3.5 / 4.0 / 4.2 |
-| `run-benchmark-check` | Benchmark compile and lint check |
-| `run-delta-build-gate` | Delta contrib build gate |
-| `run-pyarrow-udf-tests` | PyArrow UDF tests against Spark 4.0/4.1/4.2 |
-| `run-delta-tests` | Delta contrib tests against Spark 3.5, 4.0 and 4.1 |
-| `run-spark-4.1-hive-tests` | Spark SQL tests against Spark 4.1, sql_hive only |
-| `run-spark-3.5-tests` | Spark SQL tests against Spark 3.5 |
-| `run-spark-4.0-tests` | Spark SQL tests against Spark 4.0 |
-| `run-spark-4.2-tests` | Spark SQL tests against Spark 4.2 |
-| `run-iceberg-tests` | Iceberg Spark SQL tests against 1.9 and 1.10 |
+| Label | Runs |
+| -------------------------- | ------------------------------------------------------- |
+| `run-macos-tests` | macOS build and Comet test suites |
+| `run-all-spark-profiles` | Comet test suites against Spark 3.5 / 4.0 / 4.2 |
+| `run-benchmark-check` | Benchmark compile and lint check |
+| `run-delta-build-gate` | Delta contrib build gate |
+| `run-pyarrow-udf-tests` | PyArrow UDF tests against Spark 4.0/4.1/4.2 |
+| `run-delta-tests` | Delta contrib tests against Spark 3.5, 4.0, 4.1 and 4.2 |
+| `run-spark-4.1-hive-tests` | Spark SQL tests against Spark 4.1, sql_hive only |
+| `run-spark-3.5-tests` | Spark SQL tests against Spark 3.5 |
+| `run-spark-4.0-tests` | Spark SQL tests against Spark 4.0 |
+| `run-spark-4.2-tests` | Spark SQL tests against Spark 4.2 |
+| `run-iceberg-tests` | Iceberg Spark SQL tests against 1.9 and 1.10 |
For a queue-tier suite the label only brings the run forward; the queue would have run it anyway
before the change landed. For a nightly-tier suite the label is what gets a verdict _before_ the
@@ -232,7 +232,7 @@ does not cover. Some examples:
the 4.1 diff
- anything touching Iceberg reflection, the Iceberg scan or write path, or the Iceberg diffs
- a change to `contrib/delta-spark/` or to the scan and native code it builds on;
- `run-delta-tests` runs the Delta contrib suites against Spark 3.5, 4.0 and 4.1
+ `run-delta-tests` runs the Delta contrib suites against Spark 3.5, 4.0, 4.1 and 4.2
- native code with platform-specific behavior, or a dependency bump that changes what is
compiled on macOS
- a change to the benchmark sources under `spark/src/test/scala/org/apache/spark/sql/benchmark`
@@ -303,7 +303,7 @@ re-armed automatically.
`ci.yml` also runs on a schedule, at 06:00 UTC every day, against what landed on `main` since the
last successful scheduled run. That run executes only the nightly tier: the Comet test suites against the Spark
profiles other than 4.1, the Spark SQL suites for Spark 3.5, 4.0 and 4.2, the Iceberg suites for
-1.9 and 1.10, and the Delta contrib suites for Spark 3.5, 4.0 and 4.1, each only when the day's
+1.9 and 1.10, and the Delta contrib suites for Spark 3.5, 4.0, 4.1 and 4.2, each only when the day's
changes touched files it covers. The queue already ran
everything else against the same tree, so nothing in the queue tier is repeated. A day with no
merges, or with only documentation changes, runs nothing.
diff --git a/docs/source/user-guide/latest/delta.md b/docs/source/user-guide/latest/delta.md
index 2d8d7872b30..5d7b1a83919 100644
--- a/docs/source/user-guide/latest/delta.md
+++ b/docs/source/user-guide/latest/delta.md
@@ -55,6 +55,8 @@ onward, and timestamps from 1900-01-01T00:00:00Z onward (Spark's own
rebase cutoff). Disable `spark.comet.scan.delta.enabled` for such tables to
read them through Spark.
+When an effective `fs.s3a.encryption.context` is configured, Delta reads fall back to Spark, including valid context values. This conservative fallback applies on every supported profile, including older Hadoop versions that ignore the key. Hadoop 3.5 parses this KMS write-context setting during filesystem initialization. A Delta log can refer to data in another bucket, so reading the log alone does not validate that bucket's configuration. Falling back preserves Spark's initialization checks. The decline reason names the setting without printing its value.
+
## Configuration
- 4.3.1
+
+ delta-spark_4.2
+ 4.4.1
17
${java.version}