From 1a5f85e3f25e5eca26ceedc88683c34ad1a07556 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Thu, 5 Nov 2026 10:51:16 -0700 Subject: [PATCH 1/4] chore: remove Spark 3.4 Maven profiles and stop running Spark 3.4 CI --- .github/workflows/README.md | 49 +++++-------- .github/workflows/ci.yml | 45 +----------- .github/workflows/pr_build_linux.yml | 3 - .github/workflows/publish_snapshot.yml | 3 +- .github/workflows/spark_sql_writer_tests.yml | 2 +- README.md | 2 +- dev/ci/check-ci-config.py | 61 ++-------------- dev/ci/compute-changes.py | 69 ++----------------- dev/ci/linux-test-profiles.py | 1 - dev/generate-release-docs.sh | 2 +- dev/regenerate-golden-files.sh | 13 ++-- dev/release/build-release-comet.sh | 2 - dev/release/verify-release-candidate.sh | 2 +- docs/build.sh | 2 +- docs/source/contributor-guide/ci.md | 43 +++--------- docs/source/contributor-guide/development.md | 6 +- .../contributor-guide/iceberg-spark-tests.md | 20 +++--- .../contributor-guide/release_process.md | 7 +- .../contributor-guide/spark-sql-tests.md | 27 ++------ .../latest/compatibility/expressions/index.md | 4 -- .../expressions/spark-3.4/index.md | 46 ------------- .../latest/compatibility/spark-versions.md | 19 ----- docs/source/user-guide/latest/installation.md | 14 +--- pom.xml | 41 ++--------- spark/pom.xml | 59 +--------------- 25 files changed, 82 insertions(+), 460 deletions(-) delete mode 100644 docs/source/user-guide/latest/compatibility/expressions/spark-3.4/index.md diff --git a/.github/workflows/README.md b/.github/workflows/README.md index d50ef7cd58a..ce4b3f26f8e 100644 --- a/.github/workflows/README.md +++ b/.github/workflows/README.md @@ -26,8 +26,8 @@ ruleset in `.asf.yaml`. That splits CI into three tiers: and one Iceberg version, both the default profile's. - **Nightly tier** (`nightly`): the regression sweep of everything else, once a day against `main` as it stands. The Comet test suites against the other - four Spark profiles, Spark SQL on Spark 3.5, 4.0 and 4.2, Iceberg - 1.8/1.9/1.10, and the Delta contrib suites on Spark 3.5, 4.0 and 4.1. See + 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 [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 @@ -35,16 +35,6 @@ into it early, listed in the diagram below. The Lint Java matrix compiles every Spark profile on every pull request, so a shim that fails to build is caught there; only the runtime suites wait for the queue or the nightly. -`spark_3_4` is in none of the tiers. Spark 3.4 is deprecated, so its Spark SQL -suite no longer gates a merge; it runs only when a pull request carries -`run-spark-3.4-tests`, or from a `workflow_dispatch`. Anyone who wants to -check a change against 3.4 can still do so, but note when that result starts -to block a merge. Adding the label fires a `labeled` event, and those runs -publish the advisory `Required Checks (label run)` name rather than the -required one, so a red 3.4 there changes nothing. It is the next push with -the label still applied that runs 3.4 under `Required Checks`, and with the -queue run gone that push is the only thing that makes a 3.4 failure blocking. - Heavy jobs have no `push` tier. The queue already tested the exact tree that lands, so re-running them on push to main would double the cost of every merge. Three jobs are still on `push`: `docs`, because it deploys to @@ -118,9 +108,8 @@ tiers partition the list and that the `pr` tier is exactly the default profile. delta_build_gate run-delta-build-gate pyarrow_udf_test run-pyarrow-udf-tests spark_4_1 run-spark-4.1-tests - label or dispatch only spark_4_1 sql_hive run-spark-4.1-hive-tests - ---------------------- iceberg_1_11 run-iceberg-tests - spark_3_4 run-spark-3.4-tests + spark_4_1 sql_hive run-spark-4.1-hive-tests + iceberg_1_11 run-iceberg-tests nightly tier, or PR with label ------------------------------ pr_build_linux run-all-spark-profiles @@ -128,7 +117,6 @@ tiers partition the list and that the `pr` tier is exactly the default profile. spark_3_5 run-spark-3.5-tests spark_4_0 run-spark-4.0-tests spark_4_2 run-spark-4.2-tests - iceberg_1_8 run-iceberg-tests iceberg_1_9 run-iceberg-tests iceberg_1_10 run-iceberg-tests delta_3_5 run-delta-tests @@ -176,11 +164,9 @@ tiers partition the list and that the `pr` tier is exactly the default profile. | `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` | merge group, **or** PR with `run-spark-4.1-tests`; the `sql_hive` shards alone with `run-spark-4.1-hive-tests` | Spark 4.1 sources | -| `spark_3_4` | PR with `run-spark-3.4-tests`, or dispatch | Spark 3.4 sources | | `spark_4_0` | nightly, **or** PR with `run-spark-4.0-tests` | Spark 4.0 sources | | `spark_4_2` | nightly, **or** PR with `run-spark-4.2-tests` | Spark 4.2 sources | | `iceberg_1_11` | merge group, **or** PR with `run-iceberg-tests` | Iceberg sources | -| `iceberg_1_8` | nightly, **or** PR with `run-iceberg-tests` | Iceberg sources | | `iceberg_1_9` | nightly, **or** PR with `run-iceberg-tests` | Iceberg sources | | `iceberg_1_10` | nightly, **or** PR with `run-iceberg-tests` | Iceberg sources | | `required_checks` | always, after every job above except `docs` | none (always runs) | @@ -194,8 +180,8 @@ safe to make a required check. A pull request whose base is a release branch (`branch-N.M`) is the one exception to the table: it runs every job the table puts in the PR, merge group or nightly tier, because a release branch has no merge queue and no -nightly to run the last two later. `docs` stays push-only and `spark_3_4` -still needs its label, so a `labeled` run there adds `spark_3_4` or nothing. +nightly to run the last two later. `docs` stays push-only, and a `labeled` +run there adds nothing. `changes` hands the base branch to `compute-changes.py` as `PR_BASE_REF`. `check-ci-config.py` pins that wiring, because a dropped variable reads as an empty string and would route the pull request as if it targeted `main`. @@ -289,7 +275,7 @@ Things that follow from that, and are easy to break: ### Label events `ci_label.yml` fires on `pull_request.types: [labeled]` and calls `ci.yml` -through `workflow_call`, so applying `run-spark-3.4-tests`, +through `workflow_call`, so applying `run-spark-3.5-tests`, `run-spark-4.0-tests` or `run-iceberg-tests` starts the job that label gates without needing a new push. GitHub cannot filter a `pull_request` trigger by label name, so **every** label added to a PR starts a run, including labels @@ -393,17 +379,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` | `spark_3_4`, `spark_3_5`, `spark_4_0`, `spark_4_1`, `spark_4_2` | -| `iceberg_spark_test_reusable.yml` | `iceberg_1_8`, `iceberg_1_9`, `iceberg_1_10`, `iceberg_1_11` | +| 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` | ## Changing what runs when diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0fc66aba7f4..c2e08a520c9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,7 +32,7 @@ # the default profile's. # schedule the nightly regression sweep of everything else: the Linux # test matrix against the other Spark profiles, Spark SQL on -# Spark 3.5, 4.0 and 4.2, and Iceberg 1.8/1.9/1.10, routed by +# Spark 3.5, 4.0 and 4.2, and Iceberg 1.9/1.10, routed by # the same path filters over what landed on main since the last # green nightly. A failure opens (or comments on) an issue # labelled `ci-nightly-failure`. @@ -44,9 +44,6 @@ # on the merge result. Opt-in `run-*` suites, pull requests from a branch of this # repository and pull requests to a release branch are unaffected. # -# Spark 3.4 is deprecated and sits outside all three tiers: it runs only when -# a pull request carries `run-spark-3.4-tests`, or from a manual dispatch. -# # Release branches (`branch-N.M`) get neither the queue nor the nightly: the # queue ruleset covers only the default branch, `push` below is main-only, and # GitHub fires `schedule` only on the default branch. So a pull request that @@ -290,13 +287,11 @@ jobs: delta_gate: ${{ steps.compute.outputs.delta_gate }} pyarrow_udf: ${{ steps.compute.outputs.pyarrow_udf }} docs: ${{ steps.compute.outputs.docs }} - spark_3_4: ${{ steps.compute.outputs.spark_3_4 }} spark_3_5: ${{ steps.compute.outputs.spark_3_5 }} spark_4_0: ${{ steps.compute.outputs.spark_4_0 }} spark_4_1: ${{ steps.compute.outputs.spark_4_1 }} spark_4_1_hive: ${{ steps.compute.outputs.spark_4_1_hive }} spark_4_2: ${{ steps.compute.outputs.spark_4_2 }} - iceberg_1_8: ${{ steps.compute.outputs.iceberg_1_8 }} iceberg_1_9: ${{ steps.compute.outputs.iceberg_1_9 }} iceberg_1_10: ${{ steps.compute.outputs.iceberg_1_10 }} iceberg_1_11: ${{ steps.compute.outputs.iceberg_1_11 }} @@ -485,29 +480,6 @@ jobs: contents: write uses: ./.github/workflows/docs.yaml - spark_3_4: - name: Spark SQL Tests (Spark 3.4) - needs: changes - # Spark 3.4 is deprecated, so this is the one test job the merge queue does - # not run: it needs the `run-spark-3.4-tests` label on a pull request, or a - # manual workflow_dispatch. - # - # It is still in `required_checks.needs`, but be precise about what that - # buys. Adding the label fires a `labeled` event, and on that event the - # aggregator deliberately publishes the advisory `Required Checks (label - # run)` name instead of the required one (see its comment below), so a red - # 3.4 there leaves the commit's existing verdict alone. 3.4 only counts - # toward the required `Required Checks` on a pull_request event that is not - # `labeled` -- a push with the label already applied. With no queue run left - # to catch it, that push is the only thing that turns a 3.4 failure into a - # merge blocker. - if: needs.changes.outputs.spark_3_4 == 'true' - uses: ./.github/workflows/spark_sql_test_reusable.yml - with: - spark-short: '3.4' - spark-full: '3.4.3' - java: 17 - spark_3_5: name: Spark SQL Tests (Spark 3.5) needs: changes @@ -602,19 +574,6 @@ jobs: spark-full: '4.2.0' java: 17 - iceberg_1_8: - name: Iceberg Spark SQL Tests (Iceberg 1.8) - needs: changes - # Nightly by default; PRs need the `run-iceberg-tests` label. - if: needs.changes.outputs.iceberg_1_8 == 'true' - uses: ./.github/workflows/iceberg_spark_test_reusable.yml - with: - iceberg-short: '1.8' - iceberg-full: '1.8.1' - spark-short: '3.4' - spark-full: '3.4.3' - java: 17 - iceberg_1_9: name: Iceberg Spark SQL Tests (Iceberg 1.9) needs: changes @@ -706,7 +665,6 @@ jobs: - pr_benchmark_check - delta_build_gate - pyarrow_udf_test - - spark_3_4 - spark_3_5 - delta_3_5 - delta_4_0 @@ -714,7 +672,6 @@ jobs: - spark_4_0 - spark_4_1 - spark_4_2 - - iceberg_1_8 - iceberg_1_9 - iceberg_1_10 - iceberg_1_11 diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index 7f70f81122c..ef593ff4ed5 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.github/workflows/pr_build_linux.yml @@ -164,9 +164,6 @@ jobs: strategy: matrix: profile: - - name: "Spark 3.4, JDK 17, Scala 2.12" - java_version: "17" - maven_opts: "-Pspark-3.4 -Pscala-2.12" - name: "Spark 3.5, JDK 17, Scala 2.12" java_version: "17" maven_opts: "-Pspark-3.5 -Pscala-2.12" diff --git a/.github/workflows/publish_snapshot.yml b/.github/workflows/publish_snapshot.yml index e5d8eeea1d8..667a344870b 100644 --- a/.github/workflows/publish_snapshot.yml +++ b/.github/workflows/publish_snapshot.yml @@ -241,11 +241,10 @@ jobs: done } - # One build per published variant. Spark 3.4 and 3.5 are published for + # One build per published variant. Spark 3.5 is published for # Scala 2.12, which downstream benchmarks still use; 4.0, 4.1 and 4.2 # default to Scala 2.13. variants=( - "-Pspark-3.4 -Pscala-2.12" "-Pspark-3.5 -Pscala-2.12" "-Pspark-4.0" "-Pspark-4.1" diff --git a/.github/workflows/spark_sql_writer_tests.yml b/.github/workflows/spark_sql_writer_tests.yml index 6b5fc7f6d65..6b729cc488b 100644 --- a/.github/workflows/spark_sql_writer_tests.yml +++ b/.github/workflows/spark_sql_writer_tests.yml @@ -68,7 +68,7 @@ jobs: run: | # Map each supported Spark minor version to its full version + JDK. # Mirrors ci.yml's per-version reusable invocations (default-on PR - # versions only; 3.4 and 4.0 are label-gated and not offered here). + # versions only; 4.0 is label-gated and not offered here). case "${{ inputs.spark-version }}" in 3.5) spark_full=3.5.9; java=17 ;; 4.1) spark_full=4.1.3; java=17 ;; diff --git a/README.md b/README.md index 63827189dc0..5d666f20cfb 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,7 @@ benefits of Comet's acceleration capabilities without disrupting your Spark appl ## Getting Started -Comet supports Apache Spark 3.4, 3.5, 4.0, 4.1, and 4.2. See the +Comet supports Apache Spark 3.5, 4.0, 4.1, and 4.2. See the [installation guide](https://datafusion.apache.org/comet/user-guide/installation.html) for the detailed version, Java, and Scala compatibility matrix. diff --git a/dev/ci/check-ci-config.py b/dev/ci/check-ci-config.py index 9823869bf8c..d4b87f5ceaf 100644 --- a/dev/ci/check-ci-config.py +++ b/dev/ci/check-ci-config.py @@ -120,13 +120,11 @@ "build_linux_all_profiles", "build_macos", "build_macos_full", - "spark_3_4", "spark_3_5", "spark_4_0", "spark_4_1", "spark_4_1_hive", "spark_4_2", - "iceberg_1_8", "iceberg_1_9", "iceberg_1_10", "iceberg_1_11", @@ -230,7 +228,6 @@ "spark_3_5", "spark_4_0", "spark_4_2", - "iceberg_1_8", "iceberg_1_9", "iceberg_1_10", "delta_3_5", @@ -238,18 +235,12 @@ "delta_4_1", "build_linux_all_profiles", } -# Spark 3.4 is deprecated and sits outside the queue and nightly tiers -# entirely: a label on a pull request, or a workflow_dispatch, and nothing -# else. Keeping it in its own set is what makes the `merge_group` and -# `schedule` cases below assert its absence rather than quietly accept it -# coming back. -SPARK_DEPRECATED = {"spark_3_4"} # One label opts a pull request into every Iceberg version, whichever tier # each sits in. -ICEBERG_OPT_IN = {"iceberg_1_8", "iceberg_1_9", "iceberg_1_10", "iceberg_1_11"} +ICEBERG_OPT_IN = {"iceberg_1_9", "iceberg_1_10", "iceberg_1_11"} # Likewise one label for every Delta profile. DELTA_OPT_IN = {"delta_3_5", "delta_4_0", "delta_4_1"} -ALL_JOBS = QUEUE_TIER | NIGHTLY_TIER | SPARK_DEPRECATED | {"docs"} +ALL_JOBS = QUEUE_TIER | NIGHTLY_TIER | {"docs"} assert not QUEUE_TIER & NIGHTLY_TIER, "a job is queue or nightly, never both" # The repository whose queue gates main; check_event_policy asserts this matches @@ -277,9 +268,8 @@ def fork_pr(action="synchronize", label=None, labels=(), base="main", repository # A manual run may exercise anything. ({"name": "workflow_dispatch"}, ALL_JOBS), # The merge queue is the authoritative gate: everything except the site - # deploy, which can only run once the commit is actually on main, the - # nightly tier, and the deprecated Spark 3.4 suite, which no longer gates - # a merge. + # deploy, which can only run once the commit is actually on main, and the + # nightly tier. ({"name": "merge_group"}, QUEUE_TIER), # The scheduled run is the nightly tier and nothing else. The queue already # ran everything in QUEUE_TIER against the tree that is now main, so a @@ -413,23 +403,6 @@ def fork_pr(action="synchronize", label=None, labels=(), base="main", repository }, {"spark_4_1_hive"}, ), - # An opt-in label present on a pushed commit adds just that suite. For the - # deprecated Spark 3.4 suite the label is the *only* way it ever runs on a - # pull request or the queue, so this case and the `labeled` one below are - # what keep it reachable at all. - ( - {"name": "pull_request", "action": "synchronize", "labels": ["run-spark-3.4-tests"]}, - PR_TIER | {"spark_3_4"}, - ), - ( - { - "name": "pull_request", - "action": "labeled", - "label": "run-spark-3.4-tests", - "labels": ["run-spark-3.4-tests"], - }, - {"spark_3_4"}, - ), # Spark 4.2 runs nightly, and its label brings that run forward onto a pull # request, the same shape as 3.5 and 4.0. It pulls in no other Spark version. ( @@ -481,14 +454,13 @@ def fork_pr(action="synchronize", label=None, labels=(), base="main", repository "name": "pull_request", "action": "labeled", "label": "dependencies", - "labels": ["dependencies", "run-spark-3.4-tests"], + "labels": ["dependencies", "run-spark-3.5-tests"], }, set(), ), # A pull request against a release branch runs the queue and nightly tiers # as well, because a release branch has no queue and no nightly to run them - # later. The site deploy still never runs from a pull request, and the - # deprecated Spark 3.4 suite still waits for its label. + # later. The site deploy still never runs from a pull request. ( {"name": "pull_request", "action": "opened", "labels": [], "base": "branch-1.1"}, QUEUE_TIER | NIGHTLY_TIER, @@ -497,27 +469,6 @@ def fork_pr(action="synchronize", label=None, labels=(), base="main", repository {"name": "pull_request", "action": "synchronize", "labels": [], "base": "branch-2.0"}, QUEUE_TIER | NIGHTLY_TIER, ), - ( - { - "name": "pull_request", - "action": "synchronize", - "labels": ["run-spark-3.4-tests"], - "base": "branch-1.1", - }, - QUEUE_TIER | NIGHTLY_TIER | SPARK_DEPRECATED, - ), - # There the commit run already covered everything a label gates except - # Spark 3.4, so a `labeled` run adds Spark 3.4 or nothing. - ( - { - "name": "pull_request", - "action": "labeled", - "label": "run-spark-3.4-tests", - "labels": ["run-spark-3.4-tests"], - "base": "branch-1.1", - }, - SPARK_DEPRECATED, - ), ( { "name": "pull_request", diff --git a/dev/ci/compute-changes.py b/dev/ci/compute-changes.py index f07abdb8329..4dc6eca5003 100644 --- a/dev/ci/compute-changes.py +++ b/dev/ci/compute-changes.py @@ -198,34 +198,6 @@ "spark/src/main/scala/org/apache/comet/expressions/**", "spark/src/main/spark-*/**", ], - "spark_3_4": [ - "native/**/src/**", - "native/**/Cargo.toml", - "native/Cargo.lock", - "common/src/main/**", - "common/pom.xml", - "spark/src/main/**", - "!spark/src/main/spark-3.5/**", - "!spark/src/main/spark-4.0/**", - "!spark/src/main/spark-4.1/**", - "!spark/src/main/spark-4.2/**", - "!spark/src/main/spark-4.x/**", - "!spark/src/main/scala/org/apache/comet/GenerateDocs.scala", - "spark/pom.xml", - "dev/diffs/3.4.3.diff", - "pom.xml", - "rust-toolchain.toml", - ".github/workflows/ci.yml", - ".github/workflows/spark_sql_test_reusable.yml", - "dev/ci/spark-sql-modules.py", - ".github/actions/setup-builder/**", - ".github/actions/setup-spark-builder/**", - ".github/actions/upload-artifact-retry/**", - ".github/actions/download-artifact-retry/**", - ".github/actions/maven-bootstrap/**", - ".mvn/**", - "mvnw", - ], "spark_3_5": [ "native/**/src/**", "native/**/Cargo.toml", @@ -343,31 +315,6 @@ # ci.yml folds the two outputs into the reusable workflow's `modules` # input. Populated below, after the dict, so the two lists cannot drift. "spark_4_1_hive": [], - "iceberg_1_8": [ - "native/**/src/**", - "native/**/Cargo.toml", - "native/Cargo.lock", - "common/src/main/**", - "common/pom.xml", - "spark/src/main/**", - "!spark/src/main/scala/org/apache/comet/GenerateDocs.scala", - "spark/pom.xml", - "dev/diffs/iceberg/**", - "pom.xml", - "rust-toolchain.toml", - ".github/workflows/ci.yml", - ".github/workflows/iceberg_spark_test_reusable.yml", - ".github/actions/setup-builder/**", - ".github/actions/setup-iceberg-builder/**", - "dev/ci/iceberg-test-shards.gradle", - "dev/ci/check-iceberg-shards.py", - "dev/ci/test-iceberg-shards.py", - ".github/actions/upload-artifact-retry/**", - ".github/actions/download-artifact-retry/**", - ".github/actions/maven-bootstrap/**", - ".mvn/**", - "mvnw", - ], "iceberg_1_9": [ "native/**/src/**", "native/**/Cargo.toml", @@ -514,8 +461,8 @@ # `schedule` only on the default branch. A tier there would not defer a suite, # it would drop it. And the tiers exist because of main's volume, which a # release branch does not have: branch-1.0 took 28 pull requests in its first -# two months. "push" jobs (the site deploy) and label-only jobs (Spark 3.4) -# keep their rules on a release branch. +# two months. "push" jobs (the site deploy) and any label-only job keep their +# rules on a release branch. POLICY = { # The one test job that also runs on push to main, and only because of # actions/cache scoping: a pull request can restore caches saved on its @@ -573,11 +520,6 @@ # docs deploys to asf-site, so it must not run from a pull request or from # the queue's throwaway branch. "docs": ["push"], - # Spark 3.4 is deprecated, so it is the one test job outside the queue - # tier: a failure there no longer blocks a merge. It stays runnable on - # demand -- the label on a pull request, or a workflow_dispatch -- so - # anyone who wants to check a change against 3.4 still can. - "spark_3_4": ["label:run-spark-3.4-tests"], # Spark 4.1 is the default build profile and the one Spark SQL suite the # queue runs; 3.5, 4.0 and 4.2 run nightly, or on a pull request with # their label. @@ -607,9 +549,8 @@ # until someone thinks to ask for it. "spark_4_2": ["nightly", "label:run-spark-4.2-tests"], # Same shape for Iceberg: 1.11 is the only Spark 4.1 coverage, so it is - # the one Iceberg version the queue runs; the three older versions run - # nightly. One label opts a pull request into all four. - "iceberg_1_8": ["nightly", "label:run-iceberg-tests"], + # the one Iceberg version the queue runs; the older versions run + # nightly. One label opts a pull request into all three. "iceberg_1_9": ["nightly", "label:run-iceberg-tests"], "iceberg_1_10": ["nightly", "label:run-iceberg-tests"], "iceberg_1_11": ["queue", "label:run-iceberg-tests"], @@ -727,7 +668,7 @@ def release_branch_allows(job, event): automatic = any(tier in tiers for tier in RELEASE_BRANCH_TIERS) # The opened/synchronize run already ran every automatic job at this # commit, so a `labeled` run adds only a job that nothing else runs there, - # which leaves the label-only Spark 3.4 suite. + # which leaves only the label-only jobs, of which there are currently none. if event.get("action") == "labeled": return not automatic and event.get("label") in gates return automatic or any(label in event.get("labels", []) for label in gates) diff --git a/dev/ci/linux-test-profiles.py b/dev/ci/linux-test-profiles.py index 741ef7d3b1e..25c51c87b0a 100644 --- a/dev/ci/linux-test-profiles.py +++ b/dev/ci/linux-test-profiles.py @@ -46,7 +46,6 @@ # `run-all-spark-profiles` label). The goal of the list is coverage of every # Java, Scala and Spark version without testing every combination. PROFILES = [ - {"name": "Spark 3.4, JDK 17, Scala 2.12", "java_version": "17", "maven_opts": "-Pspark-3.4 -Pscala-2.12", "tier": "nightly"}, {"name": "Spark 3.5, JDK 17, Scala 2.13", "java_version": "17", "maven_opts": "-Pspark-3.5 -Pscala-2.13", "tier": "nightly"}, {"name": "Spark 4.0, JDK 21", "java_version": "21", "maven_opts": "-Pspark-4.0", "tier": "nightly"}, # The default build profile, and the one a contributor builds locally. diff --git a/dev/generate-release-docs.sh b/dev/generate-release-docs.sh index 8ba19a875b7..2a214d6b683 100755 --- a/dev/generate-release-docs.sh +++ b/dev/generate-release-docs.sh @@ -57,7 +57,7 @@ TEMPLATE_DIR="${COMPAT_EXPR_DIR}/_category_template" # and a source-tree subdirectory under compatibility/expressions/. # Every run also rewrites the shared configs.md and expressions.md pages, so the last # entry decides their content. Keep the default build profile (spark-4.1) last. -SPARK_PROFILES=(spark-3.4 spark-3.5 spark-4.0 spark-4.2 spark-4.1) +SPARK_PROFILES=(spark-3.5 spark-4.0 spark-4.2 spark-4.1) echo "Compiling and generating documentation content..." for profile in "${SPARK_PROFILES[@]}"; do diff --git a/dev/regenerate-golden-files.sh b/dev/regenerate-golden-files.sh index 1feb88e60d5..c24f74b7322 100755 --- a/dev/regenerate-golden-files.sh +++ b/dev/regenerate-golden-files.sh @@ -22,7 +22,7 @@ # Usage: ./dev/regenerate-golden-files.sh [--spark-version ] # # Options: -# --spark-version Only regenerate for specified Spark version (3.4, 3.5, 4.0, 4.1, or 4.2) +# --spark-version Only regenerate for specified Spark version (3.5, 4.0, 4.1, or 4.2) # If not specified, regenerates for all versions. # # Examples: @@ -77,8 +77,9 @@ build_native() { # Regenerate golden files for a specific Spark version. CometPlanStabilitySuite's # afterAll() prunes any query directory whose contents match what its fallback # chain resolves to, so each version-suffixed directory ends up with only the -# queries whose plans diverge from the previous tier. The base (3.4) profile has -# no fallback chain, so the prune is a no-op there. +# queries whose plans diverge from the previous tier. The shared base tier was +# generated by the Spark 3.4 profile, which no longer exists, so this script does not +# regenerate it. regenerate_golden_files() { local spark_version=$1 @@ -113,7 +114,7 @@ main() { echo "Usage: $0 [--spark-version ]" echo "" echo "Options:" - echo " --spark-version Only regenerate for specified Spark version (3.4, 3.5, 4.0, 4.1, or 4.2)" + echo " --spark-version Only regenerate for specified Spark version (3.5, 4.0, 4.1, or 4.2)" echo " If not specified, regenerates for all versions." exit 0 ;; @@ -129,7 +130,7 @@ main() { if [ -n "$target_version" ]; then if [[ ! "$target_version" =~ ^(3\.4|3\.5|4\.0|4\.1|4\.2)$ ]]; then echo "[ERROR] Invalid Spark version: $target_version" - echo "[ERROR] Supported versions: 3.4, 3.5, 4.0, 4.1, 4.2" + echo "[ERROR] Supported versions: 3.5, 4.0, 4.1, 4.2" exit 1 fi fi @@ -149,7 +150,7 @@ main() { if [ -n "$target_version" ]; then versions=("$target_version") else - versions=("3.4" "3.5" "4.0" "4.1" "4.2") + versions=("3.5" "4.0" "4.1" "4.2") fi # Regenerate for each version diff --git a/dev/release/build-release-comet.sh b/dev/release/build-release-comet.sh index fcf181ba54e..36834dba3dd 100755 --- a/dev/release/build-release-comet.sh +++ b/dev/release/build-release-comet.sh @@ -237,8 +237,6 @@ pushd "$BUILD_DIR" LOCAL_REPO=$(mktemp -d /tmp/comet-staging-repo-XXXXX) -./mvnw "-Dmaven.repo.local=${LOCAL_REPO}" -P spark-3.4 -P scala-2.12 -DskipTests install -./mvnw "-Dmaven.repo.local=${LOCAL_REPO}" -P spark-3.4 -P scala-2.13 -DskipTests install ./mvnw "-Dmaven.repo.local=${LOCAL_REPO}" -P spark-3.5 -P scala-2.12 -DskipTests install ./mvnw "-Dmaven.repo.local=${LOCAL_REPO}" -P spark-3.5 -P scala-2.13 -DskipTests install # The spark-4.x profiles pin their own Scala 2.13.x patch versions to match the diff --git a/dev/release/verify-release-candidate.sh b/dev/release/verify-release-candidate.sh index df150835ee3..9f7e49aa3e3 100755 --- a/dev/release/verify-release-candidate.sh +++ b/dev/release/verify-release-candidate.sh @@ -111,7 +111,7 @@ test_source_distribution() { RUSTFLAGS="-Ctarget-cpu=native" cargo build --release popd # test with the latest supported version of Spark - ./mvnw verify -Prelease -DskipTests -P"spark-3.4" -Dmaven.gitcommitid.skip=true + ./mvnw verify -Prelease -DskipTests -P"spark-3.5" -Dmaven.gitcommitid.skip=true } TEST_SUCCESS=no diff --git a/docs/build.sh b/docs/build.sh index efaaa0f94ab..fa87486331c 100755 --- a/docs/build.sh +++ b/docs/build.sh @@ -53,7 +53,7 @@ TEMPLATE_DIR="${COMPAT_EXPR_DIR}/_category_template" # docs/source/user-guide/latest/compatibility/expressions/. # Every run also rewrites the shared configs.md and expressions.md pages, so the last # entry decides their content. Keep the default build profile (spark-4.1) last. -SPARK_PROFILES=(spark-3.4 spark-3.5 spark-4.0 spark-4.2 spark-4.1) +SPARK_PROFILES=(spark-3.5 spark-4.0 spark-4.2 spark-4.1) for profile in "${SPARK_PROFILES[@]}"; do echo "Generating compatibility pages for ${profile}..." diff --git a/docs/source/contributor-guide/ci.md b/docs/source/contributor-guide/ci.md index 9088ddbcf22..8d8a9485cec 100644 --- a/docs/source/contributor-guide/ci.md +++ b/docs/source/contributor-guide/ci.md @@ -37,7 +37,6 @@ flowchart LR PR([pull request]) --> PRTIER PR -. with label .-> QUEUE PR -. with label .-> NIGHTLY - PR -. with label .-> S34 MQ([merge queue]) --> PRTIER MQ --> QUEUE CRON([schedule
06:00 UTC]) --> NIGHTLY @@ -45,8 +44,7 @@ flowchart LR PRTIER["PR tier
Linux build, lint, Rust tests
TPC-H / TPC-DS
Comet suites, Spark 4.1"] QUEUE["Queue tier, on top of the PR tier
Spark SQL, Spark 4.1
Iceberg 1.11
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.4 / 3.5 / 4.0 / 4.2
Spark SQL, Spark 3.5 / 4.0 / 4.2
Iceberg 1.8 / 1.9 / 1.10
Delta contrib, Spark 3.5 / 4.0 / 4.1"] - S34["Neither tier
Spark SQL, Spark 3.4"] + 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"] CACHE["Cache-refresh-only mode
the three cache-writing jobs, plus Lint"] ``` @@ -63,11 +61,10 @@ Suite by suite: | Benchmark compile and lint check | with label | yes | no | | Delta contrib build gate | with label | yes | no | | PyArrow UDF tests, Spark 4.0 / 4.1 / 4.2 | with label | yes | no | -| Comet test suites, Spark 3.4 / 3.5 / 4.0 / 4.2 | with label | no | yes | +| 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.8 / 1.9 / 1.10 | 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 | -| Spark SQL tests, Spark 3.4 | with label | no | no | 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 @@ -90,16 +87,6 @@ runner-minutes a push and about 40. If you add a job to `pr_build_linux.yml`, gi `if: ${{ !inputs.cache-refresh-only }}` unless it writes a cache that `main` needs; `dev/ci/check-ci-config.py` fails the build if you forget. -Spark 3.4 is the one suite in neither tier. [Spark 3.4 support is deprecated](../user-guide/latest/compatibility/spark-versions.md#spark-34), -so its Spark SQL suite no longer gates a merge. It remains available on demand: apply the -`run-spark-3.4-tests` label to run it against a pull request, or trigger `ci.yml` from the Actions -page with **Run workflow**, which runs every suite regardless of tier. A contributor touching Spark -3.4 code can still get a verdict before merging, but applying the label is only half of that: the -run it starts is advisory. Label the pull request and then push, and Spark 3.4 joins the required -verdict like any other suite. See -[Opting a pull request into a suite the PR tier skips](#opting-a-pull-request-into-a-suite-the-pr-tier-skips) -for the mechanics. - For a pull request from a fork, the pull request column's "yes" rows run on your fork rather than here; see [Pull requests from forks](#pull-requests-from-forks). @@ -184,14 +171,13 @@ 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.4 / 3.5 / 4.0 / 4.2 | +| `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-tests` | Spark SQL tests against Spark 4.1, every module | | `run-spark-4.1-hive-tests` | Spark SQL tests against Spark 4.1, sql_hive only | -| `run-spark-3.4-tests` | Spark SQL tests against Spark 3.4 | | `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 | @@ -199,8 +185,7 @@ Each suite outside the PR tier has a label that runs it on a pull request: 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 -change lands: without it, the first run is the nightly after the merge. For Spark 3.4 the label is -the only way the suite runs at all. +change lands: without it, the first run is the nightly after the merge. Apply a label from the pull request sidebar, or from the command line: @@ -216,12 +201,12 @@ read alongside the commit run without replacing it. For a queue-tier suite, that separate name costs nothing: the merge queue runs the suite again before the change lands, so a failure a label run surfaced still blocks the merge later. A -nightly-tier suite and Spark 3.4 have no queue run behind them. A red `Required Checks (label run)` +nightly-tier suite has no queue run behind it. A red `Required Checks (label run)` leaves an earlier green `Required Checks` in place and the pull request mergeable, so treat a label run as feedback to read, not as a gate. For one of those suites to count toward the required verdict the label has to already be on the pull request when a commit is pushed, which is what the next push gives you, since the label stays applied. A nightly-tier regression that does slip -through is caught by the next nightly run and filed as an issue; a Spark 3.4 one is not. +through is caught by the next nightly run and filed as an issue. The label stays on the pull request, so every later push runs the suite as part of the normal PR run. Remove the label once it has served its purpose. To re-run the suite at the same commit, @@ -233,7 +218,7 @@ does not cover. Some examples: - a change to the serde, the planner, or a native operator, where the Comet test suites pass but Spark's own SQL suite is the thing that would catch a behavior difference; `run-spark-4.1-tests` runs it against the default profile -- code under `spark/src/main/spark-3.4/`, `spark-3.5/`, `spark-4.0/` or the shared `spark-3.x/` +- code under `spark/src/main/spark-3.5/`, `spark-4.0/` or the shared `spark-3.x/` directory, or any change to `CometExprShim` and friends; `run-all-spark-profiles` runs the Comet test suites against every Spark version rather than 4.1 alone (the Lint Java matrix already compiles every Spark profile on every pull request, so this is for runtime differences) @@ -248,11 +233,6 @@ does not cover. Some examples: - a change to the benchmark sources under `spark/src/test/scala/org/apache/spark/sql/benchmark` or to `native/*/benches` -Reach for `run-spark-3.4-tests` deliberately, since nothing else will run it: a change to -`spark/src/main/spark-3.4/`, to `dev/diffs/3.4.3.diff`, or to shared code whose Spark 3.4 behavior -you are unsure of. Without the label, a Spark 3.4 regression will not be caught before the change -lands. - Labels that gate nothing, such as `dependencies` or the type labels, also start a run. That run executes nothing and finishes in a minute. It is a consequence of GitHub firing the `labeled` event for every label; see the workflows README for why it is handled this way. @@ -415,7 +395,7 @@ gh run view "$run" --repo apache/datafusion-comet --json jobs \ A healthy run over a day of normal merges reports about 53 successes — nine Spark SQL shards for each of 3.5, 4.0 and 4.2, plus the Iceberg shards — with the handful of skips being the suites that -belong to the queue tier rather than the nightly one — Spark 3.4, Spark 4.1 +belong to the queue tier rather than the nightly one — Spark 4.1 and Iceberg 1.11. If everything is skipped, open the run's `Detect changes` job: it logs the `Nightly base:` commit it diffed against and the list of changed files, which is enough to tell a genuinely quiet day from a base that has drifted. @@ -431,9 +411,8 @@ On `main`, the tiers hold the expensive suites back for the queue and the nightl test every change. A release branch has neither, so holding a suite back there would mean never running it. A pull request that targets a release branch, such as a backport, therefore runs the PR, queue and nightly tiers together. Release branches get few pull requests, so this costs little. The -path filters still apply, so a documentation-only change runs none of the heavy suites. The Spark -SQL suite for Spark 3.4 still needs its `run-spark-3.4-tests` label, and the other `run-*` labels -have nothing to add there. +path filters still apply, so a documentation-only change runs none of the heavy suites. The +`run-*` labels have nothing to add there. A release branch runs the workflow files committed on it, so a change to these rules on `main` reaches a release branch only if it is backported. `branch-1.0` predates the tiers and follows its diff --git a/docs/source/contributor-guide/development.md b/docs/source/contributor-guide/development.md index 8efec3ea7c3..16a458d180b 100644 --- a/docs/source/contributor-guide/development.md +++ b/docs/source/contributor-guide/development.md @@ -447,7 +447,7 @@ from a clean IntelliJ configuration: 6. Open the repository root as a new project in IntelliJ IDEA. 7. In the Maven tool window, enable the `spark-4.0` and `jdk17` profiles. Disable conflicting Spark - version profiles such as `spark-3.4` or `spark-3.5` if IntelliJ selected them. + version profiles such as `spark-3.5` if IntelliJ selected them. 8. Re-import or reload the Maven project. 9. Build the project in IntelliJ. @@ -543,7 +543,6 @@ The tests can be run with: ```sh export SPARK_HOME=`pwd` -./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV1_4_PlanStabilitySuite" -Pspark-3.4 -nsu test ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV1_4_PlanStabilitySuite" -Pspark-3.5 -nsu test ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV1_4_PlanStabilitySuite" -Pspark-4.0 -nsu test ``` @@ -552,7 +551,6 @@ and ```sh export SPARK_HOME=`pwd` -./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV2_7_PlanStabilitySuite" -Pspark-3.4 -nsu test ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV2_7_PlanStabilitySuite" -Pspark-3.5 -nsu test ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV2_7_PlanStabilitySuite" -Pspark-4.0 -nsu test ``` @@ -562,7 +560,6 @@ To regenerate the golden files, you can run the following commands. ```sh export SPARK_HOME=`pwd` -SPARK_GENERATE_GOLDEN_FILES=1 ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV1_4_PlanStabilitySuite" -Pspark-3.4 -nsu test SPARK_GENERATE_GOLDEN_FILES=1 ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV1_4_PlanStabilitySuite" -Pspark-3.5 -nsu test SPARK_GENERATE_GOLDEN_FILES=1 ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV1_4_PlanStabilitySuite" -Pspark-4.0 -nsu test ``` @@ -571,7 +568,6 @@ and ```sh export SPARK_HOME=`pwd` -SPARK_GENERATE_GOLDEN_FILES=1 ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV2_7_PlanStabilitySuite" -Pspark-3.4 -nsu test SPARK_GENERATE_GOLDEN_FILES=1 ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV2_7_PlanStabilitySuite" -Pspark-3.5 -nsu test SPARK_GENERATE_GOLDEN_FILES=1 ./mvnw -Dsuites="org.apache.spark.sql.comet.CometTPCDSV2_7_PlanStabilitySuite" -Pspark-4.0 -nsu test ``` diff --git a/docs/source/contributor-guide/iceberg-spark-tests.md b/docs/source/contributor-guide/iceberg-spark-tests.md index 9e35a82881c..5a7508e2665 100644 --- a/docs/source/contributor-guide/iceberg-spark-tests.md +++ b/docs/source/contributor-guide/iceberg-spark-tests.md @@ -79,8 +79,8 @@ Clone Apache Iceberg locally and apply the diff file from Comet against the matc ```shell git clone git@github.com:apache/iceberg.git apache-iceberg cd apache-iceberg -git checkout apache-iceberg-1.8.1 -git apply ../datafusion-comet/dev/diffs/iceberg/1.8.1.diff +git checkout apache-iceberg-1.9.1 +git apply ../datafusion-comet/dev/diffs/iceberg/1.9.1.diff ``` ## 3. Run Iceberg Spark Tests @@ -112,26 +112,26 @@ regenerate: ```shell cd apache-iceberg -git reset --hard apache-iceberg-1.8.1 && git clean -fd -git apply ../datafusion-comet/dev/diffs/iceberg/1.8.1.diff +git reset --hard apache-iceberg-1.9.1 && git clean -fd +git apply ../datafusion-comet/dev/diffs/iceberg/1.9.1.diff # Make changes, then run spotless to fix formatting ./gradlew spotlessApply # Stage any new or deleted files, then generate the diff git add -A -git diff apache-iceberg-1.8.1 > ../datafusion-comet/dev/diffs/iceberg/1.8.1.diff +git diff apache-iceberg-1.9.1 > ../datafusion-comet/dev/diffs/iceberg/1.9.1.diff ``` -Repeat for each Iceberg version (1.8.1, 1.9.1, 1.10.0, 1.11.0). The file contents differ between versions, so each +Repeat for each Iceberg version (1.9.1, 1.10.0, 1.11.0). The file contents differ between versions, so each diff must be generated against its own tag. ## Running Tests in CI -The `iceberg_1_8`, `iceberg_1_9`, `iceberg_1_10` and `iceberg_1_11` jobs in `ci.yml` apply these diffs and -run the three Gradle targets above against each Iceberg version, all with Java 17. Iceberg 1.8.1 runs against -Spark 3.4.3; Iceberg 1.9.1 and 1.10.0 run against Spark 3.5.9; Iceberg 1.11.0 runs against Spark 4.1.3. -Iceberg 1.11.0 runs in the merge queue; 1.8.1, 1.9.1 and 1.10.0 run once a night against `main`. All four +The `iceberg_1_9`, `iceberg_1_10` and `iceberg_1_11` jobs in `ci.yml` apply these diffs and +run the three Gradle targets above against each Iceberg version, all with Java 17. Iceberg 1.9.1 and 1.10.0 run +against Spark 3.5.9; Iceberg 1.11.0 runs against Spark 4.1.3. +Iceberg 1.11.0 runs in the merge queue; 1.9.1 and 1.10.0 run once a night against `main`. All three run earlier on a pull request labeled `run-iceberg-tests`; none runs on an unlabeled pull request. Each job calls `iceberg_spark_test_reusable.yml`, which holds the build and test job logic. See [.github/workflows/README.md](https://github.com/apache/datafusion-comet/blob/main/.github/workflows/README.md) diff --git a/docs/source/contributor-guide/release_process.md b/docs/source/contributor-guide/release_process.md index 0dac5641c8a..e0820d2fead 100644 --- a/docs/source/contributor-guide/release_process.md +++ b/docs/source/contributor-guide/release_process.md @@ -195,7 +195,7 @@ match a bare version too. Prefer patterns such as `comet-spark-spark3.5_2.12-*.j The release branch has the same CI workflows as `main`, so a `-SNAPSHOT`-only glob in a test harness or script fails only after the release branch is cut. The version bump pull request is where it shows up. It changes the `pom.xml` files, so it runs the suites that find the jar by file name or version, such as the PyArrow UDF tests and -the Spark SQL and Iceberg suites. The Spark SQL suite for Spark 3.4 runs only with the `run-spark-3.4-tests` label. +the Spark SQL and Iceberg suites. ### Update Version in main @@ -294,8 +294,7 @@ change log file into `main`. Once the generated docs, version bump, and change log have merged to the release branch, run every CI suite against it. Each pull request ran against the branch as it stood when its run started, so nothing has yet tested -the branch with all of them merged. Pull requests there also skip the Spark SQL suite for Spark 3.4 unless it is -labeled, and Miri, which runs only on a schedule on `main`. +the branch with all of them merged. Pull requests there also skip Miri, which runs only on a schedule on `main`. A dispatch runs every job in the release branch's own `ci.yml`, including `docs`, which publishes the website. So first check that the branch limits that job to `main`: the `if:` this prints must require @@ -314,7 +313,7 @@ gh workflow run miri.yml --repo apache/datafusion-comet --ref branch-0.13 ``` A dispatched `ci.yml` run ignores the tiers and the path filters. It runs every suite in the -[tier table](ci.md#three-tiers), including the Spark SQL suite for Spark 3.4, which sits outside every tier. +[tier table](ci.md#three-tiers). `miri.yml` runs the unsafe code checks, which are not part of `ci.yml`. Expect the runs to take a few hours. A failed dispatched run does not open a `ci-nightly-failure` issue, so check the result yourself. This prints the diff --git a/docs/source/contributor-guide/spark-sql-tests.md b/docs/source/contributor-guide/spark-sql-tests.md index 989cd85bf93..f2775a04c09 100644 --- a/docs/source/contributor-guide/spark-sql-tests.md +++ b/docs/source/contributor-guide/spark-sql-tests.md @@ -37,14 +37,14 @@ Here is an overview of the changes that we need to make to Spark: Comet's cache format although it is disabled by default. The serializer has to be a default for the whole JVM because Spark keeps the first cache serializer it loads for the rest of the JVM -Here are the steps involved in running the Spark SQL tests with Comet, using Spark 3.4.3 for this example. +Here are the steps involved in running the Spark SQL tests with Comet, using Spark 3.5.9 for this example. `dev/local-ci.sh` runs all of the steps below the way CI runs them: ```shell dev/local-ci.sh spark # every matrix row dev/local-ci.sh spark catalyst # one row -dev/local-ci.sh spark 3.4 catalyst # a non-default Spark version +dev/local-ci.sh spark 3.5 catalyst # a non-default Spark version ``` See [Continuous Integration](ci.md#reproducing-a-suite-failure-locally). The manual steps below @@ -55,7 +55,7 @@ are still the reference, and are what you want when creating or updating a diff Run `make release` in Comet to install the Comet JAR into the local Maven repository, specifying the Spark version. ```shell -PROFILES="-Pspark-3.4" make release +PROFILES="-Pspark-3.5" make release ``` ## 2. Clone Spark and Apply Diff @@ -65,9 +65,9 @@ Clone Apache Spark locally and apply the diff file from Comet. Note: this is a shallow clone of a tagged Spark commit and is not suitable for general Spark development. ```shell -git clone -b 'v3.4.3' --single-branch --depth 1 git@github.com:apache/spark.git apache-spark +git clone -b 'v3.5.9' --single-branch --depth 1 git@github.com:apache/spark.git apache-spark cd apache-spark -git apply ../datafusion-comet/dev/diffs/3.4.3.diff +git apply ../datafusion-comet/dev/diffs/3.5.9.diff ``` ## 3. Run Spark SQL Tests @@ -115,7 +115,7 @@ ENABLE_COMET=true ENABLE_COMET_ONHEAP=true sbt -J-Xmx4096m -Dspark.test.includeS 2. Set `ENABLE_COMET=true` and `ENABLE_COMET_ONHEAP=true` in environment variables. Without `ENABLE_COMET_ONHEAP`, Comet stays disabled, because the tests do not enable off-heap memory. ![img.png](img.png) -3. After the above tests are configured, spark tests can be run with debugging enabled on spark/comet code. Note that Comet is added as a dependency and the classes are readonly while debugging from Spark. Any new changes to Comet are to be built and deployed locally through the command (`PROFILES="-Pspark-3.4" make release`) +3. After the above tests are configured, spark tests can be run with debugging enabled on spark/comet code. Note that Comet is added as a dependency and the classes are readonly while debugging from Spark. Any new changes to Comet are to be built and deployed locally through the command (`PROFILES="-Pspark-3.5" make release`) ## Creating a diff file for a new Spark version @@ -194,21 +194,6 @@ earlier on a pull request carrying the `run-spark-3.5-tests` / `run-spark-4.0-te request. Apply the 3.5, 4.0 or 4.2 label when a change could behave differently on those versions, since without it the first verdict is the nightly run's, after the change has landed. -Spark 3.4 is deprecated and is not run by the merge queue. It runs only when a pull request carries -the `run-spark-3.4-tests` label, or when `ci.yml` is dispatched manually from the Actions page. -Apply the label if your change touches `spark/src/main/spark-3.4/`, `dev/diffs/3.4.3.diff`, or -shared code whose Spark 3.4 behavior you are unsure of: - -```shell -gh pr edit --add-label run-spark-3.4-tests -``` - -The run that command starts is advisory: it reports under `Required Checks (label run)`, not the -`Required Checks` status `main` requires, so a Spark 3.4 failure there does not block the merge on -its own. The label stays applied, so the next push you make runs Spark 3.4 as part of the required -verdict. Read the label run's result before merging — with Spark 3.4 out of the merge queue, -nothing else will. - See [Continuous Integration](ci.md) for how the tiers and labels work. When bringing up a new Spark version, the version needs its own job in `.github/workflows/ci.yml` plus entries in `FILTERS` and `POLICY` in `dev/ci/compute-changes.py`. diff --git a/docs/source/user-guide/latest/compatibility/expressions/index.md b/docs/source/user-guide/latest/compatibility/expressions/index.md index 2681261ce13..dbfe346b2f8 100644 --- a/docs/source/user-guide/latest/compatibility/expressions/index.md +++ b/docs/source/user-guide/latest/compatibility/expressions/index.md @@ -26,16 +26,12 @@ version you are running: ```{toctree} :maxdepth: 2 -Spark 3.4 Spark 3.5 Spark 4.0 Spark 4.1 Spark 4.2 ``` -Spark 3.4 support is deprecated and will be removed in a future release; see -[Spark Version Compatibility](../spark-versions.md#spark-34). - Expressions that are not 100% Spark-compatible fall back to Spark by default, except those with a JVM codegen-dispatch path, which stay in Comet's native pipeline and match Spark exactly. Set `spark.comet.expression.EXPRNAME.allowIncompatible=true`, where `EXPRNAME` is diff --git a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/index.md b/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/index.md deleted file mode 100644 index 0550dfb0c66..00000000000 --- a/docs/source/user-guide/latest/compatibility/expressions/spark-3.4/index.md +++ /dev/null @@ -1,46 +0,0 @@ - - -# Expression Compatibility (Spark 3.4) - -Compatibility notes for Comet running on Apache Spark 3.4. Expressions that are not 100% -Spark-compatible fall back to Spark by default, except those with a JVM codegen-dispatch -path, which stay in Comet's native pipeline and match Spark exactly. Set -`spark.comet.expression.EXPRNAME.allowIncompatible=true`, where `EXPRNAME` is the Spark -expression class name, to run Comet's native implementation despite its differences -from Spark. See the [Comet Supported Expressions Guide](../../../expressions.md) -for more information on this configuration setting. - -Spark 3.4 support is deprecated and will be removed in a future release; see -[Spark Version Compatibility](../../spark-versions.md#spark-34). - -```{toctree} -:maxdepth: 1 - -aggregate -array -datetime -map -math -misc -string -struct -url -cast -``` diff --git a/docs/source/user-guide/latest/compatibility/spark-versions.md b/docs/source/user-guide/latest/compatibility/spark-versions.md index 3ae6fb1b336..ffbbfc1059c 100644 --- a/docs/source/user-guide/latest/compatibility/spark-versions.md +++ b/docs/source/user-guide/latest/compatibility/spark-versions.md @@ -26,25 +26,6 @@ compatibility guide. For the rule that governs how long each Spark minor is supp [Apache Spark version support section](../../../about/versioning_policy.md#apache-spark-version-support) of the versioning policy. -## Spark 3.4 - -Spark 3.4.3 is supported with Java 17 and Scala 2.12/2.13. - -```{warning} -Spark 3.4 support is deprecated as of the 1.0.0 release and will be removed in a future release. -Comet continues to build and publish Spark 3.4 binaries in the meantime, but Apache Spark's own -SQL test suite no longer runs against Spark 3.4 automatically: it runs only when a contributor -opts a pull request into it. Regressions specific to Spark 3.4 are therefore more likely to reach a -release than on the other supported versions. We recommend moving to Spark 3.5 or later. -``` - -### Known Limitations - -- **Extra `SparkException` layer in Parquet schema mismatch errors**: when a Parquet read is - rejected because a file's type cannot be converted to the requested type, the error's cause chain - has one more `SparkException` layer than Spark's own reader produces. - See [Parquet](../parquet.md#schema-mismatch-handling) for details. - ## Spark 3.5 Spark 3.5.9 is supported with Java 17 and Scala 2.12/2.13. diff --git a/docs/source/user-guide/latest/installation.md b/docs/source/user-guide/latest/installation.md index 3ee1562a215..434b9b0d07b 100644 --- a/docs/source/user-guide/latest/installation.md +++ b/docs/source/user-guide/latest/installation.md @@ -47,15 +47,8 @@ Other versions may work well enough for development and evaluation purposes. Comet requires JDK 17 or later. JDK 11 is no longer supported as of the 1.1.0 release. -```{warning} -Spark 3.4 support is deprecated as of the 1.0.0 release and will be removed in a future release. -Apache Spark's own SQL test suite no longer runs against Spark 3.4 automatically; it runs only on -demand. We recommend moving to Spark 3.5 or later. -``` - | Spark Version | Java Version | Scala Version | Comet Tests in CI | Spark SQL Tests in CI | | ------------- | ------------ | ------------- | ----------------- | --------------------- | -| 3.4.3 | 17 | 2.12/2.13 | Nightly | On demand | | 3.5.9 | 17 | 2.12/2.13 | Nightly | Nightly | | 4.0.4 | 17/21 | 2.13 | Nightly | Nightly | | 4.1.3 | 17/21 | 2.13 | Before merge | Before merge | @@ -65,9 +58,7 @@ Note that we do not test the full matrix of supported Java and Scala versions in "Before merge" in the table above means the suite must pass before a change is merged. "Nightly" means the suite runs once a day against the `main` branch, so a regression it finds is caught after the change -has been merged rather than before. "On demand" means the suite does not run automatically at all. A -contributor can still run it against an individual pull request, but Spark 3.4 is no longer covered by -default. +has been merged rather than before. Note that Comet may not fully work with proprietary forks of Apache Spark such as the Spark versions offered by Cloud Service Providers. @@ -93,7 +84,6 @@ records the commit each snapshot was built from. The following artifacts are published: -- `comet-spark-spark3.4_2.12` - `comet-spark-spark3.5_2.12` - `comet-spark-spark4.0_2.13` - `comet-spark-spark4.1_2.13` @@ -133,8 +123,6 @@ repository describing your environment, and [build from source] for your target Here are the direct links for downloading the Comet $COMET_VERSION jar file. -- [Comet plugin for Spark 3.4 / Scala 2.12](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark3.4_2.12/$COMET_VERSION/comet-spark-spark3.4_2.12-$COMET_VERSION.jar) -- [Comet plugin for Spark 3.4 / Scala 2.13](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark3.4_2.13/$COMET_VERSION/comet-spark-spark3.4_2.13-$COMET_VERSION.jar) - [Comet plugin for Spark 3.5 / Scala 2.12](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark3.5_2.12/$COMET_VERSION/comet-spark-spark3.5_2.12-$COMET_VERSION.jar) - [Comet plugin for Spark 3.5 / Scala 2.13](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark3.5_2.13/$COMET_VERSION/comet-spark-spark3.5_2.13-$COMET_VERSION.jar) - [Comet plugin for Spark 4.0 / Scala 2.13](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark4.0_2.13/$COMET_VERSION/comet-spark-spark4.0_2.13-$COMET_VERSION.jar) diff --git a/pom.xml b/pom.xml index a215e2a818f..49efb28cf0a 100644 --- a/pom.xml +++ b/pom.xml @@ -105,17 +105,16 @@ under the License. 1.12.262 3.4.1 + in this pom sets delta.spark.version to its matching Delta release. The delta.artifact + property names the Delta artifact (delta-spark for the 3.x/4.x lines). The defaults + match the default spark-4.1 profile and only serve invocations that activate no + Spark profile (e.g. `mvn -Pdelta spotless:apply`). This is a separate property from + `delta.version` (the in-tree `-Pcontrib-delta` contrib) on purpose, so bumping one + pairing cannot silently change the other. --> delta-spark 4.3.1 - delta-core - 2.4.0 - 17 - ${java.version} - ${java.version} - - - spark-3.5 diff --git a/spark/pom.xml b/spark/pom.xml index 16a43db3200..071140fd5ef 100644 --- a/spark/pom.xml +++ b/spark/pom.xml @@ -344,63 +344,6 @@ under the License. - - spark-3.4 - - - - com.amazonaws - aws-java-sdk-s3 - ${aws-java-sdk-v1.version} - provided - - - commons-logging - commons-logging - - - - - com.amazonaws - aws-java-sdk-dynamodb - ${aws-java-sdk-v1.version} - provided - - - commons-logging - commons-logging - - - - - org.apache.iceberg - iceberg-spark-runtime-${spark.version.short}_${scala.binary.version} - 1.5.2 - test - - - - org.eclipse.jetty - jetty-server - 9.4.53.v20231009 - test - - - org.eclipse.jetty - jetty-servlet - 9.4.53.v20231009 - test - - - - spark-3.5 - Spark 3.4, 3.5, and 4.0 all have slightly different QueryExecutionErrors APIs. - Comet ships a separate ShimSparkErrorConverter implementation for each version. + Spark 3.5 and 4.x have slightly different QueryExecutionErrors APIs. + Comet ships a separate ShimSparkErrorConverter implementation for each. @@ -16,53 +16,40 @@ Parses JSON → calls convertErrorType() - - - + + - spark-3.4 - spark-3.5 - spark-4.0 + spark-3.5 + spark-4.x - - ShimSparkErrorConverter - spark-3.4 edition - Uses Spark 3.4 API signatures - (no SQLQueryContext in some calls) - - - ShimSparkErrorConverter - spark-3.5 edition - Uses Spark 3.5 API signatures - (takes SQLQueryContext directly) - - - ShimSparkErrorConverter - spark-4.0 edition - Uses Spark 4.0 API signatures - (new functionName param in some) + + ShimSparkErrorConverter + spark-3.5 edition + Uses Spark 3.5 API signatures + (takes SQLQueryContext directly) + + + ShimSparkErrorConverter + spark-4.x edition + Uses Spark 4.x API signatures (4.0, 4.1, 4.2) + (new functionName param in some) - - - + + - - QueryExecutionErrors - Spark 3.4 internal API - - - QueryExecutionErrors - Spark 3.5 internal API + + QueryExecutionErrors + Spark 3.5 internal API - - QueryExecutionErrors - Spark 4.0 internal API + + QueryExecutionErrors + Spark 4.x internal API - Example difference: "BinaryArithmeticOverflow" — Spark 3.x does NOT take functionName param, Spark 4.0 does. + Example difference: "BinaryArithmeticOverflow" — Spark 3.5 does NOT take functionName param, Spark 4.x does. diff --git a/docs/source/contributor-guide/sql_error_propagation.md b/docs/source/contributor-guide/sql_error_propagation.md index 59e139f0c90..eb253466b64 100644 --- a/docs/source/contributor-guide/sql_error_propagation.md +++ b/docs/source/contributor-guide/sql_error_propagation.md @@ -405,8 +405,8 @@ def convertToSparkException(e: CometQueryExecutionException): Throwable = { ### `ShimSparkErrorConverter` calls the real Spark API -Because Spark's `QueryExecutionErrors` API changes between Spark versions (3.4, 3.5, and the 4.x line), -there is a separate implementation per branch (in `spark-3.4/`, `spark-3.5/`, and `spark-4.x/`, which is +Because Spark's `QueryExecutionErrors` API changes between Spark versions (3.5 and the 4.x line), +there is a separate implementation per branch (in `spark-3.5/` and `spark-4.x/`, which is shared by Spark 4.0, 4.1, and 4.2). ![Shim pattern for per-version Spark API bridging](./shim_pattern.svg) diff --git a/docs/source/user-guide/latest/compatibility/floating-point.md b/docs/source/user-guide/latest/compatibility/floating-point.md index d185e254b74..e43f602ea9a 100644 --- a/docs/source/user-guide/latest/compatibility/floating-point.md +++ b/docs/source/user-guide/latest/compatibility/floating-point.md @@ -118,7 +118,7 @@ natively on Spark 4.0.5+, 4.1.4+, and 4.2+. These releases treat `-0.0` and `0.0 representation, as one value in these functions, including inside nested arrays and structs, and return the normalized value. Spark 4.2.0 normalizes the arguments in the plan (SPARK-54918), and 4.0.5, 4.1.4, and 4.2.1 normalize while these functions evaluate (SPARK-59602). Comet normalizes -the elements the same way. Spark 3.4, 3.5, 4.0.0 to 4.0.4, and 4.1.0 to 4.1.3 keep positive and +the elements the same way. Spark 3.5, 4.0.0 to 4.0.4, and 4.1.0 to 4.1.3 keep positive and negative zero distinct in flat arrays, so on those versions these functions fall back to Spark. Other element types remain native on every version. diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index a24cc12a8ce..7c98b3fb93a 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -136,7 +136,7 @@ Comet puts the files of each store into separate partitions, so such a scan can number of partitions than Spark would use. A native Parquet scan packs each store's files on their own, so `spark.sql.files.maxPartitionNum` applies to each store's files separately. A native CSV scan that would split a partition falls back to Spark when Spark planned the operators above it for -the scan's own partitioning, as Spark 3.4 does for a scan of one partition. +the scan's own partitioning. Such a scan forwards the object store settings of every scheme it reads, so its files can mix schemes such as `s3a://`, `gs://` and `hdfs://`. It falls back to Spark when it reads one bucket @@ -458,7 +458,7 @@ Azure scans discover Root CA Certificates the same way S3 scans do. See [Root CA ### Authentication is resolved by Hadoop -For each `abfs` or `abfss` path, the Spark driver asks the hadoop-azure library on its classpath which authentication mechanism applies to that container and account, and which values it needs. The question goes to the same `AbfsConfiguration` the ABFS `FileSystem` uses, so every Hadoop rule holds: account-scoped keys (`.`), container-scoped keys (`..`, hadoop-azure 3.4.2 and later), credential providers behind `hadoop.security.credential.provider.path` (a JCEKS keystore, for example) and `${...}` substitution behave exactly as they do for Spark's own reads. This holds for the hadoop-azure matching each Spark line's Hadoop: 3.3.4 with Spark 3.4 and 3.5, 3.4.1 with 4.0, 3.4.2 with 4.1, 3.5.0 with 4.2. +For each `abfs` or `abfss` path, the Spark driver asks the hadoop-azure library on its classpath which authentication mechanism applies to that container and account, and which values it needs. The question goes to the same `AbfsConfiguration` the ABFS `FileSystem` uses, so every Hadoop rule holds: account-scoped keys (`.`), container-scoped keys (`..`, hadoop-azure 3.4.2 and later), credential providers behind `hadoop.security.credential.provider.path` (a JCEKS keystore, for example) and `${...}` substitution behave exactly as they do for Spark's own reads. This holds for the hadoop-azure matching each Spark line's Hadoop: 3.3.4 with Spark 3.5, 3.4.1 with 4.0, 3.4.2 with 4.1, 3.5.0 with 4.2. The native scan receives the resolved values and builds the `object_store` client from them. It reads no `fs.azure.*` key of its own. A scan whose files span more than one ABFS container or account, or that mixes `abfs`/`abfss` paths with other schemes, falls back to Spark's own scan. diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index cf8baac011d..58ba641a2f8 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -769,4 +769,4 @@ still falls back when needed. ## See also - [Comet Compatibility Guide](compatibility/index.md) - known incompatibilities and edge cases for supported expressions. -- [Expression Audits (contributor guide)](../../contributor-guide/expression-audits/index.md) - per-version (Spark 3.4 / 3.5 / 4.0 / 4.1) audit notes for audited expressions. +- [Expression Audits (contributor guide)](../../contributor-guide/expression-audits/index.md) - per-version (Spark 3.5 / 4.0 / 4.1) audit notes for audited expressions. diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index f788ab1da9c..8618ac105c2 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -411,14 +411,13 @@ summary-aware writer commit contract. For copy-on-write row-level DML, the mechanism differs by Spark version: on Spark 4.0+ the analyzer emits operation-coded rows that Comet's writer dispatches through `ReplaceData`'s -projections, while on Spark 3.4/3.5 the rewritten rows are written as a plain row stream. The +projections, while on Spark 3.5 the rewritten rows are written as a plain row stream. The supported set of operations is the same either way. Merge-on-read uses Spark's `WriteDelta`, and Iceberg's JVM `DeltaWriter` writes its rows under either plan, so it keeps Spark's stock write plan. On Spark 3.5+ the testing setting `spark.comet.write.iceberg.splitOperator.enabled` plans it as the split plan, with the same driver commit and reporting path, but `CometIcebergWriteExec` is never used for position-delta rows. -Spark 3.4 always leaves `WriteDelta` on Spark's stock write plan. On Spark 4.1+ the split plan matches two further stock-Spark behaviours: MERGE metrics are forwarded to the writer's commit (Iceberg 1.11+ records them in the snapshot summary), and @@ -438,8 +437,6 @@ The rewrite is skipped — and the write runs through Spark's stock combined ope - the write is neither an Iceberg `SparkWrite` nor a supported Iceberg position-delta write; - the table uses merge-on-read (`WriteDelta`), unless the testing setting `spark.comet.write.iceberg.splitOperator.enabled` is on with Spark 3.5+; -- the statement is CTAS / RTAS on Spark 3.4, where the staged exec writes inline; on Spark - 3.5+ those statements re-plan their inner append, which is intercepted normally; - the write is a streaming write (`writeStream` into an Iceberg table); - the write requires Spark's commit coordinator, which Comet's per-task commit protocol does not use; diff --git a/docs/source/user-guide/latest/in-memory-cache.md b/docs/source/user-guide/latest/in-memory-cache.md index c1bbb479c42..ef73632b925 100644 --- a/docs/source/user-guide/latest/in-memory-cache.md +++ b/docs/source/user-guide/latest/in-memory-cache.md @@ -34,10 +34,6 @@ $SPARK_HOME/bin/spark-shell \ --conf spark.comet.exec.inMemoryCache.enabled=true ``` -On Spark 3.4, a union of a shuffle and a relation cached in Comet's format keeps every shuffle -partition. Spark 3.4 has no hook for the rule through which Comet lets AQE coalesce the shuffle -partitions of a union that Comet runs. - It has to be set before the `SparkContext` starts. Comet's driver plugin chooses `spark.sql.cache.serializer` once, while the context is initializing, so an application keeps the cache format it started with however the config is set afterwards. The plugin installs Comet's diff --git a/docs/source/user-guide/latest/installation.md b/docs/source/user-guide/latest/installation.md index 434b9b0d07b..05e17ef325b 100644 --- a/docs/source/user-guide/latest/installation.md +++ b/docs/source/user-guide/latest/installation.md @@ -205,7 +205,7 @@ WARN CometExecRule: Comet cannot execute some parts of this plan natively (set s +- LocalTableScan [COMET: Native support for operator LocalTableScanExec is disabled. Set spark.comet.exec.localTableScan.enabled=true to enable it.] ``` -On Spark 3.4 and 3.5 the native writer replaces the whole write command rather than just the +On Spark 3.5 the native writer replaces the whole write command rather than just the per-task write, so the same message appears on `Execute InsertIntoHadoopFsRelationCommand` and names `DataWritingCommandExec`. diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md index 2e1a49d23a1..969319e0c11 100644 --- a/docs/source/user-guide/latest/operators.md +++ b/docs/source/user-guide/latest/operators.md @@ -137,7 +137,7 @@ combined with `OR`, which Spark plans as an existence `BroadcastNestedLoopJoinEx | Operator | Status | Notes | | ------------------------------------------------------------------------------------------------------------------ | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `WriteFilesExec` | ⚠️ | Spark 4.0+. Experimental native Parquet writes, disabled by default (opt-in). Non-partitioned, non-bucketed writes only, and not when `spark.sql.files.maxRecordsPerFile` is set. See [Parquet](parquet.md#writing). | -| `DataWritingCommandExec` | ⚠️ | Spark 3.4/3.5 only. Experimental native Parquet writes, disabled by default (opt-in). Replaced by `WriteFilesExec` on Spark 4.0+ and removed with Spark 3.x support. See [Parquet](parquet.md#writing). | +| `DataWritingCommandExec` | ⚠️ | Spark 3.5 only. Experimental native Parquet writes, disabled by default (opt-in). Replaced by `WriteFilesExec` on Spark 4.0+ and removed with Spark 3.x support. See [Parquet](parquet.md#writing). | | Iceberg writes: `AppendDataExec`, `OverwriteByExpressionExec`, `OverwritePartitionsDynamicExec`, `ReplaceDataExec` | ⚠️ | Disabled by default (opt-in). `spark.comet.write.iceberg.enabled=true` plans the write as `IcebergWrite` and `IcebergCommit` and writes eligible data files natively (`CometIcebergWrite`). Other writes use Iceberg's Java writer. Covers `INSERT INTO`, `INSERT OVERWRITE`, and copy-on-write `DELETE` / `UPDATE` / `MERGE`. Merge-on-read writes (`WriteDeltaExec`) fall back. See [Iceberg Writes](iceberg.md#writes). | | `MergeRowsExec` | ⚠️ | Spark 3.5+. Experimental, disabled by default (`spark.comet.exec.mergeRows.enabled`). On Spark 4.1+, stock V2 writers retain Spark MergeRowsExec; Comet's split Iceberg path can run it natively while preserving MergeSummary. See [MERGE INTO](compatibility/operators.md#merge-into-mergerowsexec). | diff --git a/docs/source/user-guide/latest/parquet.md b/docs/source/user-guide/latest/parquet.md index 04132cf8e47..22b8fe71a4d 100644 --- a/docs/source/user-guide/latest/parquet.md +++ b/docs/source/user-guide/latest/parquet.md @@ -113,8 +113,8 @@ reads that fail where Spark succeeds are listed in ### Missing Nested Fields -When all requested leaf fields under a nested column are missing from a Parquet file, Spark 3.4 -through 4.0 return NULL for the entire missing subtree, including nested arrays and maps. Comet +When all requested leaf fields under a nested column are missing from a Parquet file, Spark 3.5 +and 4.0 return NULL for the entire missing subtree, including nested arrays and maps. Comet follows this behavior. On Spark 4.1+, the default (`false`) preserves the file's parent nullness. Set `spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing=true` for legacy behavior. A field that exists in the file but contains only NULL values is not a missing field. @@ -172,7 +172,7 @@ SchemaColumnConvertNotSupportedException`) instead of the one-level chain Spark' - **Spark 3.x: `TIMESTAMP_MILLIS` overflow past the first batch is wrapped in a file-read error**. When a Parquet `TIMESTAMP_MILLIS` value overflows on conversion to microseconds, Spark's reader - throws `ArithmeticException: long overflow`. Spark 3.4 and 3.5 wrap a reader error in a + throws `ArithmeticException: long overflow`. Spark 3.5 wraps a reader error in a `SparkException` that names the file only while reading the first batch of each file split. For the vectorized reader, that batch is the first `spark.sql.parquet.columnarReaderBatchSize` rows (4096 by default), or the first row group if that is smaller. An overflow in a later batch fails @@ -315,7 +315,7 @@ for production, so do not turn it off to avoid an entry on this page. With `spark.comet.write.parquet.enabled=true`: -- **The `EXCEPTION` rebase mode is ignored.** On Spark 3.4 and 3.5, whose default rebase mode is +- **The `EXCEPTION` rebase mode is ignored.** On Spark 3.5, whose default rebase mode is `EXCEPTION`, the native writer writes dates before October 15, 1582 and timestamps before January 1, 1900 that Spark refuses to write ([#7047](https://github.com/apache/datafusion-comet/issues/7047)). With @@ -326,7 +326,7 @@ With `spark.comet.write.parquet.enabled=true`: as written by Spark 3.0 or later and reads the values back unchanged under every `spark.sql.parquet.datetimeRebaseModeInRead` setting. Files written by a Comet build that predates that key have no version, and Spark then fails to read their pre-1582 values under its - default read mode on 3.4 and 3.5, or reads them shifted by several days under `LEGACY`. + default read mode on 3.5, or reads them shifted by several days under `LEGACY`. - **`spark.sql.parquet.outputTimestampType` and `spark.sql.parquet.writeLegacyFormat` are ignored.** Timestamps are always written as `INT64` microseconds, never as Spark's default `INT96`, so with `TIMESTAMP_MILLIS` they keep the sub-millisecond digits that Spark drops. Spark's @@ -335,7 +335,7 @@ With `spark.comet.write.parquet.enabled=true`: `spark.sql.parquet.inferTimestampNTZ.enabled=false` it reads `TIMESTAMP_NTZ` columns as `TIMESTAMP` ([#7048](https://github.com/apache/datafusion-comet/issues/7048), [#3425](https://github.com/apache/datafusion-comet/issues/3425)). -- **Spark 3.4 and 3.5: `INSERT INTO` leaves the target stale.** Comet skips the steps Spark runs +- **Spark 3.5: `INSERT INTO` leaves the target stale.** Comet skips the steps Spark runs after a write: file index refresh, cache refresh, partition refresh and statistics update. Later queries can return the table's previous contents until `REFRESH TABLE` ([#6315](https://github.com/apache/datafusion-comet/issues/6315)). diff --git a/docs/source/user-guide/latest/pyarrow-udfs.md b/docs/source/user-guide/latest/pyarrow-udfs.md index 48848ecd6e3..39846b2b6bb 100644 --- a/docs/source/user-guide/latest/pyarrow-udfs.md +++ b/docs/source/user-guide/latest/pyarrow-udfs.md @@ -287,7 +287,7 @@ on the unoptimized path. `org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager` and enable `spark.comet.shuffle.enabled=true` at session startup. With a vanilla Spark `Exchange` in the plan the data leaves the shuffle as rows and the optimization cannot fire. -- Spark 4.0 or newer is required. On Spark 3.4 and 3.5 the optimization is a no-op even when +- Spark 4.0 or newer is required. On Spark 3.5 the optimization is a no-op even when enabled; vanilla `PythonMapInArrowExec` / `MapInPandasExec` handle the operation. The Spark 3.5 `PythonArrowInput` trait has a different contract than 4.x and a separate implementation has not been written. Track 3.5 support as a future follow-on if there is user demand. diff --git a/docs/source/user-guide/latest/s3-credential-providers.md b/docs/source/user-guide/latest/s3-credential-providers.md index c32dcd75d04..d33ce766929 100644 --- a/docs/source/user-guide/latest/s3-credential-providers.md +++ b/docs/source/user-guide/latest/s3-credential-providers.md @@ -245,7 +245,7 @@ Publish a real `expirationEpochMillis` when you have one. On both paths Comet re On the Parquet/`object_store` path, `object_store` signs a request once and sends the same signature on every retry, for up to 3 minutes by default, and it does not retry a 403. The five minutes Comet leaves before an expiry cover those retries. A credential you return with less time left than that is used for the request that asked for it and not kept. -The built-in adapters report an expiry when the AWS SDK exposes one, which it does on the Spark 4.x builds (SDK v2). On the Spark 3.4 and 3.5 builds (SDK v1) they cannot, and report `0`. +The built-in adapters report an expiry when the AWS SDK exposes one, which it does on the Spark 4.x builds (SDK v2). On the Spark 3.5 build (SDK v1) they cannot, and report `0`. ### Returned fields diff --git a/spark/src/main/scala/org/apache/comet/GenerateDocs.scala b/spark/src/main/scala/org/apache/comet/GenerateDocs.scala index f36487080d3..06a34c907f8 100644 --- a/spark/src/main/scala/org/apache/comet/GenerateDocs.scala +++ b/spark/src/main/scala/org/apache/comet/GenerateDocs.scala @@ -144,7 +144,7 @@ object GenerateDocs { * Args: * - args(0): user guide root directory (e.g. `docs/source/user-guide/latest/`). * - args(1) (optional): per-Spark-version subdirectory for compatibility pages (e.g. - * `spark-3.4`). When omitted, compat pages are written to `compatibility/expressions/` + * `spark-3.5`). When omitted, compat pages are written to `compatibility/expressions/` * directly, preserving the legacy flat layout used by released-version doc trees. */ def main(args: Array[String]): Unit = { diff --git a/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergDeltaLogicalShim.scala b/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergDeltaLogicalShim.scala deleted file mode 100644 index afbf7d60b70..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergDeltaLogicalShim.scala +++ /dev/null @@ -1,27 +0,0 @@ -/* - * 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 org.apache.spark.sql.catalyst.plans.logical.LogicalPlan - -/** WriteDelta interception starts with Spark 3.5. */ -private[iceberg] object IcebergDeltaLogicalShim extends IcebergDeltaLogicalShimApi { - override def extract(plan: LogicalPlan): Option[DeltaLogicalFields] = None -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala b/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala deleted file mode 100644 index 4d66a374663..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala +++ /dev/null @@ -1,40 +0,0 @@ -/* - * 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 org.apache.spark.sql.connector.catalog.TableCatalog -import org.apache.spark.sql.execution.SparkPlan -import org.apache.spark.sql.execution.datasources.v2.{AtomicCreateTableAsSelectExec, AtomicReplaceTableAsSelectExec, CreateTableAsSelectExec, ReplaceTableAsSelectExec} - -/** - * Spark 3.4: CTAS and RTAS write the table from the create or replace exec itself, through - * `TableWriteExecHelper.writeWithV2`, so the write never appears as a write node of its own. - */ -private[iceberg] object IcebergTableAsSelectShim { - - /** The catalog `plan` writes through, when `plan` is a CTAS or RTAS exec. */ - def writeCatalog(plan: SparkPlan): Option[TableCatalog] = plan match { - case p: CreateTableAsSelectExec => Some(p.catalog) - case p: AtomicCreateTableAsSelectExec => Some(p.catalog) - case p: ReplaceTableAsSelectExec => Some(p.catalog) - case p: AtomicReplaceTableAsSelectExec => Some(p.catalog) - case _ => None - } -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala deleted file mode 100644 index f983a6cb39e..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/CometExprShim.scala +++ /dev/null @@ -1,78 +0,0 @@ -/* - * 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.shims - -import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.catalyst.expressions.aggregate.Sum - -import org.apache.comet.expressions.CometEvalMode -import org.apache.comet.serde.{CometAggregateExpressionSerde, CometEncode, CometExpressionSerde, CometStringDecode} -import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr} - -/** - * `CometExprShim` acts as a shim for parsing expressions from different Spark versions. - */ -trait CometExprShim { - protected def evalMode(c: Cast): CometEvalMode.Value = - CometEvalModeUtil.fromSparkEvalMode(c.evalMode) - - def binaryOutputStyle: BinaryOutputStyle = BinaryOutputStyle.HEX_DISCRETE - - // Spark 3.4 ships Jackson 2.14, which does not impose the default numeric-token length limit, - // but it can run with a newer Jackson, so choose the native variant from the runtime version. - def getJsonObjectNativeFunctionName: String = - if (CometExprShim.jacksonLimitsNumberLength) "get_json_object" - else "get_json_object_spark34" - - def sparkVersionSpecificStringExpressions - : Map[Class[_ <: Expression], CometExpressionSerde[_]] = - Map(classOf[StringDecode] -> CometStringDecode, classOf[Encode] -> CometEncode) - def sparkVersionSpecificMathExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = - Map.empty - def sparkVersionSpecificMiscExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = - Map.empty - def sparkVersionSpecificMapExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = - Map.empty - def sparkVersionSpecificAggregates: Map[Class[_], CometAggregateExpressionSerde[_]] = - Map.empty - - def sparkVersionSpecificExprToProtoInternal( - expr: Expression, - inputs: Seq[Attribute], - binding: Boolean): Option[Expr] = None -} - -object CometExprShim { - private lazy val jacksonLimitsNumberLength: Boolean = { - // Jackson 2.15 added StreamReadConstraints' default 1000-digit number limit - val v = com.fasterxml.jackson.core.json.PackageVersion.VERSION - v.getMajorVersion > 2 || (v.getMajorVersion == 2 && v.getMinorVersion >= 15) - } -} - -object CometEvalModeUtil { - def fromSparkEvalMode(evalMode: EvalMode.Value): CometEvalMode.Value = evalMode match { - case EvalMode.LEGACY => CometEvalMode.LEGACY - case EvalMode.TRY => CometEvalMode.TRY - case EvalMode.ANSI => CometEvalMode.ANSI - } - - def sumEvalMode(s: Sum): EvalMode.Value = s.evalMode -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimBatchReader.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimBatchReader.scala deleted file mode 100644 index 17b60e0e5a1..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimBatchReader.scala +++ /dev/null @@ -1,37 +0,0 @@ -/* - * 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.shims - -import org.apache.spark.paths.SparkPath -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.execution.datasources.PartitionedFile - -object ShimBatchReader { - - def newPartitionedFile(partitionValues: InternalRow, file: String): PartitionedFile = - PartitionedFile( - partitionValues, - SparkPath.fromPathString(file), - -1, // -1 means we read the entire file - -1, - Array.empty[String], - 0, - 0) -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometBroadcastExchangeExec.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometBroadcastExchangeExec.scala deleted file mode 100644 index 93eee60923d..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometBroadcastExchangeExec.scala +++ /dev/null @@ -1,51 +0,0 @@ -/* - * 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.shims - -import org.apache.spark.SparkContext -import org.apache.spark.network.util.JavaUtils -import org.apache.spark.sql.execution.exchange.BroadcastExchangeLike -import org.apache.spark.sql.internal.SQLConf - -import org.apache.comet.shims.ShimCometBroadcastExchangeExec.SPARK_MAX_BROADCAST_TABLE_SIZE - -trait ShimCometBroadcastExchangeExec { - - def setJobGroupOrTag(sc: SparkContext, broadcastExchange: BroadcastExchangeLike): Unit = { - // Setup a job group here so later it may get cancelled by groupId if necessary. - sc.setJobGroup( - broadcastExchange.runId.toString, - s"broadcast exchange (runId ${broadcastExchange.runId})", - interruptOnCancel = true) - } - - def cancelJobGroup(sc: SparkContext, broadcastExchange: BroadcastExchangeLike): Unit = { - sc.cancelJobGroup(broadcastExchange.runId.toString) - } - - def maxBroadcastTableBytes(conf: SQLConf): Long = { - JavaUtils.byteStringAsBytes(conf.getConfString(SPARK_MAX_BROADCAST_TABLE_SIZE, "8GB")) - } - -} - -object ShimCometBroadcastExchangeExec { - val SPARK_MAX_BROADCAST_TABLE_SIZE = "spark.sql.maxBroadcastTableSize" -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometMergeRows.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometMergeRows.scala deleted file mode 100644 index 488e9734347..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometMergeRows.scala +++ /dev/null @@ -1,35 +0,0 @@ -/* - * 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.shims - -import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan -import org.apache.spark.sql.execution.SparkPlan - -import org.apache.comet.serde.CometOperatorSerde - -/** - * Spark 3.4 predates the core `MergeRowsExec` implementation used by newer Spark versions. - * Nothing is registered here, so MERGE continues through the version-specific JVM path. - */ -object ShimCometMergeRows { - val nativeExecs: Map[Class[_ <: SparkPlan], CometOperatorSerde[_]] = Map.empty - - def withNativeMergeSummary(query: LogicalPlan): LogicalPlan = query -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometSparkSessionExtensions.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometSparkSessionExtensions.scala deleted file mode 100644 index 55e3e1cf8a3..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometSparkSessionExtensions.scala +++ /dev/null @@ -1,60 +0,0 @@ -/* - * 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.shims - -import org.apache.spark.sql.SparkSessionExtensions -import org.apache.spark.sql.catalyst.rules.Rule -import org.apache.spark.sql.execution.{QueryExecution, SparkPlan} - -trait ShimCometSparkSessionExtensions { - - protected val EXTENDED_EXPLAIN_PROVIDERS_KEY = "spark.sql.extendedExplainProviders" - - def supportsExtendedExplainInfo(qe: QueryExecution): Boolean = { - try { - qe.getClass.getDeclaredMethod( - "extendedExplainInfo", - classOf[String => Unit], - classOf[SparkPlan]) - } catch { - case _: NoSuchMethodException | _: SecurityException => return false - } - true - } - - // injectQueryStageOptimizerRule not available on Spark 3.4. - // CometPlanAdaptiveDynamicPruningFilters and CometReuseSubquery are not registered. - // On 3.4, Spark's PlanAdaptiveDynamicPruningFilters handles SABs directly (converting to - // TrueLiteral when it can't find BroadcastHashJoinExec, disabling DPP). - def injectQueryStageOptimizerRuleShim( - extensions: SparkSessionExtensions, - rule: Rule[SparkPlan]): Unit = {} - - // Registers a queryStagePrepRule only on Spark < 3.5. The 3.5+ variants no-op this shim. - // Used by CometSpark34AqeDppFallbackRule, which is a correctness workaround for AQE DPP on - // Spark 3.4 where injectQueryStageOptimizerRule (and therefore - // CometPlanAdaptiveDynamicPruningFilters) is unavailable. See - // CometSpark34AqeDppFallbackRule's class docstring for details. - def injectPreSpark35QueryStagePrepRuleShim( - extensions: SparkSessionExtensions, - rule: Rule[SparkPlan]): Unit = { - extensions.injectQueryStagePrepRule(_ => rule) - } -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometWindowGroupLimit.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometWindowGroupLimit.scala deleted file mode 100644 index b338a32b05f..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimCometWindowGroupLimit.scala +++ /dev/null @@ -1,40 +0,0 @@ -/* - * 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.shims - -import org.apache.spark.sql.comet.CometWindowGroupLimitExec.Fields -import org.apache.spark.sql.execution.SparkPlan - -/** - * Spark 3.4 does not have `WindowGroupLimitExec` (introduced by SPARK-37099 in Spark 3.5). This - * no-op shim keeps the shared Comet code compiling against Spark 3.4 while ensuring the operator - * is never registered for conversion on this profile. - */ -object ShimCometWindowGroupLimit { - - /** - * The `WindowGroupLimitExec` class object on Spark 3.5+, or `None` on Spark 3.4. - * `CometExecRule` uses this to conditionally register the serde in `nativeExecs`. - */ - def windowGroupLimitClass: Option[Class[_ <: SparkPlan]] = None - - /** Extract WGL fields when `op` is a `WindowGroupLimitExec` (Spark 3.5+). */ - def extract(op: SparkPlan): Option[Fields] = None -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimFileFormat.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimFileFormat.scala deleted file mode 100644 index 6906a39fbfc..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimFileFormat.scala +++ /dev/null @@ -1,67 +0,0 @@ -/* - * 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.shims - -import org.apache.hadoop.fs.Path -import org.apache.spark.sql.catalyst.expressions.Literal -import org.apache.spark.sql.execution.datasources.{FileFormat, PartitionedFile, RowIndexUtil} -import org.apache.spark.sql.types.{DataType, StructType} - -object ShimFileFormat { - - // A name for a temporary column that holds row indexes computed by the file format reader - // until they can be placed in the _metadata struct. - val ROW_INDEX_TEMPORARY_COLUMN_NAME: String = FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME - - def findRowIndexColumnIndexInSchema(sparkSchema: StructType): Int = - RowIndexUtil.findRowIndexColumnIndexInSchema(sparkSchema) - - // Spark 3.4 has no per-format metadata extractor concept (added in Spark 3.5, SPARK-43868); - // it derives these values inline in FileFormat.updateMetadataInternalRow. Replicate that - // logic here so callers can use the same extractor-map shape across all Spark versions. - private val baseMetadataExtractors: Map[String, PartitionedFile => Any] = Map( - FileFormat.FILE_PATH -> { pf: PartitionedFile => - // Use `new Path(Path.toString)` as a form of canonicalization - new Path(pf.filePath.toPath.toString).toUri.toString - }, - FileFormat.FILE_NAME -> { pf: PartitionedFile => - pf.filePath.toUri.getRawPath.split("/").lastOption.getOrElse("") - }, - FileFormat.FILE_SIZE -> { pf: PartitionedFile => pf.fileSize }, - FileFormat.FILE_BLOCK_START -> { pf: PartitionedFile => pf.start }, - FileFormat.FILE_BLOCK_LENGTH -> { pf: PartitionedFile => pf.length }, - // The modificationTime from the file has millisecond granularity, but the TimestampType for - // `file_modification_time` has microsecond granularity. - FileFormat.FILE_MODIFICATION_TIME -> { pf: PartitionedFile => pf.modificationTime * 1000 }) - - def fileConstantMetadataExtractors( - fileFormat: FileFormat): Map[String, PartitionedFile => Any] = - baseMetadataExtractors - - // dataType is unused on this Spark version; getFileConstantMetadataColumnValue only gained - // a dataType parameter in Spark 4.2 (SPARK-56931). Accepted here so callers can pass it - // uniformly across Spark versions. - def getFileConstantMetadataColumnValue( - name: String, - file: PartitionedFile, - extractors: Map[String, PartitionedFile => Any], - dataType: DataType): Literal = - Literal(extractors(name)(file)) -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSQLConf.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSQLConf.scala deleted file mode 100644 index 3ab5e5cbaca..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSQLConf.scala +++ /dev/null @@ -1,36 +0,0 @@ -/* - * 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.shims - -import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy - -trait ShimSQLConf { - protected val LEGACY = LegacyBehaviorPolicy.LEGACY - protected val CORRECTED = LegacyBehaviorPolicy.CORRECTED - - /** - * Reads `spark.sql.execution.arrow.useLargeVarTypes`. Spark 3.4 has no typed accessor for this - * conf, so read by raw key. The conf only governs the destination Arrow IPC root width on Spark - * 4.x, so the value returned here matters only to callers that look it up explicitly. - */ - protected def arrowUseLargeVarTypes(conf: SQLConf): Boolean = - conf.getConfString("spark.sql.execution.arrow.useLargeVarTypes", "false").toBoolean -} diff --git a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSubqueryBroadcast.scala b/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSubqueryBroadcast.scala deleted file mode 100644 index 20663bb05ba..00000000000 --- a/spark/src/main/spark-3.4/org/apache/comet/shims/ShimSubqueryBroadcast.scala +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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.shims - -import org.apache.spark.sql.catalyst.expressions.Expression -import org.apache.spark.sql.execution.{SparkPlan, SubqueryAdaptiveBroadcastExec, SubqueryBroadcastExec} - -trait ShimSubqueryBroadcast { - - def getSubqueryBroadcastIndices(sab: SubqueryAdaptiveBroadcastExec): Seq[Int] = { - Seq(sab.index) - } - - def getSubqueryBroadcastExecIndices(sub: SubqueryBroadcastExec): Seq[Int] = { - Seq(sub.index) - } - - def createSubqueryBroadcastExec( - name: String, - indices: Seq[Int], - buildKeys: Seq[Expression], - child: SparkPlan): SubqueryBroadcastExec = { - assert(indices.length == 1, s"Multi-index DPP not supported on Spark 3.4: indices=$indices") - SubqueryBroadcastExec(name, indices.head, buildKeys, child) - } -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometInMemoryTableScanExec.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometInMemoryTableScanExec.scala deleted file mode 100644 index b5d7421b7ee..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometInMemoryTableScanExec.scala +++ /dev/null @@ -1,25 +0,0 @@ -/* - * 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.spark.sql.comet.shims - -import org.apache.spark.sql.execution.LeafExecNode - -// Spark 3.4 has no InMemoryTableScanLike or table-cache query stages. -trait ShimCometInMemoryTableScanExec extends LeafExecNode diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometScanExec.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometScanExec.scala deleted file mode 100644 index 7d8ba9f0f8b..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometScanExec.scala +++ /dev/null @@ -1,83 +0,0 @@ -/* - * 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.spark.sql.comet.shims - -import org.apache.hadoop.fs.{FileStatus, Path} -import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression} -import org.apache.spark.sql.execution.{FileSourceScanExec, PartitionedFileUtil} -import org.apache.spark.sql.execution.datasources._ -import org.apache.spark.sql.execution.datasources.parquet.ParquetOptions -import org.apache.spark.sql.sources.Filter -import org.apache.spark.sql.types.StructType - -import org.apache.comet.shims.ShimFileFormat - -trait ShimCometScanExec { - def wrapped: FileSourceScanExec - - lazy val fileConstantMetadataColumns: Seq[AttributeReference] = - wrapped.fileConstantMetadataColumns - - protected def newFileScanRDD( - fsRelation: HadoopFsRelation, - readFunction: PartitionedFile => Iterator[InternalRow], - filePartitions: Seq[FilePartition], - readSchema: StructType, - options: ParquetOptions): FileScanRDD = new FileScanRDD( - fsRelation.sparkSession, - readFunction, - filePartitions, - readSchema, - fileConstantMetadataColumns, - options) - - protected def isNeededForSchema(sparkSchema: StructType): Boolean = { - // TODO: remove after PARQUET-2161 becomes available in Parquet (tracked in SPARK-39634) - ShimFileFormat.findRowIndexColumnIndexInSchema(sparkSchema) >= 0 - } - - protected def getPartitionedFile(f: FileStatus, p: PartitionDirectory): PartitionedFile = - PartitionedFileUtil.getPartitionedFile(f, f.getPath, p.values) - - protected def splitFiles( - sparkSession: SparkSession, - file: FileStatus, - filePath: Path, - isSplitable: Boolean, - maxSplitBytes: Long, - partitionValues: InternalRow): Seq[PartitionedFile] = - PartitionedFileUtil.splitFiles( - sparkSession, - file, - filePath, - isSplitable, - maxSplitBytes, - partitionValues) - - protected def getPushedDownFilters( - relation: HadoopFsRelation, - dataFilters: Seq[Expression]): Seq[Filter] = { - val supportNestedPredicatePushdown = DataSourceUtils.supportNestedPredicatePushdown(relation) - dataFilters.flatMap(DataSourceStrategy.translateFilter(_, supportNestedPredicatePushdown)) - } - -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometShuffleMaterialization.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometShuffleMaterialization.scala deleted file mode 100644 index 26ba626a665..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometShuffleMaterialization.scala +++ /dev/null @@ -1,44 +0,0 @@ -/* - * 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.spark.sql.comet.shims - -import org.apache.spark.sql.SparkSession - -/** - * Restores the session context captured before shuffle materialization leaves the caller thread. - */ -trait ShimCometShuffleMaterialization { - private val activeSession = SparkSession.getActiveSession - - protected def withCapturedSession[T](body: => T): T = { - val previousSession = SparkSession.getActiveSession - activeSession match { - case Some(session) => SparkSession.setActiveSession(session) - case None => SparkSession.clearActiveSession() - } - try body - finally { - previousSession match { - case Some(session) => SparkSession.setActiveSession(session) - case None => SparkSession.clearActiveSession() - } - } - } -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometUnionExec.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometUnionExec.scala deleted file mode 100644 index 57db09d5f38..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometUnionExec.scala +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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.spark.sql.comet.shims - -import scala.reflect.ClassTag - -import org.apache.spark.SparkContext -import org.apache.spark.rdd.RDD -import org.apache.spark.sql.catalyst.plans.physical.Partitioning - -object ShimCometUnionExec { - - /** - * Unions a sequence of RDDs while preserving the declared output partitioning. Before Spark - * 4.1, [[org.apache.spark.sql.execution.UnionExec]] always reports - * [[org.apache.spark.sql.catalyst.plans.physical.UnknownPartitioning]], so this shim simply - * concatenates partitions via `SparkContext.union`. The partitioning-aware path is only needed - * on Spark 4.1+ (see SPARK-52921). - */ - def unionRDDs[T: ClassTag]( - sc: SparkContext, - rdds: Seq[RDD[T]], - outputPartitioning: Partitioning): RDD[T] = { - sc.union(rdds) - } -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala deleted file mode 100644 index 70978c5c3d8..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala +++ /dev/null @@ -1,451 +0,0 @@ -/* - * 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.spark.sql.comet.shims - -import java.io.FileNotFoundException - -import scala.util.matching.Regex - -import org.apache.spark.{QueryContext, SparkDateTimeException, SparkException} -import org.apache.spark.sql.catalyst.trees.SQLQueryContext -import org.apache.spark.sql.errors.QueryExecutionErrors -import org.apache.spark.sql.execution.datasources.{DataSourceUtils, SchemaColumnConvertNotSupportedException} -import org.apache.spark.sql.types._ -import org.apache.spark.unsafe.types.UTF8String - -object ShimSparkErrorConverter { - val ObjectLocationPattern: Regex = "Object at location (.+?) not found".r -} - -/** - * Spark 3.4 implementation for converting error types to proper Spark exceptions. - * - * Handles all error cases using the Spark 3.4 QueryExecutionErrors API. Differs from the 3.5 - * version in 4 places where the API changed between 3.4 and 3.5. - */ -trait ShimSparkErrorConverter { - - private def sqlCtx(context: Array[QueryContext]): SQLQueryContext = - context.headOption.map(_.asInstanceOf[SQLQueryContext]).getOrElse(null) - - private def parseFloatLiteral(value: String): Float = { - value.toLowerCase match { - case "inf" | "+inf" | "infinity" | "+infinity" => Float.PositiveInfinity - case "-inf" | "-infinity" => Float.NegativeInfinity - case "nan" | "+nan" | "-nan" => Float.NaN - case _ => value.toFloat - } - } - - private def parseDoubleLiteral(value: String): Double = { - val normalized = value.toLowerCase.stripSuffix("d") - normalized match { - case "inf" | "+inf" | "infinity" | "+infinity" => Double.PositiveInfinity - case "-inf" | "-infinity" => Double.NegativeInfinity - case "nan" | "+nan" | "-nan" => Double.NaN - case _ => normalized.toDouble - } - } - - def convertErrorType( - errorType: String, - errorClass: String, - params: Map[String, Any], - context: Array[QueryContext], - summary: String): Option[Throwable] = { - val _ = (errorClass, summary) - - errorType match { - - case "DivideByZero" => - Some(QueryExecutionErrors.divideByZeroError(sqlCtx(context))) - - case "RemainderByZero" => - Some(QueryExecutionErrors.divideByZeroError(sqlCtx(context))) - - case "IntervalDividedByZero" => - Some(QueryExecutionErrors.intervalDividedByZeroError(sqlCtx(context))) - - case "BinaryArithmeticOverflow" => - // Spark 3.x does not take functionName parameter - Some( - QueryExecutionErrors.binaryArithmeticCauseOverflowError( - params("value1").toString.toShort, - params("symbol").toString, - params("value2").toString.toShort)) - - case "ArithmeticOverflow" => - val fromType = params("fromType").toString - val functionName = params.get("functionName").map(_.toString).getOrElse("") - Some( - QueryExecutionErrors - .arithmeticOverflowError(fromType + " overflow", functionName, sqlCtx(context))) - - case "IntegralDivideOverflow" => - Some(QueryExecutionErrors.overflowInIntegralDivideError(sqlCtx(context))) - - case "DecimalSumOverflow" => - // Spark 3.x takes SQLQueryContext, not QueryContext - Some(QueryExecutionErrors.overflowInSumOfDecimalError(sqlCtx(context))) - - case "NumericValueOutOfRange" => - // Use Java BigDecimal to avoid Scala BigDecimal's DECIMAL128 MathContext rewriting. - val decimal = Decimal(new java.math.BigDecimal(params("value").toString)) - Some( - QueryExecutionErrors.cannotChangeDecimalPrecisionError( - decimal, - params("precision").toString.toInt, - params("scale").toString.toInt, - sqlCtx(context))) - - case "DatetimeOverflow" => - Some( - new SparkException( - errorClass = "DATETIME_OVERFLOW", - messageParameters = params.map { case (k, v) => (k, v.toString) }, - cause = null)) - - case "InvalidArrayIndex" => - Some( - QueryExecutionErrors.invalidArrayIndexError( - params("indexValue").toString.toInt, - params("arraySize").toString.toInt, - sqlCtx(context))) - - case "InvalidElementAtIndex" => - Some( - QueryExecutionErrors.invalidElementAtIndexError( - params("indexValue").toString.toInt, - params("arraySize").toString.toInt, - sqlCtx(context))) - - case "InvalidIndexOfZero" => - Some(QueryExecutionErrors.invalidIndexOfZeroError(sqlCtx(context))) - - case "InvalidBitmapPosition" => - // invalidBitmapPositionError does not exist in Spark 3.4 - Some( - new SparkException( - errorClass = "INVALID_BITMAP_POSITION", - messageParameters = params.map { case (k, v) => (k, v.toString) }, - cause = null)) - - case "DuplicatedMapKey" => - Some(QueryExecutionErrors.duplicateMapKeyFoundError(params("key"))) - - case "NullMapKey" => - Some(QueryExecutionErrors.nullAsMapKeyNotAllowedError()) - - case "MapKeyValueDiffSizes" => - Some(QueryExecutionErrors.mapDataKeyArrayLengthDiffersFromValueArrayLengthError()) - - case "ExceedMapSizeLimit" => - Some(QueryExecutionErrors.exceedMapSizeLimitError(params("size").toString.toInt)) - - case "CollectionSizeLimitExceeded" => - // createArrayWithElementsExceedLimitError takes (count: Any) in Spark 3.4; pass the - // decimal string through since the reported length can exceed Long range. - Some( - QueryExecutionErrors.createArrayWithElementsExceedLimitError( - params("numElements").toString)) - - case "SequenceIllegalBoundaries" => - // Spark 3.x codegen throws a plain IllegalArgumentException for sequence boundaries. - Some( - new IllegalArgumentException( - s"Illegal sequence boundaries: ${params("start")} to ${params("stop")} " + - s"by ${params("step")}")) - - case "SequenceBatchTooLarge" => - // Comet-specific per-batch limit for native `sequence`. Point the user at - // spark.comet.batchSize since Spark itself has no equivalent guard. - Some( - new SparkException( - "Comet's native `sequence` kernel cannot materialize a batch with " + - s"${params("totalElements")} total elements: it exceeds the per-batch " + - "limit or the allocator refused the reservation. Lower " + - "`spark.comet.batchSize` so fewer rows are grouped per batch.", - null)) - - case "Internal" => - Some(SparkException.internalError(params("message").toString)) - - case "NotNullAssertViolation" => - Some( - QueryExecutionErrors.foundNullValueForNotNullableFieldError( - params("fieldName").toString)) - - case "ValueIsNull" => - Some( - QueryExecutionErrors.fieldCannotBeNullError( - params.getOrElse("rowIndex", 0).toString.toInt, - params("fieldName").toString)) - - case "CannotParseTimestamp" => - Some( - QueryExecutionErrors.ansiDateTimeParseError(new Exception(params("message").toString))) - - case "IllegalDayOfWeek" => - Some( - QueryExecutionErrors - .ansiIllegalArgumentError(s"Illegal input for day of week: ${params("string")}")) - - case "DatetimeFieldOutOfBounds" => - Some( - QueryExecutionErrors.ansiDateTimeError( - new java.time.DateTimeException(params("rangeMessage").toString))) - - case "InvalidFractionOfSecond" => - Some(QueryExecutionErrors.invalidFractionOfSecondError()) - - case "CastInvalidValue" => - val str = UTF8String.fromString(params("value").toString) - val targetType = getDataType(params("toType").toString) - Some( - QueryExecutionErrors - .invalidInputInCastToNumberError(targetType, str, sqlCtx(context))) - - case "InvalidInputInCastToDatetime" => - val expression = - s"'${params("value").toString.replace("\\", "\\\\").replace("'", "\\'")}'" - val sourceType = s""""${params("fromType").toString}"""" - val targetType = s""""${params("toType").toString}"""" - Some( - new SparkDateTimeException( - errorClass = "CAST_INVALID_INPUT", - messageParameters = Map( - "expression" -> expression, - "sourceType" -> sourceType, - "targetType" -> targetType, - "ansiConfig" -> "\"spark.sql.ansi.enabled\""), - context = context, - summary = summary)) - - case "CastOverFlow" => - val fromType = getDataType(params("fromType").toString) - val toType = getDataType(params("toType").toString) - val valueStr = params("value").toString - - val typedValue: Any = fromType match { - case _: DecimalType => - val cleanStr = if (valueStr.endsWith("BD")) valueStr.dropRight(2) else valueStr - Decimal(cleanStr) - case ByteType => - val cleanStr = if (valueStr.endsWith("T")) valueStr.dropRight(1) else valueStr - cleanStr.toByte - case ShortType => - val cleanStr = if (valueStr.endsWith("S")) valueStr.dropRight(1) else valueStr - cleanStr.toShort - case IntegerType => valueStr.toInt - case LongType => - val cleanStr = if (valueStr.endsWith("L")) valueStr.dropRight(1) else valueStr - cleanStr.toLong - case FloatType => parseFloatLiteral(valueStr) - case DoubleType => parseDoubleLiteral(valueStr) - case StringType => UTF8String.fromString(valueStr) - case _ => valueStr - } - - Some(QueryExecutionErrors.castingCauseOverflowError(typedValue, fromType, toType)) - - case "CannotParseDecimal" => - Some(QueryExecutionErrors.cannotParseDecimalError()) - - case "InvalidUtf8String" => - // invalidUTF8StringError does not exist in Spark 3.x; use generic fallback - Some( - new SparkException( - errorClass = "INVALID_UTF8_STRING", - messageParameters = params.map { case (k, v) => (k, v.toString) }, - cause = null)) - - case "UnexpectedPositiveValue" => - Some( - QueryExecutionErrors.unexpectedValueForStartInFunctionError( - params("parameterName").toString)) - - case "UnexpectedNegativeValue" => - Some( - QueryExecutionErrors.unexpectedValueForLengthInFunctionError( - params("parameterName").toString)) - - case "InvalidRegexGroupIndex" => - // invalidRegexGroupIndexError does not exist in Spark 3.4 - Some( - new SparkException( - errorClass = "INVALID_REGEX_GROUP_INDEX", - messageParameters = params.map { case (k, v) => (k, v.toString) }, - cause = null)) - - case "InvalidUrl" => - Some( - QueryExecutionErrors.invalidUrlError( - UTF8String.fromString(params("url").toString), - new java.net.URISyntaxException(params("url").toString, "Invalid URL"))) - - case "DatatypeCannotOrder" => - // orderedOperationUnsupportedByDataTypeError takes DataType in Spark 3.4, not String - Some( - QueryExecutionErrors.orderedOperationUnsupportedByDataTypeError( - getDataType(params("dataType").toString))) - - case "ScalarSubqueryTooManyRows" => - // multipleRowScalarSubqueryError was renamed to multipleRowSubqueryError in Spark 3.x - Some(QueryExecutionErrors.multipleRowSubqueryError(sqlCtx(context))) - - case "IntervalArithmeticOverflowWithSuggestion" => - // Spark 3.x uses a single intervalArithmeticOverflowError method - Some( - QueryExecutionErrors.intervalArithmeticOverflowError( - "Interval arithmetic overflow", - params.get("functionName").map(_.toString).getOrElse(""), - sqlCtx(context))) - - case "IntervalArithmeticOverflowWithoutSuggestion" => - Some( - QueryExecutionErrors - .intervalArithmeticOverflowError("Interval arithmetic overflow", "", sqlCtx(context))) - - case "DuplicateFieldCaseInsensitive" => - Some( - QueryExecutionErrors.foundDuplicateFieldInCaseInsensitiveModeError( - params("requiredFieldName").toString, - params("matchedOrcFields").toString)) - - case "DuplicateFieldByFieldId" => - // Mirror Spark's `ParquetReadSupport.matchIdField` which calls - // `foundDuplicateFieldInFieldIdLookupModeError` when more than one Parquet field - // shares an id requested by the read schema. - Some( - QueryExecutionErrors.foundDuplicateFieldInFieldIdLookupModeError( - params("requiredId").toString.toInt, - params("matchedFields").toString)) - - case "ParquetMissingFieldIds" => - // Mirror Spark's `ParquetReadSupport.inferSchema`, which throws a plain - // `RuntimeException` (not a SparkException) when the read schema requests field - // ids and the file carries none. - Some( - new RuntimeException( - "Spark read schema expects field Ids, but Parquet file schema doesn't " + - "contain any field Ids. Please remove the field ids from Spark schema or " + - "ignore missing ids by setting " + - "`spark.sql.parquet.fieldId.read.ignoreMissing = true`")) - - case "ParquetSchemaConvert" => - // Mirror Spark's FileScanRDD: wrap the SchemaColumnConvertNotSupportedException - // in a SparkException whose message is "Parquet column cannot be converted in - // file ...". The native side may not have the file path; an empty path - // still produces a message that contains "Parquet column cannot be converted in - // file" (which is what Spark's own SQL tests assert). - val column = params("column").toString - val physicalType = params("physicalType").toString - val logicalType = params("sparkType").toString - val filePath = params.get("filePath").map(_.toString).getOrElse("") - val cause = - new SchemaColumnConvertNotSupportedException(column, physicalType, logicalType) - Some( - QueryExecutionErrors.unsupportedSchemaColumnConvertError( - filePath, - column, - logicalType, - physicalType, - cause)) - - case "FileNotFound" => - val msg = params("message").toString - // Extract file path from native error message and format like Hadoop's - // FileNotFoundException: "File does not exist" - val path = ShimSparkErrorConverter.ObjectLocationPattern - .findFirstMatchIn(msg) - .map(_.group(1)) - .getOrElse(msg) - Some( - QueryExecutionErrors.readCurrentFileNotFoundError( - new FileNotFoundException(s"File $path does not exist"))) - - case "ParquetTimestampOverflow" => - // Spark 3.4's FileScanRDD wraps a reader error in cannotReadFilesError only while - // nextIterator reads the first batch of a file split; an overflow in a later batch - // reaches the user as the bare ArithmeticException. Comet wraps the overflow from every - // batch, as Spark 4.x does, instead of mirroring those batch boundaries. The difference - // is declared in the Parquet compatibility guide (compatibility/scans.md). - val filePath = params.get("filePath").map(_.toString).getOrElse("") - Some( - QueryExecutionErrors - .cannotReadFilesError(new ArithmeticException("long overflow"), filePath)) - - case "ReadAncientDatetime" => - // Spark raises this unwrapped, not as FAILED_READ_FILE. The helper picks the rebase - // config for the format and throws on a format it does not know. - Some(DataSourceUtils.newRebaseExceptionInRead(params("format").toString)) - - case "CannotReadFile" => - // A per-file read failure of a readable-but-broken file (corrupt/truncated parquet, - // object_store, IO) classified by typed DataFusionError variant on the native side. Wrap - // in the FAILED_READ_FILE SparkException Spark itself produces when its own parquet reader - // fails. (A genuinely-missing file is "FileNotFound" above.) `filePath` is filled by - // SparkErrorConverter from the per-task file list when the native error carried none. - val message = params.get("message").map(_.toString).getOrElse("") - val filePath = params.get("filePath").map(_.toString).getOrElse("") - Some(QueryExecutionErrors.cannotReadFilesError(new SparkException(message), filePath)) - - case _ => - None - } - } - - private def getDataType(typeName: String): DataType = { - typeName.toUpperCase match { - case "BYTE" | "TINYINT" => ByteType - case "SHORT" | "SMALLINT" => ShortType - case "INT" | "INTEGER" => IntegerType - case "LONG" | "BIGINT" => LongType - case "FLOAT" | "REAL" => FloatType - case "DOUBLE" => DoubleType - case "DECIMAL" => DecimalType.SYSTEM_DEFAULT - case "STRING" | "VARCHAR" => StringType - case "BINARY" => BinaryType - case "BOOLEAN" => BooleanType - case "DATE" => DateType - case "TIMESTAMP" => TimestampType - case _ => - try { - DataType.fromDDL(typeName) - } catch { - case _: Exception => - // fromDDL rejects types that are syntactically invalid in SQL DDL, such as - // DECIMAL(p,s) with a negative scale (valid when allowNegativeScaleOfDecimal=true). - // Parse those manually rather than silently falling back to StringType. - if (typeName.toUpperCase.startsWith("DECIMAL(") && typeName.endsWith(")")) { - val inner = typeName.substring("DECIMAL(".length, typeName.length - 1) - val parts = inner.split(",") - if (parts.length == 2) { - try { - DataTypes.createDecimalType(parts(0).trim.toInt, parts(1).trim.toInt) - } catch { - case _: Exception => StringType - } - } else StringType - } else StringType - } - } - } -} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimTaskMetrics.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimTaskMetrics.scala deleted file mode 100644 index 5b2a5fb5bf5..00000000000 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimTaskMetrics.scala +++ /dev/null @@ -1,29 +0,0 @@ -/* - * 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.spark.sql.comet.shims - -import org.apache.spark.executor.TaskMetrics -import org.apache.spark.util.AccumulatorV2 - -object ShimTaskMetrics { - - def getTaskAccumulator(taskMetrics: TaskMetrics): Option[AccumulatorV2[_, _]] = - taskMetrics.externalAccums.lastOption -} diff --git a/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala index 59d8c7f251b..3f8ddb6561f 100644 --- a/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala +++ b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala @@ -30,15 +30,13 @@ import org.apache.spark.sql.vectorized.ColumnarBatch /** * Spark 3.x stub for the PyArrow UDF acceleration support. * - * The columnar runner introduced in #4234 only targets Spark 4.0+. On Spark 3.4 / 3.5 the + * The columnar runner introduced in #4234 only targets Spark 4.0+. On Spark 3.5 the * matchers return `None`, the rewrite does not fire, and vanilla Spark handles `mapInArrow` / * `mapInPandas` unchanged. The runner factory throws; it is never called because the matchers * always return `None`. 3.x support can be added later if there is user demand. * - * Shared across spark-3.4 and spark-3.5 because both are identical: 3.4 lacks the modern - * `ArrowPythonRunner` constructor and `arrowUseLargeVarTypes`, and 3.5's `PythonArrowInput` trait - * has a different contract (`writeIteratorToArrowStream` one-shot vs 4.x's - * `writeNextBatchToArrowStream` batch-at-a-time), so neither version can host the columnar input + * 3.5's `PythonArrowInput` trait has a different contract (`writeIteratorToArrowStream` one-shot + * vs 4.x's `writeNextBatchToArrowStream` batch-at-a-time), so it cannot host the columnar input * implementation without a separate rewrite. */ trait ShimCometMapInBatch { diff --git a/spark/src/test/spark-3.4/org/apache/comet/exec/CometMergeRowsSuite.scala b/spark/src/test/spark-3.4/org/apache/comet/exec/CometMergeRowsSuite.scala deleted file mode 100644 index b97c0e5bb5a..00000000000 --- a/spark/src/test/spark-3.4/org/apache/comet/exec/CometMergeRowsSuite.scala +++ /dev/null @@ -1,31 +0,0 @@ -/* - * 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.exec - -import org.apache.spark.sql.CometTestBase - -import org.apache.comet.shims.ShimCometMergeRows - -/** Spark 3.4 compatibility coverage for the version-specific MergeRows registration shim. */ -class CometMergeRowsSuite extends CometTestBase { - test("Spark 3.4 leaves native MergeRows execution unregistered") { - assert(ShimCometMergeRows.nativeExecs.isEmpty) - } -} diff --git a/spark/src/test/spark-3.4/org/apache/comet/shims/ShimCometTPCHQuerySuite.scala b/spark/src/test/spark-3.4/org/apache/comet/shims/ShimCometTPCHQuerySuite.scala deleted file mode 100644 index f19f03ff2fd..00000000000 --- a/spark/src/test/spark-3.4/org/apache/comet/shims/ShimCometTPCHQuerySuite.scala +++ /dev/null @@ -1,30 +0,0 @@ -/* - * 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.shims - -import org.apache.spark.sql.{SparkSession, SQLQueryTestHelper} - -trait ShimCometTPCHQuerySuite extends SQLQueryTestHelper { - protected def getNormalizedQueryExecutionResult( - session: SparkSession, - sql: String): (String, Seq[String]) = { - getNormalizedResult(session, sql) - } -} diff --git a/spark/src/test/spark-3.4/org/apache/spark/sql/ShimCometTestBase.scala b/spark/src/test/spark-3.4/org/apache/spark/sql/ShimCometTestBase.scala deleted file mode 100644 index ce56eeb4364..00000000000 --- a/spark/src/test/spark-3.4/org/apache/spark/sql/ShimCometTestBase.scala +++ /dev/null @@ -1,61 +0,0 @@ -/* - * 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.spark.sql - -import org.apache.spark.SparkConf -import org.apache.spark.sql.catalyst.expressions.{Expression, MakeDecimal} -import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan - -trait ShimCometTestBase { - type SparkSessionType = SparkSession - - def createSparkSessionWithExtensions(conf: SparkConf): SparkSessionType = { - SparkSession - .builder() - .config(conf) - .master("local[1]") - .withExtensions(new org.apache.comet.CometSparkSessionExtensions) - .getOrCreate() - } - - def datasetOfRows(spark: SparkSession, plan: LogicalPlan): DataFrame = { - Dataset.ofRows(spark, plan) - } - - def getColumnFromExpression(expr: Expression): Column = { - new Column(expr) - } - - def extractLogicalPlan(df: DataFrame): LogicalPlan = { - df.logicalPlan - } - - def createMakeDecimalColumn(child: Expression, precision: Int, scale: Int): Column = { - new Column(MakeDecimal(child, precision, scale)) - } - - def createMakeDecimalColumn( - child: Expression, - precision: Int, - scale: Int, - nullOnOverflow: Boolean): Column = { - new Column(MakeDecimal(child, precision, scale, nullOnOverflow)) - } -} From 63c524a3221378c3871032fc71c8023075757dd6 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 9 Nov 2026 05:52:36 -0700 Subject: [PATCH 4/4] style: apply spotless formatting to ShimCometMapInBatch --- .../apache/spark/sql/comet/shims/ShimCometMapInBatch.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala index 3f8ddb6561f..4b522347568 100644 --- a/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala +++ b/spark/src/main/spark-3.x/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala @@ -30,8 +30,8 @@ import org.apache.spark.sql.vectorized.ColumnarBatch /** * Spark 3.x stub for the PyArrow UDF acceleration support. * - * The columnar runner introduced in #4234 only targets Spark 4.0+. On Spark 3.5 the - * matchers return `None`, the rewrite does not fire, and vanilla Spark handles `mapInArrow` / + * The columnar runner introduced in #4234 only targets Spark 4.0+. On Spark 3.5 the matchers + * return `None`, the rewrite does not fire, and vanilla Spark handles `mapInArrow` / * `mapInPandas` unchanged. The runner factory throws; it is never called because the matchers * always return `None`. 3.x support can be added later if there is user demand. *