From d04b838a2d6e35237cff0bd36e396d6502f3199c Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Tue, 10 Nov 2026 08:37:57 -0300 Subject: [PATCH 1/6] feat(delta): pair Spark 4.2 with Delta 4.4.1 Use the Spark-specific Delta artifact and run its contrib suites in the nightly and opt-in CI lanes. Pin malformed encryption context failure before Delta scan planning while retaining encryption declines. --- .github/workflows/README.md | 24 ++++--- .github/workflows/ci.yml | 17 ++++- contrib/delta-spark/README.md | 8 ++- .../contrib/delta/DeltaScanContribSuite.scala | 67 +++++++++++++++++++ dev/ci/check-ci-config.py | 4 +- dev/ci/compute-changes.py | 7 +- docs/source/contributor-guide/ci.md | 34 +++++----- pom.xml | 7 +- 8 files changed, 129 insertions(+), 39 deletions(-) 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/contrib/delta-spark/README.md b/contrib/delta-spark/README.md index 89664d3d61a..6bbbce519e5 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 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..540f1d0552a 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 @@ -46,6 +46,69 @@ import org.apache.comet.rules.CometScanRule */ class DeltaScanContribSuite extends CometDeltaTestBase { + if (org.apache.spark.SPARK_VERSION.startsWith("4.2.")) { + 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 = Seq(false, true).map { 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) + ( + error.getClass, + error.getMessage, + frames.filter(_.startsWith("org.apache.hadoop."))) + } finally { + spark.sparkContext.clearJobGroup() + } + } + } + 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 => @@ -2894,6 +2957,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" -> + ("write-request KMS context, never read by the native scan; malformed values fail " + + "in S3AFileSystem.initialize during Delta metadata reads before scan planning, " + + "as covered by the malformed S3 encryption context driver test above"), "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/pom.xml b/pom.xml index c3d3912b537..dbd6bd56483 100644 --- a/pom.xml +++ b/pom.xml @@ -841,10 +841,9 @@ under the License. spark-4.x spark-4.1+ spark-4.2 - - 4.3.1 + + delta-spark_4.2 + 4.4.1 17 ${java.version} From a2a747d02779efe2746f115e1064e5e7603a732a Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Tue, 10 Nov 2026 08:48:15 -0300 Subject: [PATCH 2/6] test(delta): compare captured setup failures --- .../comet/contrib/delta/DeltaScanContribSuite.scala | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) 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 540f1d0552a..8a7adf70ff5 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 @@ -77,7 +77,8 @@ class DeltaScanContribSuite extends CometDeltaTestBase { } finally { fs.close() } - val errors = Seq(false, true).map { enabled => + 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) @@ -89,15 +90,16 @@ class DeltaScanContribSuite extends CometDeltaTestBase { assert(frames.exists(_.contains("S3AEncryption.getS3EncryptionContext"))) assert(frames.exists(_.contains("S3AFileSystem.initialize"))) assert(spark.sparkContext.statusTracker.getJobIdsForGroup(group).isEmpty) - ( - error.getClass, + errors += (( + error.getClass.getName, error.getMessage, - frames.filter(_.startsWith("org.apache.hadoop."))) + frames.filter(_.startsWith("org.apache.hadoop.")))) } finally { spark.sparkContext.clearJobGroup() } } } + assert(errors.size == 2) assert(errors.head == errors.last) } } finally { From 161744a341df281ae0d095697a90f28f2aad3370 Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Tue, 10 Nov 2026 09:26:15 -0300 Subject: [PATCH 3/6] fix(delta): decline S3 encryption contexts A local shallow clone can reference S3 data without initializing its filesystem during metadata reads. Malformed context then fails in Spark but was bypassed by the native reader. Fall back for effective context values and verify all alias forms with real MinIO reads. Keep values out of decline reasons. --- .github/workflows/delta_contrib_test.yml | 14 +-- contrib/delta-spark/README.md | 6 ++ .../contrib/delta/DeltaScanSupport.scala | 66 ++++++++++--- .../contrib/delta/CometDeltaS3Suite.scala | 62 +++++++++++- .../contrib/delta/DeltaScanContribSuite.scala | 98 ++++++++++++++++++- docs/source/user-guide/latest/delta.md | 2 + 6 files changed, 225 insertions(+), 23 deletions(-) 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 6bbbce519e5..8d26dd8177c 100644 --- a/contrib/delta-spark/README.md +++ b/contrib/delta-spark/README.md @@ -72,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 context is attached to KMS write requests, but Hadoop 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..bd4ffdb424b 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,66 @@ class CometDeltaS3Suite extends CometDeltaTestBase with CometS3TestBase with Log private def tablePath(bucket: String, relPath: String): String = s"s3a://$bucket/$relPath" + if (org.apache.spark.SPARK_VERSION.startsWith("4.2.")) { + 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 { + 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 = + 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/DeltaScanContribSuite.scala b/contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala index 8a7adf70ff5..86541f7293c 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 @@ -2115,6 +2115,98 @@ class DeltaScanContribSuite extends CometDeltaTestBase { assert(!reason.get.contains("realSecretValue")) } + 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", "${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", "${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") { @@ -2960,9 +3052,9 @@ class DeltaScanContribSuite extends CometDeltaTestBase { "fs.s3a.server-side-encryption.key" -> "deprecated alias of fs.s3a.encryption.key, same reasoning", "fs.s3a.encryption.context" -> - ("write-request KMS context, never read by the native scan; malformed values fail " + - "in S3AFileSystem.initialize during Delta metadata reads before scan planning, " + - "as covered by the malformed S3 encryption context driver test above"), + ("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/docs/source/user-guide/latest/delta.md b/docs/source/user-guide/latest/delta.md index 2d8d7872b30..2744007f347 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. Hadoop 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