{"owner":"apache","repo":"spark","hasSkills":true,"totalSkillsCount":6,"totalTokensCount":13570,"categories":["root-instruction","claude-rule","mcp-config","marketplace","plugin-manifest","cursor-rule"],"hasMcp":true,"mcpConfig":{"mcpServers":{"spark":{"command":"npx","args":["-y","@modelcontextprotocol/server-spark"]}}},"found":["AGENTS.md","CLAUDE.md",".cursor/mcp.json",".claude-plugin/marketplace.json",".claude-plugin/plugin.json",".cursorrules (pyspark-etl-best-practices-cursorrules-prompt-file.mdc)"],"skills":{"AGENTS.md":"# Apache Spark\n\n## Pre-flight Checks\n\nBefore the first code edit or running test in a session, ensure a clean working environment. DO NOT skip these checks:\n\n1. Run `git remote -v` to identify the personal fork and upstream (`apache/spark`). If unclear, ask the user to configure their remotes following the standard convention (`origin` for the fork, `upstream` for `apache/spark`).\n2. If the latest commit on `<upstream>/master` is more than a day old (check with `git log -1 --format=\"%ci\" <upstream>/master`), run `git fetch <upstream> master`.\n3. If there are uncommitted changes (check with `git status`), ask the user to stash them before proceeding.\n4. Switch to the appropriate branch:\n   - **Existing PR**: resolve the PR branch name via `gh api repos/apache/spark/pulls/<number> --jq '.head.ref'`, then look for a local branch matching that name. If found, switch to it and inform the user. If not found, ask whether to fetch it or if there is a local branch under a different name.\n   - **New edits**: ask the user to choose: create a new git worktree from `<upstream>/master` with `--no-track` and work from there (recommended), or create and switch to a new branch from `<upstream>/master` with `--no-track`.\n   - **Running tests**: use `<upstream>/master`.\n\n## Development Notes\n\nSQL golden file tests are managed by `SQLQueryTestSuite` and its variants. Read the class documentation before running or updating these tests. DO NOT edit the generated golden files (`.sql.out`) directly. Always regenerate them when needed, and carefully review the diff to make sure it's expected.\n\nSpark Connect protocol is defined in proto files under `sql/connect/common/src/main/protobuf/`. Read the README there before modifying proto definitions.\n\nWhen adding a member to an existing class or object, follow the sectioning the file already uses and put the new member with the code it belongs with. The common failure mode is dropping it wherever it is first used without checking how the file is organized, splitting a section of unrelated code in the process. Beyond grouping related code there is no prescribed order -- the Databricks Scala guide, which Spark follows, asks only that a long class group its members into logical sections with comment headers. Do not reorganize existing members unless the change requires it.\n\nAvoid introducing non-ASCII characters in code or comments. String literals may contain non-ASCII when the content requires it (error messages, test data, etc.). Identifiers are ASCII by convention. The common failure mode is typographic characters (em-dash, smart quotes, ellipsis, non-breaking space) sneaking into comments; scalastyle flags some of these. Spot-check before committing: `grep -rn -P \"[^\\x00-\\x7F]\" <files>`.\n\nKeep source lines within 100 characters — the linters enforce this for Scala, Java, and Python, and LLMs commonly overrun it in comments and long expressions. A quick scan of just the changed files catches most cases in seconds, far cheaper than a CI round trip:\n\n    { git diff --name-only --diff-filter=ACM HEAD; git ls-files --others --exclude-standard; } \\\n      | grep -E '\\.(scala|java|py)$' | sort -u \\\n      | xargs -r awk 'length>100 && $0 !~ /^[[:space:]]*(import|package) / && $0 !~ /https?:\\/\\// \\\n          {print FILENAME\":\"FNR\": \"length\" chars\"}'\n\nThis is only a hint: it approximates the linters' exemptions (imports, URLs) rather than matching them exactly, so it can over- or under-report. The linters remain the source of truth.\n\n## Scala Test Base Classes\n\nWhen writing a new Scala test suite, pick the lowest base class that provides what the test actually needs. Spark uses the `AnyFunSuite` ScalaTest style throughout, so the bases below are the chain to choose from. Each adds capability on top of the previous:\n\n    SparkFunSuite                                                           (core)\n      <- PlanTest                                                           (sql/catalyst)\n        <- QueryTest                                                        (sql/core)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain JVM/Scala — no Spark SQL | `SparkFunSuite` | `core` utilities, RDD, network, util classes, etc. Adds per-test timeout, `testRetry`, `gridTest`, thread audit, fixed timezone/locale, `withTempDir`, `withLogAppender`, `checkError`. |\n| Catalyst plan tests — no `SparkSession` | `PlanTest` | Adds `comparePlans`, `normalizePlan`, `normalizeExprIds`. For analyzer / optimizer / planner rule tests. |\n| SQL/DataFrame tests — needs a `SparkSession` | `QueryTest` | Adds `checkAnswer`, codegen-on/off helpers. `spark: SparkSession` is abstract and must be supplied by a session-providing trait (see below). |\n\n### Providing a `SparkSession` for `QueryTest`\n\n`QueryTest` declares `spark: SparkSession` abstractly via `SparkSessionProvider`, so it cannot be instantiated on its own. A concrete suite mixes in one of the session-providing traits below:\n\n    QueryTest                                                               (abstract `spark`)\n      + SharedSparkSession (sql/core)        -> classic in-process `TestSparkSession`\n      + TestHiveSingleton  (sql/hive)        -> Hive-backed `TestHive` session\n\n| Session provider | Module / location | Typical usage |\n|---|---|---|\n| `SharedSparkSession` | `sql/core` | Already extends `QueryTest` for historical reasons, but still mix in `QueryTest` explicitly, e.g. `class X extends QueryTest with SharedSparkSession`. Default for tests under `sql/core`. |\n| `TestHiveSingleton` | `sql/hive` | Mixed in alongside `QueryTest`, e.g. `class X extends QueryTest with TestHiveSingleton`. Used by tests under `sql/hive`. |\n\n## Python Test Base Classes\n\nPySpark tests use the stdlib `unittest` framework: every suite subclasses `unittest.TestCase` (run via `python/run-tests`, see below). As with Scala, pick the lowest base that provides what the test actually needs. The bases live under `python/pyspark/testing/` and each adds capability on top of the previous:\n\n    unittest.TestCase                                     (stdlib)\n      <- PySparkBaseTestCase                              (pyspark.testing.utils)\n        <- ReusedPySparkTestCase                          (pyspark.testing.utils)\n          <- ReusedSQLTestCase                            (pyspark.testing.sqlutils)\n            <- PandasOnSparkTestCase                      (pyspark.testing.pandasutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain Python — no Spark | `unittest.TestCase` | Use the stdlib base directly. `PySparkBaseTestCase` is the same thing plus a SIGTERM fault-handler dump enabled when `PYSPARK_TEST_TIMEOUT` is set; subclass it only when you want that. |\n| RDD / `SparkContext` — no `SparkSession` | `PySparkTestCase` (fresh) or `ReusedPySparkTestCase` (shared) | Both create a `SparkContext(\"local[4]\")`. `PySparkTestCase` makes a new one per test (isolation); `ReusedPySparkTestCase` shares one per class (faster, the usual choice) and adds `quiet()` and an overridable `conf()` / `master()`. |\n| SQL / DataFrame — needs a `SparkSession` | `ReusedSQLTestCase` | The workhorse for classic tests under `python/pyspark/sql`. Adds a shared `cls.spark`, sample `cls.df` / `cls.testData`, and mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. Classic (non-Connect) mode. |\n| pandas API on Spark | `PandasOnSparkTestCase` | Extends `ReusedSQLTestCase` with Arrow enabled and pandas-on-Spark assertions (`PandasOnSparkTestUtils`); `ComparisonTestBase` builds on it. |\n\n### Spark Connect test bases\n\nSpark Connect suites live in `pyspark.testing.connectutils` and are auto-skipped (via `should_test_connect`) when Connect dependencies are missing:\n\n    PySparkBaseTestCase                                   (pyspark.testing.utils)\n      <- PlanOnlyTestFixture                              (pyspark.testing.connectutils)\n      <- ReusedConnectTestCase                            (pyspark.testing.connectutils)\n           <- ReusedMixedTestCase                         (pyspark.testing.connectutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plan / proto construction — no server | `PlanOnlyTestFixture` | Uses a `MockRemoteSession`; builds and inspects plans without a running Connect server. For proto / plan-shape assertions. |\n| Connect DataFrame — real session | `ReusedConnectTestCase` | The Connect analog of `ReusedSQLTestCase`; starts a session via `.remote(...)` (honoring `SPARK_CONNECT_TESTING_REMOTE`, default `local[4]`). Mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. |\n| Classic + Connect side by side | `ReusedMixedTestCase` | Extends `ReusedConnectTestCase`. For directly comparing classic vs Connect: it exposes a classic `self.spark` and a Connect `self.connect` so a test can run the same operation on each and assert they agree (`compare_by_show`, `both_conf`). Requires JVM access. |\n\n### Mixins and helpers\n\nThese are combined with a base above rather than used on their own:\n\n- `SQLTestUtils` — context managers `sql_conf`, `table`, `temp_view`, `view`, `database`, `function`, `temp_func`, `temp_env`; assumes `self.spark`. Already mixed into `ReusedSQLTestCase` and `ReusedConnectTestCase`.\n- `PySparkErrorTestUtils` — `check_error(...)` to assert on a `PySparkException`'s error class and message parameters. The Python counterpart of Scala's `checkError`.\n- `assertDataFrameEqual` / `assertSchemaEqual` (public API, from `pyspark.testing`) — standalone assertion functions that work in any test, no particular base class required.\n- Domain bases outside the main ladder: `SparkSessionTestCase` (`pyspark.testing.mlutils`, for `ml`), `MLlibTestCase` (`pyspark.testing.mllibutils`), and `PySparkStreamingTestCase` (`pyspark.testing.streamingutils`, DStreams).\n\n## Build and Test\n\nBuild and tests can take a long time, so prefer the narrowest relevant build or test command.\n\nFor build and test setup, including how to run tests and troubleshoot common\nlocal failures, see `docs/building-spark.md`.\n\nPrefer SBT over Maven for faster incremental compilation. Module names are defined in `project/SparkBuild.scala`.\n\nCompile a single module:\n\n    build/sbt <module>/compile\n\nCompile test code for a single module:\n\n    build/sbt <module>/Test/compile\n\nRun test suites by wildcard or full class name:\n\n    build/sbt '<module>/testOnly *MySuite'\n    build/sbt '<module>/testOnly org.apache.spark.sql.MySuite'\n\nRun test cases matching a substring:\n\n    build/sbt '<module>/testOnly *MySuite -- -z \"test name\"'\n\nRun test cases in an optional module:\n\n    build/sbt -P<maven-profiles> '<module>/testOnly *MySuite'\n\nFor faster iteration, keep SBT open in interactive mode:\n\n    build/sbt\n    > project <module>\n    > testOnly *MySuite\n\n### PySpark Tests\n\nPySpark tests require building Spark with Hive support first:\n\n    build/sbt -Phive package\n\nActivate the virtual environment specified by the user, or default to `.venv`:\n\n    source <venv>/bin/activate\n\nIf the default venv does not exist, create it:\n\n    python3 -m venv .venv\n    source .venv/bin/activate\n    pip install --upgrade pip\n    pip install --group dev\n\nRun a single test suite:\n\n    python/run-tests --testnames pyspark.sql.tests.arrow.test_arrow\n\nRun a single test case:\n\n    python/run-tests --testnames \"pyspark.sql.tests.test_catalog CatalogTests.test_current_database\"\n\n## Investigating PR CI Failures\n\nEnumerate all failing check runs first, then drill into each by type. Do not assume a single failure: a PR can fail tests, linters, and the build at once, and these surface through different channels.\n\nStep 1 — Get the fork owner and the latest commit SHA of the PR:\n\n    gh api repos/apache/spark/pulls/<PR_NUMBER> --jq '{owner: .head.repo.owner.login, sha: .head.sha}'\n\nStep 2 — List every failing check run on the fork's commit. This is the complete failure set:\n\n    gh api repos/<OWNER>/spark/commits/<SHA>/check-runs --paginate \\\n      --jq '.check_runs[] | select(.conclusion == \"failure\") | {name, id: .id}'\n\nA passing (or absent) \"Report test results\" does NOT mean CI is green. That check aggregates only test-case failures; linter, license, dependency, MiMa, compile, and doc-build failures are separate check runs that produce no test annotations. Always work from the list in Step 2, not from any single check.\n\nStep 3 — Drill into each failure according to its kind:\n\n- **Test jobs** (e.g. \"Report test results\", \"Build modules: ...\"): fetch failure annotations. Each annotation contains the test class, test name, and failure message:\n\n      gh api repos/<OWNER>/spark/check-runs/<CHECK_RUN_ID>/annotations\n\n- **Non-test jobs** (e.g. \"Linters, licenses, and dependencies\", \"Build\"): find the failed step, then read only that job's log:\n\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID> \\\n        --jq '{name, steps: [.steps[] | select(.conclusion == \"failure\") | .name]}'\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID>/logs\n\nAvoid downloading the large per-shard *test* job logs — they are very large and slow; use the annotations for those. Lint, license, dependency, and build job logs are small and fine to read directly when a step fails.\n\n## Checking PR Merge Status\n\nSpark merges PRs with `dev/merge_spark_pr.py`, not the GitHub merge button, so a **merged PR shows up on GitHub as Closed, not Merged** — its `merged` / `mergedAt` are empty, and backports to maintenance branches are plain pushes. **Do not read a Closed PR as rejected**; most Closed PRs were in fact merged.\n\nTo check whether, and to which branches, a PR was merged, run `dev/pr_merge_status.py <pr-number>`:\n\n    $ dev/pr_merge_status.py 56356\n    PR #56356 (base master): [SPARK-57295][SQL] Make database location validation ...\n    merged: yes\n      master       9357bc9ae05\n      branch-4.x   72edddb358e\n\nIt lists `master` and the latest major's release branches the commit reached (e.g. `branch-4.x`, `branch-4.2`) — or reports `open` or `closed without merging`. It needs `gh` authenticated and a non-shallow checkout, and exits non-zero with a clear message if the PR is unknown (404) or the environment isn't ready (no `gh`/auth, no apache/spark remote). Older majors' branches are omitted; pass `--all-branches` to list every branch the commit reached.\n\n## Pull Request Workflow\n\nPR title format is `[SPARK-xxxx][COMPONENT] Title`. Draft, WIP, MINOR, and TRIVIAL PRs may omit the JIRA ID. The component tag comes from the `COMPONENTS` registry in `dev/merge_spark_pr.py`, the source of truth: each entry pairs a PR-title tag with its canonical JIRA component name and accepted aliases (e.g. `Documentation` → `[DOC]`, `Pandas API on Spark` → `[PS]`, `SQL` → `[SQL]`). Read that registry to pick the right tag.\n\nUse `[FOLLOWUP]` only for small PRs that directly modify or correct unreleased earlier PRs. For separately planned work, non-trivial changes, or work outside the earlier JIRA's scope, create a separate JIRA ticket for each PR and use the normal title format without `[FOLLOWUP]`.\n\nInfer the PR title from the changes. If no ticket ID is given and the PR is not draft, WIP, MINOR, or TRIVIAL, create one using `dev/create_spark_jira.py`, using the PR title (without the JIRA ID and component tag) as the ticket title.\n\n    python3 dev/create_spark_jira.py \"<title>\" -c <component> { -t <type> | -p <parent-jira-id> }\n\n- **Component** (`-c`): the exact JIRA component name (not the PR title shorthand), e.g. \"SQL\", \"Spark Core\", \"PySpark\", \"Connect\". Run `python3 dev/create_spark_jira.py --list-components` for the full list.\n- **Issue type** (`-t`): \"Bug\", \"Improvement\", \"New Feature\", \"Test\", \"Documentation\", or \"Dependency upgrade\".\n- **Parent** (`-p`): if the user mentions a parent JIRA ticket (e.g., \"this is a subtask of SPARK-12345\"), pass it instead of `-t`. The issue type is automatically \"Sub-task\".\n\nThe script sets the latest unreleased version as the default affected version.\n\nAfter creating a JIRA ticket, print a prominent notice so the user does not miss it:\n\n    ============================================================\n    JIRA ticket created: SPARK-XXXXX\n    https://issues.apache.org/jira/browse/SPARK-XXXXX\n\n    Title:              <title>\n    Component(s):       <component>\n    Issue type:         <type>\n    Affected version(s): <version>\n    Priority:           <priority>\n\n    Please review and adjust these fields if needed.\n    ============================================================\n\nBefore writing the PR description, read `.github/PULL_REQUEST_TEMPLATE` and fill in every section from that file.\n\nDO NOT push to the upstream repo. Always push to the personal fork. Open PRs against `master` on the upstream repo.\n\nDO NOT force push or use `--amend` on pushed commits unless the user explicitly asks. If the remote branch has new commits, fetch and rebase before pushing.\n\nAlways get user approval before external operations such as pushing commits, creating PRs, or posting comments. Use `gh pr create` to open PRs. If `gh` is not installed, generate the GitHub PR URL for the user and recommend installing the GitHub CLI.\n\n## Project Instructions Discovery\n\nWhen exploring or working in any directory, always check for nested `AGENTS.md` files in that\ndirectory and its ancestors. Read and follow every applicable file; instructions in a more specific\ndirectory take precedence for that directory's scope.\n\nA directory may carry these instructions as `CLAUDE.md` rather than `AGENTS.md`, so check **both** names — a directory with only a `CLAUDE.md` has project instructions you must read too. Where both files exist, one is typically the real file and the other is a symlink to it, although either name may be the real file. Reading either one is sufficient. When adding instructions to a directory that has neither file, create AGENTS.md as the real file and add a CLAUDE.md symlink beside it (ln -s AGENTS.md CLAUDE.md). This keeps a single source of truth accessible under both names.\n\n## Versioning and Branch Policy\n\nWhen a change needs a version — `@since` annotations, config `.version(\"...\")` (`SQLConf` / `*Conf`), new `MimaExcludes` sections, etc. — use the version of the branch it first ships in, with `-SNAPSHOT` stripped. Determine that branch:\n\n- **PR opened against a non-`master` base branch** (e.g. a maintenance line like `branch-4.2`): use that base branch's version -- when you're checked out on it, that's just the working tree's `pom.xml`. The helper below covers only the common `master`-base case.\n- **PR opened against `master`:** most PRs merge to **both** `master` and the latest `branch-<N>.x` (the branch for the next feature release, e.g. `branch-4.x`), so use the `branch-<N>.x` version. The exception is **master-only** changes — use `master`'s version — which are only:\n  - breaking / binary-incompatible changes that can't ship in a minor release;\n  - dependency upgrades that don't fix a critical issue worth backporting.\n\nDo **not** just read `master`'s version: a normally-backported PR ships first in `branch-<N>.x`, whose version is lower than `master`'s. If unsure whether a change is master-only, ask the user.\n\n`dev/next_version_candidates.py` (no arguments) prints both candidate versions, reading from the local `apache/spark` remote (the `upstream` configured during pre-flight). It reports the mechanical facts only -- choosing between them per the rules above is the judgement call (the numbers below are illustrative and advance over time):\n\n    $ dev/next_version_candidates.py\n    master       5.0.0\n    branch-4.x   4.3.0\n\n## Security\n\nSecurity model: [SECURITY.md](./SECURITY.md)\n\nAgents that scan this repository should consult `SECURITY.md` for the project's threat model, in-scope / out-of-scope declarations, and known non-findings before reporting issues.\n","CLAUDE.md":"# Apache Spark\n\n## Pre-flight Checks\n\nBefore the first code edit or running test in a session, ensure a clean working environment. DO NOT skip these checks:\n\n1. Run `git remote -v` to identify the personal fork and upstream (`apache/spark`). If unclear, ask the user to configure their remotes following the standard convention (`origin` for the fork, `upstream` for `apache/spark`).\n2. If the latest commit on `<upstream>/master` is more than a day old (check with `git log -1 --format=\"%ci\" <upstream>/master`), run `git fetch <upstream> master`.\n3. If there are uncommitted changes (check with `git status`), ask the user to stash them before proceeding.\n4. Switch to the appropriate branch:\n   - **Existing PR**: resolve the PR branch name via `gh api repos/apache/spark/pulls/<number> --jq '.head.ref'`, then look for a local branch matching that name. If found, switch to it and inform the user. If not found, ask whether to fetch it or if there is a local branch under a different name.\n   - **New edits**: ask the user to choose: create a new git worktree from `<upstream>/master` with `--no-track` and work from there (recommended), or create and switch to a new branch from `<upstream>/master` with `--no-track`.\n   - **Running tests**: use `<upstream>/master`.\n\n## Development Notes\n\nSQL golden file tests are managed by `SQLQueryTestSuite` and its variants. Read the class documentation before running or updating these tests. DO NOT edit the generated golden files (`.sql.out`) directly. Always regenerate them when needed, and carefully review the diff to make sure it's expected.\n\nSpark Connect protocol is defined in proto files under `sql/connect/common/src/main/protobuf/`. Read the README there before modifying proto definitions.\n\nWhen adding a member to an existing class or object, follow the sectioning the file already uses and put the new member with the code it belongs with. The common failure mode is dropping it wherever it is first used without checking how the file is organized, splitting a section of unrelated code in the process. Beyond grouping related code there is no prescribed order -- the Databricks Scala guide, which Spark follows, asks only that a long class group its members into logical sections with comment headers. Do not reorganize existing members unless the change requires it.\n\nAvoid introducing non-ASCII characters in code or comments. String literals may contain non-ASCII when the content requires it (error messages, test data, etc.). Identifiers are ASCII by convention. The common failure mode is typographic characters (em-dash, smart quotes, ellipsis, non-breaking space) sneaking into comments; scalastyle flags some of these. Spot-check before committing: `grep -rn -P \"[^\\x00-\\x7F]\" <files>`.\n\nKeep source lines within 100 characters — the linters enforce this for Scala, Java, and Python, and LLMs commonly overrun it in comments and long expressions. A quick scan of just the changed files catches most cases in seconds, far cheaper than a CI round trip:\n\n    { git diff --name-only --diff-filter=ACM HEAD; git ls-files --others --exclude-standard; } \\\n      | grep -E '\\.(scala|java|py)$' | sort -u \\\n      | xargs -r awk 'length>100 && $0 !~ /^[[:space:]]*(import|package) / && $0 !~ /https?:\\/\\// \\\n          {print FILENAME\":\"FNR\": \"length\" chars\"}'\n\nThis is only a hint: it approximates the linters' exemptions (imports, URLs) rather than matching them exactly, so it can over- or under-report. The linters remain the source of truth.\n\n## Scala Test Base Classes\n\nWhen writing a new Scala test suite, pick the lowest base class that provides what the test actually needs. Spark uses the `AnyFunSuite` ScalaTest style throughout, so the bases below are the chain to choose from. Each adds capability on top of the previous:\n\n    SparkFunSuite                                                           (core)\n      <- PlanTest                                                           (sql/catalyst)\n        <- QueryTest                                                        (sql/core)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain JVM/Scala — no Spark SQL | `SparkFunSuite` | `core` utilities, RDD, network, util classes, etc. Adds per-test timeout, `testRetry`, `gridTest`, thread audit, fixed timezone/locale, `withTempDir`, `withLogAppender`, `checkError`. |\n| Catalyst plan tests — no `SparkSession` | `PlanTest` | Adds `comparePlans`, `normalizePlan`, `normalizeExprIds`. For analyzer / optimizer / planner rule tests. |\n| SQL/DataFrame tests — needs a `SparkSession` | `QueryTest` | Adds `checkAnswer`, codegen-on/off helpers. `spark: SparkSession` is abstract and must be supplied by a session-providing trait (see below). |\n\n### Providing a `SparkSession` for `QueryTest`\n\n`QueryTest` declares `spark: SparkSession` abstractly via `SparkSessionProvider`, so it cannot be instantiated on its own. A concrete suite mixes in one of the session-providing traits below:\n\n    QueryTest                                                               (abstract `spark`)\n      + SharedSparkSession (sql/core)        -> classic in-process `TestSparkSession`\n      + TestHiveSingleton  (sql/hive)        -> Hive-backed `TestHive` session\n\n| Session provider | Module / location | Typical usage |\n|---|---|---|\n| `SharedSparkSession` | `sql/core` | Already extends `QueryTest` for historical reasons, but still mix in `QueryTest` explicitly, e.g. `class X extends QueryTest with SharedSparkSession`. Default for tests under `sql/core`. |\n| `TestHiveSingleton` | `sql/hive` | Mixed in alongside `QueryTest`, e.g. `class X extends QueryTest with TestHiveSingleton`. Used by tests under `sql/hive`. |\n\n## Python Test Base Classes\n\nPySpark tests use the stdlib `unittest` framework: every suite subclasses `unittest.TestCase` (run via `python/run-tests`, see below). As with Scala, pick the lowest base that provides what the test actually needs. The bases live under `python/pyspark/testing/` and each adds capability on top of the previous:\n\n    unittest.TestCase                                     (stdlib)\n      <- PySparkBaseTestCase                              (pyspark.testing.utils)\n        <- ReusedPySparkTestCase                          (pyspark.testing.utils)\n          <- ReusedSQLTestCase                            (pyspark.testing.sqlutils)\n            <- PandasOnSparkTestCase                      (pyspark.testing.pandasutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain Python — no Spark | `unittest.TestCase` | Use the stdlib base directly. `PySparkBaseTestCase` is the same thing plus a SIGTERM fault-handler dump enabled when `PYSPARK_TEST_TIMEOUT` is set; subclass it only when you want that. |\n| RDD / `SparkContext` — no `SparkSession` | `PySparkTestCase` (fresh) or `ReusedPySparkTestCase` (shared) | Both create a `SparkContext(\"local[4]\")`. `PySparkTestCase` makes a new one per test (isolation); `ReusedPySparkTestCase` shares one per class (faster, the usual choice) and adds `quiet()` and an overridable `conf()` / `master()`. |\n| SQL / DataFrame — needs a `SparkSession` | `ReusedSQLTestCase` | The workhorse for classic tests under `python/pyspark/sql`. Adds a shared `cls.spark`, sample `cls.df` / `cls.testData`, and mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. Classic (non-Connect) mode. |\n| pandas API on Spark | `PandasOnSparkTestCase` | Extends `ReusedSQLTestCase` with Arrow enabled and pandas-on-Spark assertions (`PandasOnSparkTestUtils`); `ComparisonTestBase` builds on it. |\n\n### Spark Connect test bases\n\nSpark Connect suites live in `pyspark.testing.connectutils` and are auto-skipped (via `should_test_connect`) when Connect dependencies are missing:\n\n    PySparkBaseTestCase                                   (pyspark.testing.utils)\n      <- PlanOnlyTestFixture                              (pyspark.testing.connectutils)\n      <- ReusedConnectTestCase                            (pyspark.testing.connectutils)\n           <- ReusedMixedTestCase                         (pyspark.testing.connectutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plan / proto construction — no server | `PlanOnlyTestFixture` | Uses a `MockRemoteSession`; builds and inspects plans without a running Connect server. For proto / plan-shape assertions. |\n| Connect DataFrame — real session | `ReusedConnectTestCase` | The Connect analog of `ReusedSQLTestCase`; starts a session via `.remote(...)` (honoring `SPARK_CONNECT_TESTING_REMOTE`, default `local[4]`). Mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. |\n| Classic + Connect side by side | `ReusedMixedTestCase` | Extends `ReusedConnectTestCase`. For directly comparing classic vs Connect: it exposes a classic `self.spark` and a Connect `self.connect` so a test can run the same operation on each and assert they agree (`compare_by_show`, `both_conf`). Requires JVM access. |\n\n### Mixins and helpers\n\nThese are combined with a base above rather than used on their own:\n\n- `SQLTestUtils` — context managers `sql_conf`, `table`, `temp_view`, `view`, `database`, `function`, `temp_func`, `temp_env`; assumes `self.spark`. Already mixed into `ReusedSQLTestCase` and `ReusedConnectTestCase`.\n- `PySparkErrorTestUtils` — `check_error(...)` to assert on a `PySparkException`'s error class and message parameters. The Python counterpart of Scala's `checkError`.\n- `assertDataFrameEqual` / `assertSchemaEqual` (public API, from `pyspark.testing`) — standalone assertion functions that work in any test, no particular base class required.\n- Domain bases outside the main ladder: `SparkSessionTestCase` (`pyspark.testing.mlutils`, for `ml`), `MLlibTestCase` (`pyspark.testing.mllibutils`), and `PySparkStreamingTestCase` (`pyspark.testing.streamingutils`, DStreams).\n\n## Build and Test\n\nBuild and tests can take a long time, so prefer the narrowest relevant build or test command.\n\nFor build and test setup, including how to run tests and troubleshoot common\nlocal failures, see `docs/building-spark.md`.\n\nPrefer SBT over Maven for faster incremental compilation. Module names are defined in `project/SparkBuild.scala`.\n\nCompile a single module:\n\n    build/sbt <module>/compile\n\nCompile test code for a single module:\n\n    build/sbt <module>/Test/compile\n\nRun test suites by wildcard or full class name:\n\n    build/sbt '<module>/testOnly *MySuite'\n    build/sbt '<module>/testOnly org.apache.spark.sql.MySuite'\n\nRun test cases matching a substring:\n\n    build/sbt '<module>/testOnly *MySuite -- -z \"test name\"'\n\nRun test cases in an optional module:\n\n    build/sbt -P<maven-profiles> '<module>/testOnly *MySuite'\n\nFor faster iteration, keep SBT open in interactive mode:\n\n    build/sbt\n    > project <module>\n    > testOnly *MySuite\n\n### PySpark Tests\n\nPySpark tests require building Spark with Hive support first:\n\n    build/sbt -Phive package\n\nActivate the virtual environment specified by the user, or default to `.venv`:\n\n    source <venv>/bin/activate\n\nIf the default venv does not exist, create it:\n\n    python3 -m venv .venv\n    source .venv/bin/activate\n    pip install --upgrade pip\n    pip install --group dev\n\nRun a single test suite:\n\n    python/run-tests --testnames pyspark.sql.tests.arrow.test_arrow\n\nRun a single test case:\n\n    python/run-tests --testnames \"pyspark.sql.tests.test_catalog CatalogTests.test_current_database\"\n\n## Investigating PR CI Failures\n\nEnumerate all failing check runs first, then drill into each by type. Do not assume a single failure: a PR can fail tests, linters, and the build at once, and these surface through different channels.\n\nStep 1 — Get the fork owner and the latest commit SHA of the PR:\n\n    gh api repos/apache/spark/pulls/<PR_NUMBER> --jq '{owner: .head.repo.owner.login, sha: .head.sha}'\n\nStep 2 — List every failing check run on the fork's commit. This is the complete failure set:\n\n    gh api repos/<OWNER>/spark/commits/<SHA>/check-runs --paginate \\\n      --jq '.check_runs[] | select(.conclusion == \"failure\") | {name, id: .id}'\n\nA passing (or absent) \"Report test results\" does NOT mean CI is green. That check aggregates only test-case failures; linter, license, dependency, MiMa, compile, and doc-build failures are separate check runs that produce no test annotations. Always work from the list in Step 2, not from any single check.\n\nStep 3 — Drill into each failure according to its kind:\n\n- **Test jobs** (e.g. \"Report test results\", \"Build modules: ...\"): fetch failure annotations. Each annotation contains the test class, test name, and failure message:\n\n      gh api repos/<OWNER>/spark/check-runs/<CHECK_RUN_ID>/annotations\n\n- **Non-test jobs** (e.g. \"Linters, licenses, and dependencies\", \"Build\"): find the failed step, then read only that job's log:\n\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID> \\\n        --jq '{name, steps: [.steps[] | select(.conclusion == \"failure\") | .name]}'\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID>/logs\n\nAvoid downloading the large per-shard *test* job logs — they are very large and slow; use the annotations for those. Lint, license, dependency, and build job logs are small and fine to read directly when a step fails.\n\n## Checking PR Merge Status\n\nSpark merges PRs with `dev/merge_spark_pr.py`, not the GitHub merge button, so a **merged PR shows up on GitHub as Closed, not Merged** — its `merged` / `mergedAt` are empty, and backports to maintenance branches are plain pushes. **Do not read a Closed PR as rejected**; most Closed PRs were in fact merged.\n\nTo check whether, and to which branches, a PR was merged, run `dev/pr_merge_status.py <pr-number>`:\n\n    $ dev/pr_merge_status.py 56356\n    PR #56356 (base master): [SPARK-57295][SQL] Make database location validation ...\n    merged: yes\n      master       9357bc9ae05\n      branch-4.x   72edddb358e\n\nIt lists `master` and the latest major's release branches the commit reached (e.g. `branch-4.x`, `branch-4.2`) — or reports `open` or `closed without merging`. It needs `gh` authenticated and a non-shallow checkout, and exits non-zero with a clear message if the PR is unknown (404) or the environment isn't ready (no `gh`/auth, no apache/spark remote). Older majors' branches are omitted; pass `--all-branches` to list every branch the commit reached.\n\n## Pull Request Workflow\n\nPR title format is `[SPARK-xxxx][COMPONENT] Title`. Draft, WIP, MINOR, and TRIVIAL PRs may omit the JIRA ID. The component tag comes from the `COMPONENTS` registry in `dev/merge_spark_pr.py`, the source of truth: each entry pairs a PR-title tag with its canonical JIRA component name and accepted aliases (e.g. `Documentation` → `[DOC]`, `Pandas API on Spark` → `[PS]`, `SQL` → `[SQL]`). Read that registry to pick the right tag.\n\nUse `[FOLLOWUP]` only for small PRs that directly modify or correct unreleased earlier PRs. For separately planned work, non-trivial changes, or work outside the earlier JIRA's scope, create a separate JIRA ticket for each PR and use the normal title format without `[FOLLOWUP]`.\n\nInfer the PR title from the changes. If no ticket ID is given and the PR is not draft, WIP, MINOR, or TRIVIAL, create one using `dev/create_spark_jira.py`, using the PR title (without the JIRA ID and component tag) as the ticket title.\n\n    python3 dev/create_spark_jira.py \"<title>\" -c <component> { -t <type> | -p <parent-jira-id> }\n\n- **Component** (`-c`): the exact JIRA component name (not the PR title shorthand), e.g. \"SQL\", \"Spark Core\", \"PySpark\", \"Connect\". Run `python3 dev/create_spark_jira.py --list-components` for the full list.\n- **Issue type** (`-t`): \"Bug\", \"Improvement\", \"New Feature\", \"Test\", \"Documentation\", or \"Dependency upgrade\".\n- **Parent** (`-p`): if the user mentions a parent JIRA ticket (e.g., \"this is a subtask of SPARK-12345\"), pass it instead of `-t`. The issue type is automatically \"Sub-task\".\n\nThe script sets the latest unreleased version as the default affected version.\n\nAfter creating a JIRA ticket, print a prominent notice so the user does not miss it:\n\n    ============================================================\n    JIRA ticket created: SPARK-XXXXX\n    https://issues.apache.org/jira/browse/SPARK-XXXXX\n\n    Title:              <title>\n    Component(s):       <component>\n    Issue type:         <type>\n    Affected version(s): <version>\n    Priority:           <priority>\n\n    Please review and adjust these fields if needed.\n    ============================================================\n\nBefore writing the PR description, read `.github/PULL_REQUEST_TEMPLATE` and fill in every section from that file.\n\nDO NOT push to the upstream repo. Always push to the personal fork. Open PRs against `master` on the upstream repo.\n\nDO NOT force push or use `--amend` on pushed commits unless the user explicitly asks. If the remote branch has new commits, fetch and rebase before pushing.\n\nAlways get user approval before external operations such as pushing commits, creating PRs, or posting comments. Use `gh pr create` to open PRs. If `gh` is not installed, generate the GitHub PR URL for the user and recommend installing the GitHub CLI.\n\n## Project Instructions Discovery\n\nWhen exploring or working in any directory, always check for nested `AGENTS.md` files in that\ndirectory and its ancestors. Read and follow every applicable file; instructions in a more specific\ndirectory take precedence for that directory's scope.\n\nA directory may carry these instructions as `CLAUDE.md` rather than `AGENTS.md`, so check **both** names — a directory with only a `CLAUDE.md` has project instructions you must read too. Where both files exist, one is typically the real file and the other is a symlink to it, although either name may be the real file. Reading either one is sufficient. When adding instructions to a directory that has neither file, create AGENTS.md as the real file and add a CLAUDE.md symlink beside it (ln -s AGENTS.md CLAUDE.md). This keeps a single source of truth accessible under both names.\n\n## Versioning and Branch Policy\n\nWhen a change needs a version — `@since` annotations, config `.version(\"...\")` (`SQLConf` / `*Conf`), new `MimaExcludes` sections, etc. — use the version of the branch it first ships in, with `-SNAPSHOT` stripped. Determine that branch:\n\n- **PR opened against a non-`master` base branch** (e.g. a maintenance line like `branch-4.2`): use that base branch's version -- when you're checked out on it, that's just the working tree's `pom.xml`. The helper below covers only the common `master`-base case.\n- **PR opened against `master`:** most PRs merge to **both** `master` and the latest `branch-<N>.x` (the branch for the next feature release, e.g. `branch-4.x`), so use the `branch-<N>.x` version. The exception is **master-only** changes — use `master`'s version — which are only:\n  - breaking / binary-incompatible changes that can't ship in a minor release;\n  - dependency upgrades that don't fix a critical issue worth backporting.\n\nDo **not** just read `master`'s version: a normally-backported PR ships first in `branch-<N>.x`, whose version is lower than `master`'s. If unsure whether a change is master-only, ask the user.\n\n`dev/next_version_candidates.py` (no arguments) prints both candidate versions, reading from the local `apache/spark` remote (the `upstream` configured during pre-flight). It reports the mechanical facts only -- choosing between them per the rules above is the judgement call (the numbers below are illustrative and advance over time):\n\n    $ dev/next_version_candidates.py\n    master       5.0.0\n    branch-4.x   4.3.0\n\n## Security\n\nSecurity model: [SECURITY.md](./SECURITY.md)\n\nAgents that scan this repository should consult `SECURITY.md` for the project's threat model, in-scope / out-of-scope declarations, and known non-findings before reporting issues.\n",".cursor/mcp.json":"# Model Context Protocol (MCP) Configuration\n\nPath: `.cursor/mcp.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.cursor/mcp.json)",".claude-plugin/marketplace.json":"# Claude Plugin Marketplace Catalog\n\nPath: `.claude-plugin/marketplace.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/marketplace.json)",".claude-plugin/plugin.json":"# claude-plugin Plugin Manifest\n\nPath: `.claude-plugin/plugin.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/plugin.json)",".cursorrules (pyspark-etl-best-practices-cursorrules-prompt-file.mdc)":"---\ndescription: \"Cursor rules for PySpark ETL development with code style, joins, window functions, map operations, and Iceberg patterns.\"\nglobs: **/*\nalwaysApply: false\n---\nYou are an expert in PySpark, Spark SQL, Apache Iceberg, and production data engineering. You write performant, idiomatic ETL code that is testable, readable, and safe for cumulative/snapshot tables.\n\nFollow these rules when generating or reviewing PySpark code.\n\n# PySpark ETL Best Practices\n\n## 1. Project Structure\n\n### ETL class scaffold\n\nCreate a base class that manages the SparkSession lifecycle. Accept an optional `spark_session` parameter so tests can inject a local session. Use an abstract method for the job logic.\n\n```python\nfrom abc import ABC, abstractmethod\nfrom pyspark.sql import SparkSession\n\nclass BaseETL(ABC):\n    def __init__(self, config, app_name=\"ETL Job\", spark_session=None):\n        self.spark = spark_session or SparkSession.builder.appName(app_name).getOrCreate()\n        self.config = config\n        self.logger = logging.getLogger(self.__class__.__name__)\n\n    @abstractmethod\n    def run_job(self): ...\n\n    def stop(self):\n        self.spark.stop()\n```\n\n### Config — use a factory function\n\nKeep the dataclass as pure data and put CLI parsing in a standalone factory function. This makes configs easy to construct in tests without touching `sys.argv`.\n\n```python\n@dataclass\nclass MyConfig:\n    read_date: int = 20200101\n\ndef create_config() -> MyConfig:\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--read_date\", type=int, default=20200101)\n    args = parser.parse_args()\n    return MyConfig(read_date=args.read_date)\n```\n\n### Pipeline composition with `.transform()`\n\nKeep `run_job` as orchestration. Each step is a named method.\n\n```python\nevents = self.read_source().transform(self.enrich).transform(self.merge_with_existing)\n```\n\n### Use a shared reader for partition-aware reads\n\nBuild a generic reader utility that handles partition mechanics (date filters, hour ranges, latest-partition lookups). Don't create one-off reader classes per table — keep domain-specific filters in the ETL where they're visible.\n\n```python\nclass PartitionedReader:\n    @staticmethod\n    def read_latest(spark, table_name, partition_col):\n        row = spark.read.table(table_name).agg(F.max(partition_col)).first()\n        if row is None or row[0] is None:\n            return spark.createDataFrame([], spark.read.table(table_name).schema)\n        return spark.read.table(table_name).filter(F.col(partition_col) == row[0])\n\n    @staticmethod\n    def read_by_date(spark, table_name, partition_col, date_value):\n        return spark.read.table(table_name).filter(F.col(partition_col) == date_value)\n\n# Reader handles partitioning\nevents = PartitionedReader.read_by_date(spark, \"catalog.my_table\", \"event_date\", 20260319)\n\n# Business filters stay in the ETL\nevents = events.filter(F.col(\"event_type\").isin(\"login\", \"purchase\"))\n```\n\n### Shared merge utilities\n\nFor simple outer-join-with-coalesce merges, build a reusable merge function that handles aliasing, join key coalescing, and per-column defaults. Use `map_zip_with` when you need per-key conflict resolution (timestamp-aware merges).\n\n## 2. Code Style\n\n### Use `F.col()` — always use the `F.` prefix\n\nImport functions as `import pyspark.sql.functions as F` and use `F.col()`, `F.when()`, `F.lit()`, etc. throughout. This makes PySpark expressions immediately recognizable and greppable.\n\nAvoid `df.colA` attribute access — it binds the column to a specific DataFrame variable, which breaks after joins or when the variable is reassigned. Use `F.col()` with `.alias()` on the DataFrame if disambiguation is needed.\n\n```python\n# BAD — binds column to a specific DataFrame variable, breaks after joins\ndf.select(F.lower(df1.colA), F.upper(df2.colB))\n\n# GOOD\ndf.select(F.lower(F.col('colA')), F.upper(F.col('colB')))\n```\n\n### Extract complex conditions into named variables\n\nLimit logic inside `.filter()` or `F.when()` to 3 expressions. Extract the rest.\n\n```python\n# BAD — redundant logic hidden in nested parentheses\nF.when((F.col('status') == 'Delivered') | (((F.datediff('date_a', 'date_b') < 0) & ...)), 'Active')\n\n# GOOD\nis_delivered = (F.col('status') == 'Delivered')\ndate_passed = (F.datediff(F.col('date_a'), F.col('date_b')) < 0)\nhas_registration = (F.col('registration').rlike('.+'))\nF.when(is_delivered | (date_passed & has_registration), 'Active')\n```\n\n### Prefer `select` over `withColumn` chains\n\n`select` specifies the output schema in one pass. `withColumn` chains create intermediate DataFrames and can degrade performance — each call triggers a new projection in the query plan.\n\n```python\n# BAD — 3 intermediate DataFrames\ndf = df.withColumn(\"a\", F.col(\"a\").cast(\"double\"))\ndf = df.withColumn(\"b\", F.upper(F.col(\"b\")))\ndf = df.withColumn(\"c\", F.lit(1))\n\n# GOOD — 1 DataFrame, explicit schema contract\ndf = df.select(\n    F.col(\"a\").cast(\"double\"),\n    F.upper(F.col(\"b\")).alias(\"b\"),\n    F.lit(1).alias(\"c\"),\n)\n```\n\n### Use `alias` over `withColumnRenamed`\n\n```python\n# BAD\ndf.select('key', 'comments').withColumnRenamed('comments', 'num_comments')\n\n# GOOD\ndf.select('key', F.col('comments').alias('num_comments'))\n```\n\n### Chaining limits\n\nMax 5 statements per chain. Separate by operation type (select/filter vs withColumn vs join).\n\n```python\n# BAD — mixed concerns in one chain\ndf = (df.select('a', 'b', 'key')\n    .filter(F.col('a') == 'x')\n    .withColumn('ratio', F.col('a') / F.col('b'))\n    .join(df2, 'key', how='inner')\n    .drop('b'))\n\n# GOOD — separated by concern\ndf = df.select('a', 'b', 'key').filter(F.col('a') == 'x')\ndf = df.withColumn('ratio', F.col('a') / F.col('b'))\ndf = df.join(df2, 'key', how='inner').drop('b')\n```\n\n## 3. Joins\n\n### Always specify `how=` explicitly\n\n```python\n# BAD\ndf.join(other, 'key')\n\n# GOOD\ndf.join(other, 'key', how='inner')\n```\n\n### Prefer left joins over right joins\n\nFlip the DataFrame order and use `left` instead of `right` for readability — the primary DataFrame stays on the left.\n\n```python\nflights = flights.join(aircraft, 'aircraft_id', how='left')\n```\n\n### Use `.alias()` for disambiguation after joins\n\n```python\n# BAD — renaming every column before join\nfor c in columns:\n    flights = flights.withColumnRenamed(c, 'flights_' + c)\n\n# GOOD — alias the whole DataFrame\nflights = flights.alias('f')\nparking = parking.alias('p')\nresult = flights.join(parking, 'code', how='left').select(\n    F.col('f.start_time').alias('flight_start'),\n    F.col('p.total_time').alias('parking_total'),\n)\n```\n\n### Broadcast small dimension tables\n\nWhen joining a large fact DataFrame with a small lookup/dimension table, wrap the small side in `F.broadcast()` to skip the shuffle on the small side.\n\nUse broadcast for tables that are small enough to fit in executor memory — typically dimension/lookup tables (category lookups, country codes, config mappings). Spark auto-broadcasts tables under 10MB by default (`spark.sql.autoBroadcastJoinThreshold`), but an explicit hint is useful when Spark can't infer the size (e.g., after filters or transformations).\n\nTo check if a table is broadcast-worthy during development:\n- **Spark UI**: after a run, check the SQL tab — scan sizes are shown per table\n- **Quick row count**: `spark.read.table(\"catalog.my_dim\").count()` (dev only, not in production code)\n- **Query plan**: `df.explain()` — Spark shows `BroadcastHashJoin` if it auto-broadcasts, `SortMergeJoin` if it doesn't\n\n```python\ndf.join(F.broadcast(category_dim), 'category_id', how='left')\n```\n\n### Never use `.dropDuplicates()` as a crutch\n\nIf duplicate rows appear, find the root cause. `.dropDuplicates()` masks the problem and adds shuffle overhead.\n\n## 4. Window Functions\n\nUse `from pyspark.sql import Window as W` alongside `import pyspark.sql.functions as F`.\n\n### Always specify an explicit frame\n\nWithout a frame, Spark picks one that changes depending on whether `orderBy` is present.\n\n```python\n# BAD — F.sum gives running sum with orderBy, total without. Surprising.\nw = W.partitionBy('key').orderBy('num')\n\n# GOOD — explicit about what you want\nw = W.partitionBy('key').orderBy('num').rowsBetween(W.unboundedPreceding, W.unboundedFollowing)\n```\n\n### `row_number` + filter vs `first` — know the difference\n\n- `row_number` + filter = **drop rows**, keep the best one\n- `first` over window = **overwrite a column value**, keep all rows\n\n### Use `ignorenulls=True` with `first` and `last`\n\nWithout it, a null in the first row gives null for the entire partition.\n\n```python\n# BAD — returns None if first row is null\nF.first('version').over(window)\n\n# GOOD\nF.first('version', ignorenulls=True).over(window)\n```\n\n### Avoid empty `partitionBy()`\n\nIt forces all data into one partition. Use `.agg()` instead for global aggregations.\n\n```python\n# BAD — single partition, kills performance\nw = W.partitionBy()\ndf.select(F.sum('num').over(w))\n\n# GOOD\ndf.agg(F.sum('num').alias('total'))\n```\n\n## 5. Map & Array Higher-Order Functions\n\n### Use `map_zip_with` when merging maps with complex logic\n\n`map_concat` is fine for simple merges with no key overlap. When you need custom logic per key (e.g., keep the newer timestamp, pick the higher value), use `map_zip_with` — it gives you a per-key merge function instead of blindly letting one side win.\n\n```python\n# BAD — no control over conflict resolution\nmap_concat(existing_map, new_map)\n\n# GOOD — keep the entry with the later timestamp\nmap_zip_with(new_map, existing_map,\n    lambda key, v1, v2: (\n        F.when(v1.isNull(), v2)\n        .when(v2.isNull(), v1)\n        .otherwise(F.when(v1.event_ts >= v2.event_ts, v1).otherwise(v2))\n    )\n)\n```\n\n### Use `transform` + `array_max` to extract from nested structs\n\n```python\n# Extract max event_ts from a map of structs\narray_max(transform(map_values(my_map), lambda x: x.event_ts))\n```\n\n### Avoid UDFs — use native Spark functions first\n\nUDFs break Catalyst optimization and add serialization overhead. Before writing one, check if a built-in Spark function or higher-order function can do the job.\n\n## 6. Cumulative / Snapshot Table Patterns\n\n### Merges must be idempotent\n\nRe-running with the same data should produce the same result, not create duplicates.\n\n### Merges must be order-independent\n\nBackfilling old data should not overwrite newer data. Use an explicit ordering criterion (e.g., event timestamp, version number, partition date) to resolve conflicts — don't rely on positional precedence like `coalesce` argument order.\n\n### Validate primary key uniqueness after writes\n\nAdd audit steps that validate primary key uniqueness and check for nulls in key columns.\n\n## 7. Data Quality & Performance\n\n### Use `F.lit(None)` for empty columns, never empty strings\n\n```python\n# BAD\ndf = df.withColumn('foo', F.lit(''))\ndf = df.withColumn('foo', F.lit('NA'))\n\n# GOOD\ndf = df.withColumn('foo', F.lit(None))\n```\n\n### Avoid `.otherwise()` as a general catch-all\n\nUnknown values silently collapse into the otherwise bucket, hiding data quality issues.\n\n```python\n# BAD — a new platform_type you didn't anticipate becomes \"Other\" silently\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n .otherwise('Other')\n\n# GOOD — unmapped values stay null, surfacing gaps in your logic\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n```\n\n### No `.show()`, `.collect()`, `.printSchema()` in production\n\nThese trigger full materialization or add unnecessary driver overhead. Use them only for local debugging, never in deployed ETL code. `.count()` is acceptable when used intentionally (e.g., logging row counts for monitoring, forcing materialization before a DAG fork).\n\n### Use `persist()` intentionally\n\nOnly persist a DataFrame when it's referenced in multiple subsequent actions — otherwise the write action will materialize it for you. `.persist()` + `.count()` is a common pattern to force materialization and log row counts for debugging; use it when needed but be aware it adds a full scan.\n\nChoose the storage level based on your use case:\n- `MEMORY_AND_DISK` — safe default, spills to disk if memory is tight\n- `MEMORY_ONLY` — faster but risks recomputation if evicted\n- `DISK_ONLY` — for very large DataFrames that don't fit in memory\n\n## 8. Iceberg Write Patterns\n\n### Use `.byName()` for schema evolution safety\n\nColumn ordering doesn't matter — Spark matches by name, not position.\n\n```python\ndf.write.byName().mode(\"overwrite\").insertInto(\"catalog.my_table\")\n```\n\n### Use `__partitions` metadata table for latest partition reads\n\nIceberg exposes a `__partitions` metadata table. Use it to find the latest snapshot instead of scanning the full table.\n\n```python\npartition_df = spark.read.table(\"catalog.my_table__partitions\").select(\n    \"partition.partition_date\", \"partition.partition_hour\"\n)\nmax_partition = partition_df.orderBy(\n    F.col(\"partition_date\").desc(), F.col(\"partition_hour\").desc()\n).first()\nif max_partition is None:\n    raise ValueError(\"No partitions found in catalog.my_table\")\nlatest_date = max_partition[\"partition_date\"]\n```\n\n### Understand `write.distribution-mode`\n\n- `\"none\"` — no re-shuffle before writing. Fastest, but output file sizes depend on upstream partitioning.\n- `\"hash\"` — redistributes data by partition key. Produces evenly sized files but adds a shuffle.\n- `\"range\"` — sorts data by partition key before writing. Good for ordered scan performance but most expensive.\n"},"files":{"AGENTS.md":"# Apache Spark\n\n## Pre-flight Checks\n\nBefore the first code edit or running test in a session, ensure a clean working environment. DO NOT skip these checks:\n\n1. Run `git remote -v` to identify the personal fork and upstream (`apache/spark`). If unclear, ask the user to configure their remotes following the standard convention (`origin` for the fork, `upstream` for `apache/spark`).\n2. If the latest commit on `<upstream>/master` is more than a day old (check with `git log -1 --format=\"%ci\" <upstream>/master`), run `git fetch <upstream> master`.\n3. If there are uncommitted changes (check with `git status`), ask the user to stash them before proceeding.\n4. Switch to the appropriate branch:\n   - **Existing PR**: resolve the PR branch name via `gh api repos/apache/spark/pulls/<number> --jq '.head.ref'`, then look for a local branch matching that name. If found, switch to it and inform the user. If not found, ask whether to fetch it or if there is a local branch under a different name.\n   - **New edits**: ask the user to choose: create a new git worktree from `<upstream>/master` with `--no-track` and work from there (recommended), or create and switch to a new branch from `<upstream>/master` with `--no-track`.\n   - **Running tests**: use `<upstream>/master`.\n\n## Development Notes\n\nSQL golden file tests are managed by `SQLQueryTestSuite` and its variants. Read the class documentation before running or updating these tests. DO NOT edit the generated golden files (`.sql.out`) directly. Always regenerate them when needed, and carefully review the diff to make sure it's expected.\n\nSpark Connect protocol is defined in proto files under `sql/connect/common/src/main/protobuf/`. Read the README there before modifying proto definitions.\n\nWhen adding a member to an existing class or object, follow the sectioning the file already uses and put the new member with the code it belongs with. The common failure mode is dropping it wherever it is first used without checking how the file is organized, splitting a section of unrelated code in the process. Beyond grouping related code there is no prescribed order -- the Databricks Scala guide, which Spark follows, asks only that a long class group its members into logical sections with comment headers. Do not reorganize existing members unless the change requires it.\n\nAvoid introducing non-ASCII characters in code or comments. String literals may contain non-ASCII when the content requires it (error messages, test data, etc.). Identifiers are ASCII by convention. The common failure mode is typographic characters (em-dash, smart quotes, ellipsis, non-breaking space) sneaking into comments; scalastyle flags some of these. Spot-check before committing: `grep -rn -P \"[^\\x00-\\x7F]\" <files>`.\n\nKeep source lines within 100 characters — the linters enforce this for Scala, Java, and Python, and LLMs commonly overrun it in comments and long expressions. A quick scan of just the changed files catches most cases in seconds, far cheaper than a CI round trip:\n\n    { git diff --name-only --diff-filter=ACM HEAD; git ls-files --others --exclude-standard; } \\\n      | grep -E '\\.(scala|java|py)$' | sort -u \\\n      | xargs -r awk 'length>100 && $0 !~ /^[[:space:]]*(import|package) / && $0 !~ /https?:\\/\\// \\\n          {print FILENAME\":\"FNR\": \"length\" chars\"}'\n\nThis is only a hint: it approximates the linters' exemptions (imports, URLs) rather than matching them exactly, so it can over- or under-report. The linters remain the source of truth.\n\n## Scala Test Base Classes\n\nWhen writing a new Scala test suite, pick the lowest base class that provides what the test actually needs. Spark uses the `AnyFunSuite` ScalaTest style throughout, so the bases below are the chain to choose from. Each adds capability on top of the previous:\n\n    SparkFunSuite                                                           (core)\n      <- PlanTest                                                           (sql/catalyst)\n        <- QueryTest                                                        (sql/core)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain JVM/Scala — no Spark SQL | `SparkFunSuite` | `core` utilities, RDD, network, util classes, etc. Adds per-test timeout, `testRetry`, `gridTest`, thread audit, fixed timezone/locale, `withTempDir`, `withLogAppender`, `checkError`. |\n| Catalyst plan tests — no `SparkSession` | `PlanTest` | Adds `comparePlans`, `normalizePlan`, `normalizeExprIds`. For analyzer / optimizer / planner rule tests. |\n| SQL/DataFrame tests — needs a `SparkSession` | `QueryTest` | Adds `checkAnswer`, codegen-on/off helpers. `spark: SparkSession` is abstract and must be supplied by a session-providing trait (see below). |\n\n### Providing a `SparkSession` for `QueryTest`\n\n`QueryTest` declares `spark: SparkSession` abstractly via `SparkSessionProvider`, so it cannot be instantiated on its own. A concrete suite mixes in one of the session-providing traits below:\n\n    QueryTest                                                               (abstract `spark`)\n      + SharedSparkSession (sql/core)        -> classic in-process `TestSparkSession`\n      + TestHiveSingleton  (sql/hive)        -> Hive-backed `TestHive` session\n\n| Session provider | Module / location | Typical usage |\n|---|---|---|\n| `SharedSparkSession` | `sql/core` | Already extends `QueryTest` for historical reasons, but still mix in `QueryTest` explicitly, e.g. `class X extends QueryTest with SharedSparkSession`. Default for tests under `sql/core`. |\n| `TestHiveSingleton` | `sql/hive` | Mixed in alongside `QueryTest`, e.g. `class X extends QueryTest with TestHiveSingleton`. Used by tests under `sql/hive`. |\n\n## Python Test Base Classes\n\nPySpark tests use the stdlib `unittest` framework: every suite subclasses `unittest.TestCase` (run via `python/run-tests`, see below). As with Scala, pick the lowest base that provides what the test actually needs. The bases live under `python/pyspark/testing/` and each adds capability on top of the previous:\n\n    unittest.TestCase                                     (stdlib)\n      <- PySparkBaseTestCase                              (pyspark.testing.utils)\n        <- ReusedPySparkTestCase                          (pyspark.testing.utils)\n          <- ReusedSQLTestCase                            (pyspark.testing.sqlutils)\n            <- PandasOnSparkTestCase                      (pyspark.testing.pandasutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain Python — no Spark | `unittest.TestCase` | Use the stdlib base directly. `PySparkBaseTestCase` is the same thing plus a SIGTERM fault-handler dump enabled when `PYSPARK_TEST_TIMEOUT` is set; subclass it only when you want that. |\n| RDD / `SparkContext` — no `SparkSession` | `PySparkTestCase` (fresh) or `ReusedPySparkTestCase` (shared) | Both create a `SparkContext(\"local[4]\")`. `PySparkTestCase` makes a new one per test (isolation); `ReusedPySparkTestCase` shares one per class (faster, the usual choice) and adds `quiet()` and an overridable `conf()` / `master()`. |\n| SQL / DataFrame — needs a `SparkSession` | `ReusedSQLTestCase` | The workhorse for classic tests under `python/pyspark/sql`. Adds a shared `cls.spark`, sample `cls.df` / `cls.testData`, and mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. Classic (non-Connect) mode. |\n| pandas API on Spark | `PandasOnSparkTestCase` | Extends `ReusedSQLTestCase` with Arrow enabled and pandas-on-Spark assertions (`PandasOnSparkTestUtils`); `ComparisonTestBase` builds on it. |\n\n### Spark Connect test bases\n\nSpark Connect suites live in `pyspark.testing.connectutils` and are auto-skipped (via `should_test_connect`) when Connect dependencies are missing:\n\n    PySparkBaseTestCase                                   (pyspark.testing.utils)\n      <- PlanOnlyTestFixture                              (pyspark.testing.connectutils)\n      <- ReusedConnectTestCase                            (pyspark.testing.connectutils)\n           <- ReusedMixedTestCase                         (pyspark.testing.connectutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plan / proto construction — no server | `PlanOnlyTestFixture` | Uses a `MockRemoteSession`; builds and inspects plans without a running Connect server. For proto / plan-shape assertions. |\n| Connect DataFrame — real session | `ReusedConnectTestCase` | The Connect analog of `ReusedSQLTestCase`; starts a session via `.remote(...)` (honoring `SPARK_CONNECT_TESTING_REMOTE`, default `local[4]`). Mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. |\n| Classic + Connect side by side | `ReusedMixedTestCase` | Extends `ReusedConnectTestCase`. For directly comparing classic vs Connect: it exposes a classic `self.spark` and a Connect `self.connect` so a test can run the same operation on each and assert they agree (`compare_by_show`, `both_conf`). Requires JVM access. |\n\n### Mixins and helpers\n\nThese are combined with a base above rather than used on their own:\n\n- `SQLTestUtils` — context managers `sql_conf`, `table`, `temp_view`, `view`, `database`, `function`, `temp_func`, `temp_env`; assumes `self.spark`. Already mixed into `ReusedSQLTestCase` and `ReusedConnectTestCase`.\n- `PySparkErrorTestUtils` — `check_error(...)` to assert on a `PySparkException`'s error class and message parameters. The Python counterpart of Scala's `checkError`.\n- `assertDataFrameEqual` / `assertSchemaEqual` (public API, from `pyspark.testing`) — standalone assertion functions that work in any test, no particular base class required.\n- Domain bases outside the main ladder: `SparkSessionTestCase` (`pyspark.testing.mlutils`, for `ml`), `MLlibTestCase` (`pyspark.testing.mllibutils`), and `PySparkStreamingTestCase` (`pyspark.testing.streamingutils`, DStreams).\n\n## Build and Test\n\nBuild and tests can take a long time, so prefer the narrowest relevant build or test command.\n\nFor build and test setup, including how to run tests and troubleshoot common\nlocal failures, see `docs/building-spark.md`.\n\nPrefer SBT over Maven for faster incremental compilation. Module names are defined in `project/SparkBuild.scala`.\n\nCompile a single module:\n\n    build/sbt <module>/compile\n\nCompile test code for a single module:\n\n    build/sbt <module>/Test/compile\n\nRun test suites by wildcard or full class name:\n\n    build/sbt '<module>/testOnly *MySuite'\n    build/sbt '<module>/testOnly org.apache.spark.sql.MySuite'\n\nRun test cases matching a substring:\n\n    build/sbt '<module>/testOnly *MySuite -- -z \"test name\"'\n\nRun test cases in an optional module:\n\n    build/sbt -P<maven-profiles> '<module>/testOnly *MySuite'\n\nFor faster iteration, keep SBT open in interactive mode:\n\n    build/sbt\n    > project <module>\n    > testOnly *MySuite\n\n### PySpark Tests\n\nPySpark tests require building Spark with Hive support first:\n\n    build/sbt -Phive package\n\nActivate the virtual environment specified by the user, or default to `.venv`:\n\n    source <venv>/bin/activate\n\nIf the default venv does not exist, create it:\n\n    python3 -m venv .venv\n    source .venv/bin/activate\n    pip install --upgrade pip\n    pip install --group dev\n\nRun a single test suite:\n\n    python/run-tests --testnames pyspark.sql.tests.arrow.test_arrow\n\nRun a single test case:\n\n    python/run-tests --testnames \"pyspark.sql.tests.test_catalog CatalogTests.test_current_database\"\n\n## Investigating PR CI Failures\n\nEnumerate all failing check runs first, then drill into each by type. Do not assume a single failure: a PR can fail tests, linters, and the build at once, and these surface through different channels.\n\nStep 1 — Get the fork owner and the latest commit SHA of the PR:\n\n    gh api repos/apache/spark/pulls/<PR_NUMBER> --jq '{owner: .head.repo.owner.login, sha: .head.sha}'\n\nStep 2 — List every failing check run on the fork's commit. This is the complete failure set:\n\n    gh api repos/<OWNER>/spark/commits/<SHA>/check-runs --paginate \\\n      --jq '.check_runs[] | select(.conclusion == \"failure\") | {name, id: .id}'\n\nA passing (or absent) \"Report test results\" does NOT mean CI is green. That check aggregates only test-case failures; linter, license, dependency, MiMa, compile, and doc-build failures are separate check runs that produce no test annotations. Always work from the list in Step 2, not from any single check.\n\nStep 3 — Drill into each failure according to its kind:\n\n- **Test jobs** (e.g. \"Report test results\", \"Build modules: ...\"): fetch failure annotations. Each annotation contains the test class, test name, and failure message:\n\n      gh api repos/<OWNER>/spark/check-runs/<CHECK_RUN_ID>/annotations\n\n- **Non-test jobs** (e.g. \"Linters, licenses, and dependencies\", \"Build\"): find the failed step, then read only that job's log:\n\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID> \\\n        --jq '{name, steps: [.steps[] | select(.conclusion == \"failure\") | .name]}'\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID>/logs\n\nAvoid downloading the large per-shard *test* job logs — they are very large and slow; use the annotations for those. Lint, license, dependency, and build job logs are small and fine to read directly when a step fails.\n\n## Checking PR Merge Status\n\nSpark merges PRs with `dev/merge_spark_pr.py`, not the GitHub merge button, so a **merged PR shows up on GitHub as Closed, not Merged** — its `merged` / `mergedAt` are empty, and backports to maintenance branches are plain pushes. **Do not read a Closed PR as rejected**; most Closed PRs were in fact merged.\n\nTo check whether, and to which branches, a PR was merged, run `dev/pr_merge_status.py <pr-number>`:\n\n    $ dev/pr_merge_status.py 56356\n    PR #56356 (base master): [SPARK-57295][SQL] Make database location validation ...\n    merged: yes\n      master       9357bc9ae05\n      branch-4.x   72edddb358e\n\nIt lists `master` and the latest major's release branches the commit reached (e.g. `branch-4.x`, `branch-4.2`) — or reports `open` or `closed without merging`. It needs `gh` authenticated and a non-shallow checkout, and exits non-zero with a clear message if the PR is unknown (404) or the environment isn't ready (no `gh`/auth, no apache/spark remote). Older majors' branches are omitted; pass `--all-branches` to list every branch the commit reached.\n\n## Pull Request Workflow\n\nPR title format is `[SPARK-xxxx][COMPONENT] Title`. Draft, WIP, MINOR, and TRIVIAL PRs may omit the JIRA ID. The component tag comes from the `COMPONENTS` registry in `dev/merge_spark_pr.py`, the source of truth: each entry pairs a PR-title tag with its canonical JIRA component name and accepted aliases (e.g. `Documentation` → `[DOC]`, `Pandas API on Spark` → `[PS]`, `SQL` → `[SQL]`). Read that registry to pick the right tag.\n\nUse `[FOLLOWUP]` only for small PRs that directly modify or correct unreleased earlier PRs. For separately planned work, non-trivial changes, or work outside the earlier JIRA's scope, create a separate JIRA ticket for each PR and use the normal title format without `[FOLLOWUP]`.\n\nInfer the PR title from the changes. If no ticket ID is given and the PR is not draft, WIP, MINOR, or TRIVIAL, create one using `dev/create_spark_jira.py`, using the PR title (without the JIRA ID and component tag) as the ticket title.\n\n    python3 dev/create_spark_jira.py \"<title>\" -c <component> { -t <type> | -p <parent-jira-id> }\n\n- **Component** (`-c`): the exact JIRA component name (not the PR title shorthand), e.g. \"SQL\", \"Spark Core\", \"PySpark\", \"Connect\". Run `python3 dev/create_spark_jira.py --list-components` for the full list.\n- **Issue type** (`-t`): \"Bug\", \"Improvement\", \"New Feature\", \"Test\", \"Documentation\", or \"Dependency upgrade\".\n- **Parent** (`-p`): if the user mentions a parent JIRA ticket (e.g., \"this is a subtask of SPARK-12345\"), pass it instead of `-t`. The issue type is automatically \"Sub-task\".\n\nThe script sets the latest unreleased version as the default affected version.\n\nAfter creating a JIRA ticket, print a prominent notice so the user does not miss it:\n\n    ============================================================\n    JIRA ticket created: SPARK-XXXXX\n    https://issues.apache.org/jira/browse/SPARK-XXXXX\n\n    Title:              <title>\n    Component(s):       <component>\n    Issue type:         <type>\n    Affected version(s): <version>\n    Priority:           <priority>\n\n    Please review and adjust these fields if needed.\n    ============================================================\n\nBefore writing the PR description, read `.github/PULL_REQUEST_TEMPLATE` and fill in every section from that file.\n\nDO NOT push to the upstream repo. Always push to the personal fork. Open PRs against `master` on the upstream repo.\n\nDO NOT force push or use `--amend` on pushed commits unless the user explicitly asks. If the remote branch has new commits, fetch and rebase before pushing.\n\nAlways get user approval before external operations such as pushing commits, creating PRs, or posting comments. Use `gh pr create` to open PRs. If `gh` is not installed, generate the GitHub PR URL for the user and recommend installing the GitHub CLI.\n\n## Project Instructions Discovery\n\nWhen exploring or working in any directory, always check for nested `AGENTS.md` files in that\ndirectory and its ancestors. Read and follow every applicable file; instructions in a more specific\ndirectory take precedence for that directory's scope.\n\nA directory may carry these instructions as `CLAUDE.md` rather than `AGENTS.md`, so check **both** names — a directory with only a `CLAUDE.md` has project instructions you must read too. Where both files exist, one is typically the real file and the other is a symlink to it, although either name may be the real file. Reading either one is sufficient. When adding instructions to a directory that has neither file, create AGENTS.md as the real file and add a CLAUDE.md symlink beside it (ln -s AGENTS.md CLAUDE.md). This keeps a single source of truth accessible under both names.\n\n## Versioning and Branch Policy\n\nWhen a change needs a version — `@since` annotations, config `.version(\"...\")` (`SQLConf` / `*Conf`), new `MimaExcludes` sections, etc. — use the version of the branch it first ships in, with `-SNAPSHOT` stripped. Determine that branch:\n\n- **PR opened against a non-`master` base branch** (e.g. a maintenance line like `branch-4.2`): use that base branch's version -- when you're checked out on it, that's just the working tree's `pom.xml`. The helper below covers only the common `master`-base case.\n- **PR opened against `master`:** most PRs merge to **both** `master` and the latest `branch-<N>.x` (the branch for the next feature release, e.g. `branch-4.x`), so use the `branch-<N>.x` version. The exception is **master-only** changes — use `master`'s version — which are only:\n  - breaking / binary-incompatible changes that can't ship in a minor release;\n  - dependency upgrades that don't fix a critical issue worth backporting.\n\nDo **not** just read `master`'s version: a normally-backported PR ships first in `branch-<N>.x`, whose version is lower than `master`'s. If unsure whether a change is master-only, ask the user.\n\n`dev/next_version_candidates.py` (no arguments) prints both candidate versions, reading from the local `apache/spark` remote (the `upstream` configured during pre-flight). It reports the mechanical facts only -- choosing between them per the rules above is the judgement call (the numbers below are illustrative and advance over time):\n\n    $ dev/next_version_candidates.py\n    master       5.0.0\n    branch-4.x   4.3.0\n\n## Security\n\nSecurity model: [SECURITY.md](./SECURITY.md)\n\nAgents that scan this repository should consult `SECURITY.md` for the project's threat model, in-scope / out-of-scope declarations, and known non-findings before reporting issues.\n","CLAUDE.md":"# Apache Spark\n\n## Pre-flight Checks\n\nBefore the first code edit or running test in a session, ensure a clean working environment. DO NOT skip these checks:\n\n1. Run `git remote -v` to identify the personal fork and upstream (`apache/spark`). If unclear, ask the user to configure their remotes following the standard convention (`origin` for the fork, `upstream` for `apache/spark`).\n2. If the latest commit on `<upstream>/master` is more than a day old (check with `git log -1 --format=\"%ci\" <upstream>/master`), run `git fetch <upstream> master`.\n3. If there are uncommitted changes (check with `git status`), ask the user to stash them before proceeding.\n4. Switch to the appropriate branch:\n   - **Existing PR**: resolve the PR branch name via `gh api repos/apache/spark/pulls/<number> --jq '.head.ref'`, then look for a local branch matching that name. If found, switch to it and inform the user. If not found, ask whether to fetch it or if there is a local branch under a different name.\n   - **New edits**: ask the user to choose: create a new git worktree from `<upstream>/master` with `--no-track` and work from there (recommended), or create and switch to a new branch from `<upstream>/master` with `--no-track`.\n   - **Running tests**: use `<upstream>/master`.\n\n## Development Notes\n\nSQL golden file tests are managed by `SQLQueryTestSuite` and its variants. Read the class documentation before running or updating these tests. DO NOT edit the generated golden files (`.sql.out`) directly. Always regenerate them when needed, and carefully review the diff to make sure it's expected.\n\nSpark Connect protocol is defined in proto files under `sql/connect/common/src/main/protobuf/`. Read the README there before modifying proto definitions.\n\nWhen adding a member to an existing class or object, follow the sectioning the file already uses and put the new member with the code it belongs with. The common failure mode is dropping it wherever it is first used without checking how the file is organized, splitting a section of unrelated code in the process. Beyond grouping related code there is no prescribed order -- the Databricks Scala guide, which Spark follows, asks only that a long class group its members into logical sections with comment headers. Do not reorganize existing members unless the change requires it.\n\nAvoid introducing non-ASCII characters in code or comments. String literals may contain non-ASCII when the content requires it (error messages, test data, etc.). Identifiers are ASCII by convention. The common failure mode is typographic characters (em-dash, smart quotes, ellipsis, non-breaking space) sneaking into comments; scalastyle flags some of these. Spot-check before committing: `grep -rn -P \"[^\\x00-\\x7F]\" <files>`.\n\nKeep source lines within 100 characters — the linters enforce this for Scala, Java, and Python, and LLMs commonly overrun it in comments and long expressions. A quick scan of just the changed files catches most cases in seconds, far cheaper than a CI round trip:\n\n    { git diff --name-only --diff-filter=ACM HEAD; git ls-files --others --exclude-standard; } \\\n      | grep -E '\\.(scala|java|py)$' | sort -u \\\n      | xargs -r awk 'length>100 && $0 !~ /^[[:space:]]*(import|package) / && $0 !~ /https?:\\/\\// \\\n          {print FILENAME\":\"FNR\": \"length\" chars\"}'\n\nThis is only a hint: it approximates the linters' exemptions (imports, URLs) rather than matching them exactly, so it can over- or under-report. The linters remain the source of truth.\n\n## Scala Test Base Classes\n\nWhen writing a new Scala test suite, pick the lowest base class that provides what the test actually needs. Spark uses the `AnyFunSuite` ScalaTest style throughout, so the bases below are the chain to choose from. Each adds capability on top of the previous:\n\n    SparkFunSuite                                                           (core)\n      <- PlanTest                                                           (sql/catalyst)\n        <- QueryTest                                                        (sql/core)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain JVM/Scala — no Spark SQL | `SparkFunSuite` | `core` utilities, RDD, network, util classes, etc. Adds per-test timeout, `testRetry`, `gridTest`, thread audit, fixed timezone/locale, `withTempDir`, `withLogAppender`, `checkError`. |\n| Catalyst plan tests — no `SparkSession` | `PlanTest` | Adds `comparePlans`, `normalizePlan`, `normalizeExprIds`. For analyzer / optimizer / planner rule tests. |\n| SQL/DataFrame tests — needs a `SparkSession` | `QueryTest` | Adds `checkAnswer`, codegen-on/off helpers. `spark: SparkSession` is abstract and must be supplied by a session-providing trait (see below). |\n\n### Providing a `SparkSession` for `QueryTest`\n\n`QueryTest` declares `spark: SparkSession` abstractly via `SparkSessionProvider`, so it cannot be instantiated on its own. A concrete suite mixes in one of the session-providing traits below:\n\n    QueryTest                                                               (abstract `spark`)\n      + SharedSparkSession (sql/core)        -> classic in-process `TestSparkSession`\n      + TestHiveSingleton  (sql/hive)        -> Hive-backed `TestHive` session\n\n| Session provider | Module / location | Typical usage |\n|---|---|---|\n| `SharedSparkSession` | `sql/core` | Already extends `QueryTest` for historical reasons, but still mix in `QueryTest` explicitly, e.g. `class X extends QueryTest with SharedSparkSession`. Default for tests under `sql/core`. |\n| `TestHiveSingleton` | `sql/hive` | Mixed in alongside `QueryTest`, e.g. `class X extends QueryTest with TestHiveSingleton`. Used by tests under `sql/hive`. |\n\n## Python Test Base Classes\n\nPySpark tests use the stdlib `unittest` framework: every suite subclasses `unittest.TestCase` (run via `python/run-tests`, see below). As with Scala, pick the lowest base that provides what the test actually needs. The bases live under `python/pyspark/testing/` and each adds capability on top of the previous:\n\n    unittest.TestCase                                     (stdlib)\n      <- PySparkBaseTestCase                              (pyspark.testing.utils)\n        <- ReusedPySparkTestCase                          (pyspark.testing.utils)\n          <- ReusedSQLTestCase                            (pyspark.testing.sqlutils)\n            <- PandasOnSparkTestCase                      (pyspark.testing.pandasutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain Python — no Spark | `unittest.TestCase` | Use the stdlib base directly. `PySparkBaseTestCase` is the same thing plus a SIGTERM fault-handler dump enabled when `PYSPARK_TEST_TIMEOUT` is set; subclass it only when you want that. |\n| RDD / `SparkContext` — no `SparkSession` | `PySparkTestCase` (fresh) or `ReusedPySparkTestCase` (shared) | Both create a `SparkContext(\"local[4]\")`. `PySparkTestCase` makes a new one per test (isolation); `ReusedPySparkTestCase` shares one per class (faster, the usual choice) and adds `quiet()` and an overridable `conf()` / `master()`. |\n| SQL / DataFrame — needs a `SparkSession` | `ReusedSQLTestCase` | The workhorse for classic tests under `python/pyspark/sql`. Adds a shared `cls.spark`, sample `cls.df` / `cls.testData`, and mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. Classic (non-Connect) mode. |\n| pandas API on Spark | `PandasOnSparkTestCase` | Extends `ReusedSQLTestCase` with Arrow enabled and pandas-on-Spark assertions (`PandasOnSparkTestUtils`); `ComparisonTestBase` builds on it. |\n\n### Spark Connect test bases\n\nSpark Connect suites live in `pyspark.testing.connectutils` and are auto-skipped (via `should_test_connect`) when Connect dependencies are missing:\n\n    PySparkBaseTestCase                                   (pyspark.testing.utils)\n      <- PlanOnlyTestFixture                              (pyspark.testing.connectutils)\n      <- ReusedConnectTestCase                            (pyspark.testing.connectutils)\n           <- ReusedMixedTestCase                         (pyspark.testing.connectutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plan / proto construction — no server | `PlanOnlyTestFixture` | Uses a `MockRemoteSession`; builds and inspects plans without a running Connect server. For proto / plan-shape assertions. |\n| Connect DataFrame — real session | `ReusedConnectTestCase` | The Connect analog of `ReusedSQLTestCase`; starts a session via `.remote(...)` (honoring `SPARK_CONNECT_TESTING_REMOTE`, default `local[4]`). Mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. |\n| Classic + Connect side by side | `ReusedMixedTestCase` | Extends `ReusedConnectTestCase`. For directly comparing classic vs Connect: it exposes a classic `self.spark` and a Connect `self.connect` so a test can run the same operation on each and assert they agree (`compare_by_show`, `both_conf`). Requires JVM access. |\n\n### Mixins and helpers\n\nThese are combined with a base above rather than used on their own:\n\n- `SQLTestUtils` — context managers `sql_conf`, `table`, `temp_view`, `view`, `database`, `function`, `temp_func`, `temp_env`; assumes `self.spark`. Already mixed into `ReusedSQLTestCase` and `ReusedConnectTestCase`.\n- `PySparkErrorTestUtils` — `check_error(...)` to assert on a `PySparkException`'s error class and message parameters. The Python counterpart of Scala's `checkError`.\n- `assertDataFrameEqual` / `assertSchemaEqual` (public API, from `pyspark.testing`) — standalone assertion functions that work in any test, no particular base class required.\n- Domain bases outside the main ladder: `SparkSessionTestCase` (`pyspark.testing.mlutils`, for `ml`), `MLlibTestCase` (`pyspark.testing.mllibutils`), and `PySparkStreamingTestCase` (`pyspark.testing.streamingutils`, DStreams).\n\n## Build and Test\n\nBuild and tests can take a long time, so prefer the narrowest relevant build or test command.\n\nFor build and test setup, including how to run tests and troubleshoot common\nlocal failures, see `docs/building-spark.md`.\n\nPrefer SBT over Maven for faster incremental compilation. Module names are defined in `project/SparkBuild.scala`.\n\nCompile a single module:\n\n    build/sbt <module>/compile\n\nCompile test code for a single module:\n\n    build/sbt <module>/Test/compile\n\nRun test suites by wildcard or full class name:\n\n    build/sbt '<module>/testOnly *MySuite'\n    build/sbt '<module>/testOnly org.apache.spark.sql.MySuite'\n\nRun test cases matching a substring:\n\n    build/sbt '<module>/testOnly *MySuite -- -z \"test name\"'\n\nRun test cases in an optional module:\n\n    build/sbt -P<maven-profiles> '<module>/testOnly *MySuite'\n\nFor faster iteration, keep SBT open in interactive mode:\n\n    build/sbt\n    > project <module>\n    > testOnly *MySuite\n\n### PySpark Tests\n\nPySpark tests require building Spark with Hive support first:\n\n    build/sbt -Phive package\n\nActivate the virtual environment specified by the user, or default to `.venv`:\n\n    source <venv>/bin/activate\n\nIf the default venv does not exist, create it:\n\n    python3 -m venv .venv\n    source .venv/bin/activate\n    pip install --upgrade pip\n    pip install --group dev\n\nRun a single test suite:\n\n    python/run-tests --testnames pyspark.sql.tests.arrow.test_arrow\n\nRun a single test case:\n\n    python/run-tests --testnames \"pyspark.sql.tests.test_catalog CatalogTests.test_current_database\"\n\n## Investigating PR CI Failures\n\nEnumerate all failing check runs first, then drill into each by type. Do not assume a single failure: a PR can fail tests, linters, and the build at once, and these surface through different channels.\n\nStep 1 — Get the fork owner and the latest commit SHA of the PR:\n\n    gh api repos/apache/spark/pulls/<PR_NUMBER> --jq '{owner: .head.repo.owner.login, sha: .head.sha}'\n\nStep 2 — List every failing check run on the fork's commit. This is the complete failure set:\n\n    gh api repos/<OWNER>/spark/commits/<SHA>/check-runs --paginate \\\n      --jq '.check_runs[] | select(.conclusion == \"failure\") | {name, id: .id}'\n\nA passing (or absent) \"Report test results\" does NOT mean CI is green. That check aggregates only test-case failures; linter, license, dependency, MiMa, compile, and doc-build failures are separate check runs that produce no test annotations. Always work from the list in Step 2, not from any single check.\n\nStep 3 — Drill into each failure according to its kind:\n\n- **Test jobs** (e.g. \"Report test results\", \"Build modules: ...\"): fetch failure annotations. Each annotation contains the test class, test name, and failure message:\n\n      gh api repos/<OWNER>/spark/check-runs/<CHECK_RUN_ID>/annotations\n\n- **Non-test jobs** (e.g. \"Linters, licenses, and dependencies\", \"Build\"): find the failed step, then read only that job's log:\n\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID> \\\n        --jq '{name, steps: [.steps[] | select(.conclusion == \"failure\") | .name]}'\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID>/logs\n\nAvoid downloading the large per-shard *test* job logs — they are very large and slow; use the annotations for those. Lint, license, dependency, and build job logs are small and fine to read directly when a step fails.\n\n## Checking PR Merge Status\n\nSpark merges PRs with `dev/merge_spark_pr.py`, not the GitHub merge button, so a **merged PR shows up on GitHub as Closed, not Merged** — its `merged` / `mergedAt` are empty, and backports to maintenance branches are plain pushes. **Do not read a Closed PR as rejected**; most Closed PRs were in fact merged.\n\nTo check whether, and to which branches, a PR was merged, run `dev/pr_merge_status.py <pr-number>`:\n\n    $ dev/pr_merge_status.py 56356\n    PR #56356 (base master): [SPARK-57295][SQL] Make database location validation ...\n    merged: yes\n      master       9357bc9ae05\n      branch-4.x   72edddb358e\n\nIt lists `master` and the latest major's release branches the commit reached (e.g. `branch-4.x`, `branch-4.2`) — or reports `open` or `closed without merging`. It needs `gh` authenticated and a non-shallow checkout, and exits non-zero with a clear message if the PR is unknown (404) or the environment isn't ready (no `gh`/auth, no apache/spark remote). Older majors' branches are omitted; pass `--all-branches` to list every branch the commit reached.\n\n## Pull Request Workflow\n\nPR title format is `[SPARK-xxxx][COMPONENT] Title`. Draft, WIP, MINOR, and TRIVIAL PRs may omit the JIRA ID. The component tag comes from the `COMPONENTS` registry in `dev/merge_spark_pr.py`, the source of truth: each entry pairs a PR-title tag with its canonical JIRA component name and accepted aliases (e.g. `Documentation` → `[DOC]`, `Pandas API on Spark` → `[PS]`, `SQL` → `[SQL]`). Read that registry to pick the right tag.\n\nUse `[FOLLOWUP]` only for small PRs that directly modify or correct unreleased earlier PRs. For separately planned work, non-trivial changes, or work outside the earlier JIRA's scope, create a separate JIRA ticket for each PR and use the normal title format without `[FOLLOWUP]`.\n\nInfer the PR title from the changes. If no ticket ID is given and the PR is not draft, WIP, MINOR, or TRIVIAL, create one using `dev/create_spark_jira.py`, using the PR title (without the JIRA ID and component tag) as the ticket title.\n\n    python3 dev/create_spark_jira.py \"<title>\" -c <component> { -t <type> | -p <parent-jira-id> }\n\n- **Component** (`-c`): the exact JIRA component name (not the PR title shorthand), e.g. \"SQL\", \"Spark Core\", \"PySpark\", \"Connect\". Run `python3 dev/create_spark_jira.py --list-components` for the full list.\n- **Issue type** (`-t`): \"Bug\", \"Improvement\", \"New Feature\", \"Test\", \"Documentation\", or \"Dependency upgrade\".\n- **Parent** (`-p`): if the user mentions a parent JIRA ticket (e.g., \"this is a subtask of SPARK-12345\"), pass it instead of `-t`. The issue type is automatically \"Sub-task\".\n\nThe script sets the latest unreleased version as the default affected version.\n\nAfter creating a JIRA ticket, print a prominent notice so the user does not miss it:\n\n    ============================================================\n    JIRA ticket created: SPARK-XXXXX\n    https://issues.apache.org/jira/browse/SPARK-XXXXX\n\n    Title:              <title>\n    Component(s):       <component>\n    Issue type:         <type>\n    Affected version(s): <version>\n    Priority:           <priority>\n\n    Please review and adjust these fields if needed.\n    ============================================================\n\nBefore writing the PR description, read `.github/PULL_REQUEST_TEMPLATE` and fill in every section from that file.\n\nDO NOT push to the upstream repo. Always push to the personal fork. Open PRs against `master` on the upstream repo.\n\nDO NOT force push or use `--amend` on pushed commits unless the user explicitly asks. If the remote branch has new commits, fetch and rebase before pushing.\n\nAlways get user approval before external operations such as pushing commits, creating PRs, or posting comments. Use `gh pr create` to open PRs. If `gh` is not installed, generate the GitHub PR URL for the user and recommend installing the GitHub CLI.\n\n## Project Instructions Discovery\n\nWhen exploring or working in any directory, always check for nested `AGENTS.md` files in that\ndirectory and its ancestors. Read and follow every applicable file; instructions in a more specific\ndirectory take precedence for that directory's scope.\n\nA directory may carry these instructions as `CLAUDE.md` rather than `AGENTS.md`, so check **both** names — a directory with only a `CLAUDE.md` has project instructions you must read too. Where both files exist, one is typically the real file and the other is a symlink to it, although either name may be the real file. Reading either one is sufficient. When adding instructions to a directory that has neither file, create AGENTS.md as the real file and add a CLAUDE.md symlink beside it (ln -s AGENTS.md CLAUDE.md). This keeps a single source of truth accessible under both names.\n\n## Versioning and Branch Policy\n\nWhen a change needs a version — `@since` annotations, config `.version(\"...\")` (`SQLConf` / `*Conf`), new `MimaExcludes` sections, etc. — use the version of the branch it first ships in, with `-SNAPSHOT` stripped. Determine that branch:\n\n- **PR opened against a non-`master` base branch** (e.g. a maintenance line like `branch-4.2`): use that base branch's version -- when you're checked out on it, that's just the working tree's `pom.xml`. The helper below covers only the common `master`-base case.\n- **PR opened against `master`:** most PRs merge to **both** `master` and the latest `branch-<N>.x` (the branch for the next feature release, e.g. `branch-4.x`), so use the `branch-<N>.x` version. The exception is **master-only** changes — use `master`'s version — which are only:\n  - breaking / binary-incompatible changes that can't ship in a minor release;\n  - dependency upgrades that don't fix a critical issue worth backporting.\n\nDo **not** just read `master`'s version: a normally-backported PR ships first in `branch-<N>.x`, whose version is lower than `master`'s. If unsure whether a change is master-only, ask the user.\n\n`dev/next_version_candidates.py` (no arguments) prints both candidate versions, reading from the local `apache/spark` remote (the `upstream` configured during pre-flight). It reports the mechanical facts only -- choosing between them per the rules above is the judgement call (the numbers below are illustrative and advance over time):\n\n    $ dev/next_version_candidates.py\n    master       5.0.0\n    branch-4.x   4.3.0\n\n## Security\n\nSecurity model: [SECURITY.md](./SECURITY.md)\n\nAgents that scan this repository should consult `SECURITY.md` for the project's threat model, in-scope / out-of-scope declarations, and known non-findings before reporting issues.\n",".cursor/mcp.json":"# Model Context Protocol (MCP) Configuration\n\nPath: `.cursor/mcp.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.cursor/mcp.json)",".claude-plugin/marketplace.json":"# Claude Plugin Marketplace Catalog\n\nPath: `.claude-plugin/marketplace.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/marketplace.json)",".claude-plugin/plugin.json":"# claude-plugin Plugin Manifest\n\nPath: `.claude-plugin/plugin.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/plugin.json)",".cursorrules (pyspark-etl-best-practices-cursorrules-prompt-file.mdc)":"---\ndescription: \"Cursor rules for PySpark ETL development with code style, joins, window functions, map operations, and Iceberg patterns.\"\nglobs: **/*\nalwaysApply: false\n---\nYou are an expert in PySpark, Spark SQL, Apache Iceberg, and production data engineering. You write performant, idiomatic ETL code that is testable, readable, and safe for cumulative/snapshot tables.\n\nFollow these rules when generating or reviewing PySpark code.\n\n# PySpark ETL Best Practices\n\n## 1. Project Structure\n\n### ETL class scaffold\n\nCreate a base class that manages the SparkSession lifecycle. Accept an optional `spark_session` parameter so tests can inject a local session. Use an abstract method for the job logic.\n\n```python\nfrom abc import ABC, abstractmethod\nfrom pyspark.sql import SparkSession\n\nclass BaseETL(ABC):\n    def __init__(self, config, app_name=\"ETL Job\", spark_session=None):\n        self.spark = spark_session or SparkSession.builder.appName(app_name).getOrCreate()\n        self.config = config\n        self.logger = logging.getLogger(self.__class__.__name__)\n\n    @abstractmethod\n    def run_job(self): ...\n\n    def stop(self):\n        self.spark.stop()\n```\n\n### Config — use a factory function\n\nKeep the dataclass as pure data and put CLI parsing in a standalone factory function. This makes configs easy to construct in tests without touching `sys.argv`.\n\n```python\n@dataclass\nclass MyConfig:\n    read_date: int = 20200101\n\ndef create_config() -> MyConfig:\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--read_date\", type=int, default=20200101)\n    args = parser.parse_args()\n    return MyConfig(read_date=args.read_date)\n```\n\n### Pipeline composition with `.transform()`\n\nKeep `run_job` as orchestration. Each step is a named method.\n\n```python\nevents = self.read_source().transform(self.enrich).transform(self.merge_with_existing)\n```\n\n### Use a shared reader for partition-aware reads\n\nBuild a generic reader utility that handles partition mechanics (date filters, hour ranges, latest-partition lookups). Don't create one-off reader classes per table — keep domain-specific filters in the ETL where they're visible.\n\n```python\nclass PartitionedReader:\n    @staticmethod\n    def read_latest(spark, table_name, partition_col):\n        row = spark.read.table(table_name).agg(F.max(partition_col)).first()\n        if row is None or row[0] is None:\n            return spark.createDataFrame([], spark.read.table(table_name).schema)\n        return spark.read.table(table_name).filter(F.col(partition_col) == row[0])\n\n    @staticmethod\n    def read_by_date(spark, table_name, partition_col, date_value):\n        return spark.read.table(table_name).filter(F.col(partition_col) == date_value)\n\n# Reader handles partitioning\nevents = PartitionedReader.read_by_date(spark, \"catalog.my_table\", \"event_date\", 20260319)\n\n# Business filters stay in the ETL\nevents = events.filter(F.col(\"event_type\").isin(\"login\", \"purchase\"))\n```\n\n### Shared merge utilities\n\nFor simple outer-join-with-coalesce merges, build a reusable merge function that handles aliasing, join key coalescing, and per-column defaults. Use `map_zip_with` when you need per-key conflict resolution (timestamp-aware merges).\n\n## 2. Code Style\n\n### Use `F.col()` — always use the `F.` prefix\n\nImport functions as `import pyspark.sql.functions as F` and use `F.col()`, `F.when()`, `F.lit()`, etc. throughout. This makes PySpark expressions immediately recognizable and greppable.\n\nAvoid `df.colA` attribute access — it binds the column to a specific DataFrame variable, which breaks after joins or when the variable is reassigned. Use `F.col()` with `.alias()` on the DataFrame if disambiguation is needed.\n\n```python\n# BAD — binds column to a specific DataFrame variable, breaks after joins\ndf.select(F.lower(df1.colA), F.upper(df2.colB))\n\n# GOOD\ndf.select(F.lower(F.col('colA')), F.upper(F.col('colB')))\n```\n\n### Extract complex conditions into named variables\n\nLimit logic inside `.filter()` or `F.when()` to 3 expressions. Extract the rest.\n\n```python\n# BAD — redundant logic hidden in nested parentheses\nF.when((F.col('status') == 'Delivered') | (((F.datediff('date_a', 'date_b') < 0) & ...)), 'Active')\n\n# GOOD\nis_delivered = (F.col('status') == 'Delivered')\ndate_passed = (F.datediff(F.col('date_a'), F.col('date_b')) < 0)\nhas_registration = (F.col('registration').rlike('.+'))\nF.when(is_delivered | (date_passed & has_registration), 'Active')\n```\n\n### Prefer `select` over `withColumn` chains\n\n`select` specifies the output schema in one pass. `withColumn` chains create intermediate DataFrames and can degrade performance — each call triggers a new projection in the query plan.\n\n```python\n# BAD — 3 intermediate DataFrames\ndf = df.withColumn(\"a\", F.col(\"a\").cast(\"double\"))\ndf = df.withColumn(\"b\", F.upper(F.col(\"b\")))\ndf = df.withColumn(\"c\", F.lit(1))\n\n# GOOD — 1 DataFrame, explicit schema contract\ndf = df.select(\n    F.col(\"a\").cast(\"double\"),\n    F.upper(F.col(\"b\")).alias(\"b\"),\n    F.lit(1).alias(\"c\"),\n)\n```\n\n### Use `alias` over `withColumnRenamed`\n\n```python\n# BAD\ndf.select('key', 'comments').withColumnRenamed('comments', 'num_comments')\n\n# GOOD\ndf.select('key', F.col('comments').alias('num_comments'))\n```\n\n### Chaining limits\n\nMax 5 statements per chain. Separate by operation type (select/filter vs withColumn vs join).\n\n```python\n# BAD — mixed concerns in one chain\ndf = (df.select('a', 'b', 'key')\n    .filter(F.col('a') == 'x')\n    .withColumn('ratio', F.col('a') / F.col('b'))\n    .join(df2, 'key', how='inner')\n    .drop('b'))\n\n# GOOD — separated by concern\ndf = df.select('a', 'b', 'key').filter(F.col('a') == 'x')\ndf = df.withColumn('ratio', F.col('a') / F.col('b'))\ndf = df.join(df2, 'key', how='inner').drop('b')\n```\n\n## 3. Joins\n\n### Always specify `how=` explicitly\n\n```python\n# BAD\ndf.join(other, 'key')\n\n# GOOD\ndf.join(other, 'key', how='inner')\n```\n\n### Prefer left joins over right joins\n\nFlip the DataFrame order and use `left` instead of `right` for readability — the primary DataFrame stays on the left.\n\n```python\nflights = flights.join(aircraft, 'aircraft_id', how='left')\n```\n\n### Use `.alias()` for disambiguation after joins\n\n```python\n# BAD — renaming every column before join\nfor c in columns:\n    flights = flights.withColumnRenamed(c, 'flights_' + c)\n\n# GOOD — alias the whole DataFrame\nflights = flights.alias('f')\nparking = parking.alias('p')\nresult = flights.join(parking, 'code', how='left').select(\n    F.col('f.start_time').alias('flight_start'),\n    F.col('p.total_time').alias('parking_total'),\n)\n```\n\n### Broadcast small dimension tables\n\nWhen joining a large fact DataFrame with a small lookup/dimension table, wrap the small side in `F.broadcast()` to skip the shuffle on the small side.\n\nUse broadcast for tables that are small enough to fit in executor memory — typically dimension/lookup tables (category lookups, country codes, config mappings). Spark auto-broadcasts tables under 10MB by default (`spark.sql.autoBroadcastJoinThreshold`), but an explicit hint is useful when Spark can't infer the size (e.g., after filters or transformations).\n\nTo check if a table is broadcast-worthy during development:\n- **Spark UI**: after a run, check the SQL tab — scan sizes are shown per table\n- **Quick row count**: `spark.read.table(\"catalog.my_dim\").count()` (dev only, not in production code)\n- **Query plan**: `df.explain()` — Spark shows `BroadcastHashJoin` if it auto-broadcasts, `SortMergeJoin` if it doesn't\n\n```python\ndf.join(F.broadcast(category_dim), 'category_id', how='left')\n```\n\n### Never use `.dropDuplicates()` as a crutch\n\nIf duplicate rows appear, find the root cause. `.dropDuplicates()` masks the problem and adds shuffle overhead.\n\n## 4. Window Functions\n\nUse `from pyspark.sql import Window as W` alongside `import pyspark.sql.functions as F`.\n\n### Always specify an explicit frame\n\nWithout a frame, Spark picks one that changes depending on whether `orderBy` is present.\n\n```python\n# BAD — F.sum gives running sum with orderBy, total without. Surprising.\nw = W.partitionBy('key').orderBy('num')\n\n# GOOD — explicit about what you want\nw = W.partitionBy('key').orderBy('num').rowsBetween(W.unboundedPreceding, W.unboundedFollowing)\n```\n\n### `row_number` + filter vs `first` — know the difference\n\n- `row_number` + filter = **drop rows**, keep the best one\n- `first` over window = **overwrite a column value**, keep all rows\n\n### Use `ignorenulls=True` with `first` and `last`\n\nWithout it, a null in the first row gives null for the entire partition.\n\n```python\n# BAD — returns None if first row is null\nF.first('version').over(window)\n\n# GOOD\nF.first('version', ignorenulls=True).over(window)\n```\n\n### Avoid empty `partitionBy()`\n\nIt forces all data into one partition. Use `.agg()` instead for global aggregations.\n\n```python\n# BAD — single partition, kills performance\nw = W.partitionBy()\ndf.select(F.sum('num').over(w))\n\n# GOOD\ndf.agg(F.sum('num').alias('total'))\n```\n\n## 5. Map & Array Higher-Order Functions\n\n### Use `map_zip_with` when merging maps with complex logic\n\n`map_concat` is fine for simple merges with no key overlap. When you need custom logic per key (e.g., keep the newer timestamp, pick the higher value), use `map_zip_with` — it gives you a per-key merge function instead of blindly letting one side win.\n\n```python\n# BAD — no control over conflict resolution\nmap_concat(existing_map, new_map)\n\n# GOOD — keep the entry with the later timestamp\nmap_zip_with(new_map, existing_map,\n    lambda key, v1, v2: (\n        F.when(v1.isNull(), v2)\n        .when(v2.isNull(), v1)\n        .otherwise(F.when(v1.event_ts >= v2.event_ts, v1).otherwise(v2))\n    )\n)\n```\n\n### Use `transform` + `array_max` to extract from nested structs\n\n```python\n# Extract max event_ts from a map of structs\narray_max(transform(map_values(my_map), lambda x: x.event_ts))\n```\n\n### Avoid UDFs — use native Spark functions first\n\nUDFs break Catalyst optimization and add serialization overhead. Before writing one, check if a built-in Spark function or higher-order function can do the job.\n\n## 6. Cumulative / Snapshot Table Patterns\n\n### Merges must be idempotent\n\nRe-running with the same data should produce the same result, not create duplicates.\n\n### Merges must be order-independent\n\nBackfilling old data should not overwrite newer data. Use an explicit ordering criterion (e.g., event timestamp, version number, partition date) to resolve conflicts — don't rely on positional precedence like `coalesce` argument order.\n\n### Validate primary key uniqueness after writes\n\nAdd audit steps that validate primary key uniqueness and check for nulls in key columns.\n\n## 7. Data Quality & Performance\n\n### Use `F.lit(None)` for empty columns, never empty strings\n\n```python\n# BAD\ndf = df.withColumn('foo', F.lit(''))\ndf = df.withColumn('foo', F.lit('NA'))\n\n# GOOD\ndf = df.withColumn('foo', F.lit(None))\n```\n\n### Avoid `.otherwise()` as a general catch-all\n\nUnknown values silently collapse into the otherwise bucket, hiding data quality issues.\n\n```python\n# BAD — a new platform_type you didn't anticipate becomes \"Other\" silently\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n .otherwise('Other')\n\n# GOOD — unmapped values stay null, surfacing gaps in your logic\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n```\n\n### No `.show()`, `.collect()`, `.printSchema()` in production\n\nThese trigger full materialization or add unnecessary driver overhead. Use them only for local debugging, never in deployed ETL code. `.count()` is acceptable when used intentionally (e.g., logging row counts for monitoring, forcing materialization before a DAG fork).\n\n### Use `persist()` intentionally\n\nOnly persist a DataFrame when it's referenced in multiple subsequent actions — otherwise the write action will materialize it for you. `.persist()` + `.count()` is a common pattern to force materialization and log row counts for debugging; use it when needed but be aware it adds a full scan.\n\nChoose the storage level based on your use case:\n- `MEMORY_AND_DISK` — safe default, spills to disk if memory is tight\n- `MEMORY_ONLY` — faster but risks recomputation if evicted\n- `DISK_ONLY` — for very large DataFrames that don't fit in memory\n\n## 8. Iceberg Write Patterns\n\n### Use `.byName()` for schema evolution safety\n\nColumn ordering doesn't matter — Spark matches by name, not position.\n\n```python\ndf.write.byName().mode(\"overwrite\").insertInto(\"catalog.my_table\")\n```\n\n### Use `__partitions` metadata table for latest partition reads\n\nIceberg exposes a `__partitions` metadata table. Use it to find the latest snapshot instead of scanning the full table.\n\n```python\npartition_df = spark.read.table(\"catalog.my_table__partitions\").select(\n    \"partition.partition_date\", \"partition.partition_hour\"\n)\nmax_partition = partition_df.orderBy(\n    F.col(\"partition_date\").desc(), F.col(\"partition_hour\").desc()\n).first()\nif max_partition is None:\n    raise ValueError(\"No partitions found in catalog.my_table\")\nlatest_date = max_partition[\"partition_date\"]\n```\n\n### Understand `write.distribution-mode`\n\n- `\"none\"` — no re-shuffle before writing. Fastest, but output file sizes depend on upstream partitioning.\n- `\"hash\"` — redistributes data by partition key. Produces evenly sized files but adds a shuffle.\n- `\"range\"` — sorts data by partition key before writing. Good for ordered scan performance but most expensive.\n"},"items":[{"name":"AGENTS.md","path":"AGENTS.md","rawUrl":"https://raw.githubusercontent.com/apache/spark/HEAD/AGENTS.md","title":"AI Agent Protocol & Instructions","category":"root-instruction","format":"markdown","content":"# Apache Spark\n\n## Pre-flight Checks\n\nBefore the first code edit or running test in a session, ensure a clean working environment. DO NOT skip these checks:\n\n1. Run `git remote -v` to identify the personal fork and upstream (`apache/spark`). If unclear, ask the user to configure their remotes following the standard convention (`origin` for the fork, `upstream` for `apache/spark`).\n2. If the latest commit on `<upstream>/master` is more than a day old (check with `git log -1 --format=\"%ci\" <upstream>/master`), run `git fetch <upstream> master`.\n3. If there are uncommitted changes (check with `git status`), ask the user to stash them before proceeding.\n4. Switch to the appropriate branch:\n   - **Existing PR**: resolve the PR branch name via `gh api repos/apache/spark/pulls/<number> --jq '.head.ref'`, then look for a local branch matching that name. If found, switch to it and inform the user. If not found, ask whether to fetch it or if there is a local branch under a different name.\n   - **New edits**: ask the user to choose: create a new git worktree from `<upstream>/master` with `--no-track` and work from there (recommended), or create and switch to a new branch from `<upstream>/master` with `--no-track`.\n   - **Running tests**: use `<upstream>/master`.\n\n## Development Notes\n\nSQL golden file tests are managed by `SQLQueryTestSuite` and its variants. Read the class documentation before running or updating these tests. DO NOT edit the generated golden files (`.sql.out`) directly. Always regenerate them when needed, and carefully review the diff to make sure it's expected.\n\nSpark Connect protocol is defined in proto files under `sql/connect/common/src/main/protobuf/`. Read the README there before modifying proto definitions.\n\nWhen adding a member to an existing class or object, follow the sectioning the file already uses and put the new member with the code it belongs with. The common failure mode is dropping it wherever it is first used without checking how the file is organized, splitting a section of unrelated code in the process. Beyond grouping related code there is no prescribed order -- the Databricks Scala guide, which Spark follows, asks only that a long class group its members into logical sections with comment headers. Do not reorganize existing members unless the change requires it.\n\nAvoid introducing non-ASCII characters in code or comments. String literals may contain non-ASCII when the content requires it (error messages, test data, etc.). Identifiers are ASCII by convention. The common failure mode is typographic characters (em-dash, smart quotes, ellipsis, non-breaking space) sneaking into comments; scalastyle flags some of these. Spot-check before committing: `grep -rn -P \"[^\\x00-\\x7F]\" <files>`.\n\nKeep source lines within 100 characters — the linters enforce this for Scala, Java, and Python, and LLMs commonly overrun it in comments and long expressions. A quick scan of just the changed files catches most cases in seconds, far cheaper than a CI round trip:\n\n    { git diff --name-only --diff-filter=ACM HEAD; git ls-files --others --exclude-standard; } \\\n      | grep -E '\\.(scala|java|py)$' | sort -u \\\n      | xargs -r awk 'length>100 && $0 !~ /^[[:space:]]*(import|package) / && $0 !~ /https?:\\/\\// \\\n          {print FILENAME\":\"FNR\": \"length\" chars\"}'\n\nThis is only a hint: it approximates the linters' exemptions (imports, URLs) rather than matching them exactly, so it can over- or under-report. The linters remain the source of truth.\n\n## Scala Test Base Classes\n\nWhen writing a new Scala test suite, pick the lowest base class that provides what the test actually needs. Spark uses the `AnyFunSuite` ScalaTest style throughout, so the bases below are the chain to choose from. Each adds capability on top of the previous:\n\n    SparkFunSuite                                                           (core)\n      <- PlanTest                                                           (sql/catalyst)\n        <- QueryTest                                                        (sql/core)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain JVM/Scala — no Spark SQL | `SparkFunSuite` | `core` utilities, RDD, network, util classes, etc. Adds per-test timeout, `testRetry`, `gridTest`, thread audit, fixed timezone/locale, `withTempDir`, `withLogAppender`, `checkError`. |\n| Catalyst plan tests — no `SparkSession` | `PlanTest` | Adds `comparePlans`, `normalizePlan`, `normalizeExprIds`. For analyzer / optimizer / planner rule tests. |\n| SQL/DataFrame tests — needs a `SparkSession` | `QueryTest` | Adds `checkAnswer`, codegen-on/off helpers. `spark: SparkSession` is abstract and must be supplied by a session-providing trait (see below). |\n\n### Providing a `SparkSession` for `QueryTest`\n\n`QueryTest` declares `spark: SparkSession` abstractly via `SparkSessionProvider`, so it cannot be instantiated on its own. A concrete suite mixes in one of the session-providing traits below:\n\n    QueryTest                                                               (abstract `spark`)\n      + SharedSparkSession (sql/core)        -> classic in-process `TestSparkSession`\n      + TestHiveSingleton  (sql/hive)        -> Hive-backed `TestHive` session\n\n| Session provider | Module / location | Typical usage |\n|---|---|---|\n| `SharedSparkSession` | `sql/core` | Already extends `QueryTest` for historical reasons, but still mix in `QueryTest` explicitly, e.g. `class X extends QueryTest with SharedSparkSession`. Default for tests under `sql/core`. |\n| `TestHiveSingleton` | `sql/hive` | Mixed in alongside `QueryTest`, e.g. `class X extends QueryTest with TestHiveSingleton`. Used by tests under `sql/hive`. |\n\n## Python Test Base Classes\n\nPySpark tests use the stdlib `unittest` framework: every suite subclasses `unittest.TestCase` (run via `python/run-tests`, see below). As with Scala, pick the lowest base that provides what the test actually needs. The bases live under `python/pyspark/testing/` and each adds capability on top of the previous:\n\n    unittest.TestCase                                     (stdlib)\n      <- PySparkBaseTestCase                              (pyspark.testing.utils)\n        <- ReusedPySparkTestCase                          (pyspark.testing.utils)\n          <- ReusedSQLTestCase                            (pyspark.testing.sqlutils)\n            <- PandasOnSparkTestCase                      (pyspark.testing.pandasutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain Python — no Spark | `unittest.TestCase` | Use the stdlib base directly. `PySparkBaseTestCase` is the same thing plus a SIGTERM fault-handler dump enabled when `PYSPARK_TEST_TIMEOUT` is set; subclass it only when you want that. |\n| RDD / `SparkContext` — no `SparkSession` | `PySparkTestCase` (fresh) or `ReusedPySparkTestCase` (shared) | Both create a `SparkContext(\"local[4]\")`. `PySparkTestCase` makes a new one per test (isolation); `ReusedPySparkTestCase` shares one per class (faster, the usual choice) and adds `quiet()` and an overridable `conf()` / `master()`. |\n| SQL / DataFrame — needs a `SparkSession` | `ReusedSQLTestCase` | The workhorse for classic tests under `python/pyspark/sql`. Adds a shared `cls.spark`, sample `cls.df` / `cls.testData`, and mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. Classic (non-Connect) mode. |\n| pandas API on Spark | `PandasOnSparkTestCase` | Extends `ReusedSQLTestCase` with Arrow enabled and pandas-on-Spark assertions (`PandasOnSparkTestUtils`); `ComparisonTestBase` builds on it. |\n\n### Spark Connect test bases\n\nSpark Connect suites live in `pyspark.testing.connectutils` and are auto-skipped (via `should_test_connect`) when Connect dependencies are missing:\n\n    PySparkBaseTestCase                                   (pyspark.testing.utils)\n      <- PlanOnlyTestFixture                              (pyspark.testing.connectutils)\n      <- ReusedConnectTestCase                            (pyspark.testing.connectutils)\n           <- ReusedMixedTestCase                         (pyspark.testing.connectutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plan / proto construction — no server | `PlanOnlyTestFixture` | Uses a `MockRemoteSession`; builds and inspects plans without a running Connect server. For proto / plan-shape assertions. |\n| Connect DataFrame — real session | `ReusedConnectTestCase` | The Connect analog of `ReusedSQLTestCase`; starts a session via `.remote(...)` (honoring `SPARK_CONNECT_TESTING_REMOTE`, default `local[4]`). Mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. |\n| Classic + Connect side by side | `ReusedMixedTestCase` | Extends `ReusedConnectTestCase`. For directly comparing classic vs Connect: it exposes a classic `self.spark` and a Connect `self.connect` so a test can run the same operation on each and assert they agree (`compare_by_show`, `both_conf`). Requires JVM access. |\n\n### Mixins and helpers\n\nThese are combined with a base above rather than used on their own:\n\n- `SQLTestUtils` — context managers `sql_conf`, `table`, `temp_view`, `view`, `database`, `function`, `temp_func`, `temp_env`; assumes `self.spark`. Already mixed into `ReusedSQLTestCase` and `ReusedConnectTestCase`.\n- `PySparkErrorTestUtils` — `check_error(...)` to assert on a `PySparkException`'s error class and message parameters. The Python counterpart of Scala's `checkError`.\n- `assertDataFrameEqual` / `assertSchemaEqual` (public API, from `pyspark.testing`) — standalone assertion functions that work in any test, no particular base class required.\n- Domain bases outside the main ladder: `SparkSessionTestCase` (`pyspark.testing.mlutils`, for `ml`), `MLlibTestCase` (`pyspark.testing.mllibutils`), and `PySparkStreamingTestCase` (`pyspark.testing.streamingutils`, DStreams).\n\n## Build and Test\n\nBuild and tests can take a long time, so prefer the narrowest relevant build or test command.\n\nFor build and test setup, including how to run tests and troubleshoot common\nlocal failures, see `docs/building-spark.md`.\n\nPrefer SBT over Maven for faster incremental compilation. Module names are defined in `project/SparkBuild.scala`.\n\nCompile a single module:\n\n    build/sbt <module>/compile\n\nCompile test code for a single module:\n\n    build/sbt <module>/Test/compile\n\nRun test suites by wildcard or full class name:\n\n    build/sbt '<module>/testOnly *MySuite'\n    build/sbt '<module>/testOnly org.apache.spark.sql.MySuite'\n\nRun test cases matching a substring:\n\n    build/sbt '<module>/testOnly *MySuite -- -z \"test name\"'\n\nRun test cases in an optional module:\n\n    build/sbt -P<maven-profiles> '<module>/testOnly *MySuite'\n\nFor faster iteration, keep SBT open in interactive mode:\n\n    build/sbt\n    > project <module>\n    > testOnly *MySuite\n\n### PySpark Tests\n\nPySpark tests require building Spark with Hive support first:\n\n    build/sbt -Phive package\n\nActivate the virtual environment specified by the user, or default to `.venv`:\n\n    source <venv>/bin/activate\n\nIf the default venv does not exist, create it:\n\n    python3 -m venv .venv\n    source .venv/bin/activate\n    pip install --upgrade pip\n    pip install --group dev\n\nRun a single test suite:\n\n    python/run-tests --testnames pyspark.sql.tests.arrow.test_arrow\n\nRun a single test case:\n\n    python/run-tests --testnames \"pyspark.sql.tests.test_catalog CatalogTests.test_current_database\"\n\n## Investigating PR CI Failures\n\nEnumerate all failing check runs first, then drill into each by type. Do not assume a single failure: a PR can fail tests, linters, and the build at once, and these surface through different channels.\n\nStep 1 — Get the fork owner and the latest commit SHA of the PR:\n\n    gh api repos/apache/spark/pulls/<PR_NUMBER> --jq '{owner: .head.repo.owner.login, sha: .head.sha}'\n\nStep 2 — List every failing check run on the fork's commit. This is the complete failure set:\n\n    gh api repos/<OWNER>/spark/commits/<SHA>/check-runs --paginate \\\n      --jq '.check_runs[] | select(.conclusion == \"failure\") | {name, id: .id}'\n\nA passing (or absent) \"Report test results\" does NOT mean CI is green. That check aggregates only test-case failures; linter, license, dependency, MiMa, compile, and doc-build failures are separate check runs that produce no test annotations. Always work from the list in Step 2, not from any single check.\n\nStep 3 — Drill into each failure according to its kind:\n\n- **Test jobs** (e.g. \"Report test results\", \"Build modules: ...\"): fetch failure annotations. Each annotation contains the test class, test name, and failure message:\n\n      gh api repos/<OWNER>/spark/check-runs/<CHECK_RUN_ID>/annotations\n\n- **Non-test jobs** (e.g. \"Linters, licenses, and dependencies\", \"Build\"): find the failed step, then read only that job's log:\n\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID> \\\n        --jq '{name, steps: [.steps[] | select(.conclusion == \"failure\") | .name]}'\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID>/logs\n\nAvoid downloading the large per-shard *test* job logs — they are very large and slow; use the annotations for those. Lint, license, dependency, and build job logs are small and fine to read directly when a step fails.\n\n## Checking PR Merge Status\n\nSpark merges PRs with `dev/merge_spark_pr.py`, not the GitHub merge button, so a **merged PR shows up on GitHub as Closed, not Merged** — its `merged` / `mergedAt` are empty, and backports to maintenance branches are plain pushes. **Do not read a Closed PR as rejected**; most Closed PRs were in fact merged.\n\nTo check whether, and to which branches, a PR was merged, run `dev/pr_merge_status.py <pr-number>`:\n\n    $ dev/pr_merge_status.py 56356\n    PR #56356 (base master): [SPARK-57295][SQL] Make database location validation ...\n    merged: yes\n      master       9357bc9ae05\n      branch-4.x   72edddb358e\n\nIt lists `master` and the latest major's release branches the commit reached (e.g. `branch-4.x`, `branch-4.2`) — or reports `open` or `closed without merging`. It needs `gh` authenticated and a non-shallow checkout, and exits non-zero with a clear message if the PR is unknown (404) or the environment isn't ready (no `gh`/auth, no apache/spark remote). Older majors' branches are omitted; pass `--all-branches` to list every branch the commit reached.\n\n## Pull Request Workflow\n\nPR title format is `[SPARK-xxxx][COMPONENT] Title`. Draft, WIP, MINOR, and TRIVIAL PRs may omit the JIRA ID. The component tag comes from the `COMPONENTS` registry in `dev/merge_spark_pr.py`, the source of truth: each entry pairs a PR-title tag with its canonical JIRA component name and accepted aliases (e.g. `Documentation` → `[DOC]`, `Pandas API on Spark` → `[PS]`, `SQL` → `[SQL]`). Read that registry to pick the right tag.\n\nUse `[FOLLOWUP]` only for small PRs that directly modify or correct unreleased earlier PRs. For separately planned work, non-trivial changes, or work outside the earlier JIRA's scope, create a separate JIRA ticket for each PR and use the normal title format without `[FOLLOWUP]`.\n\nInfer the PR title from the changes. If no ticket ID is given and the PR is not draft, WIP, MINOR, or TRIVIAL, create one using `dev/create_spark_jira.py`, using the PR title (without the JIRA ID and component tag) as the ticket title.\n\n    python3 dev/create_spark_jira.py \"<title>\" -c <component> { -t <type> | -p <parent-jira-id> }\n\n- **Component** (`-c`): the exact JIRA component name (not the PR title shorthand), e.g. \"SQL\", \"Spark Core\", \"PySpark\", \"Connect\". Run `python3 dev/create_spark_jira.py --list-components` for the full list.\n- **Issue type** (`-t`): \"Bug\", \"Improvement\", \"New Feature\", \"Test\", \"Documentation\", or \"Dependency upgrade\".\n- **Parent** (`-p`): if the user mentions a parent JIRA ticket (e.g., \"this is a subtask of SPARK-12345\"), pass it instead of `-t`. The issue type is automatically \"Sub-task\".\n\nThe script sets the latest unreleased version as the default affected version.\n\nAfter creating a JIRA ticket, print a prominent notice so the user does not miss it:\n\n    ============================================================\n    JIRA ticket created: SPARK-XXXXX\n    https://issues.apache.org/jira/browse/SPARK-XXXXX\n\n    Title:              <title>\n    Component(s):       <component>\n    Issue type:         <type>\n    Affected version(s): <version>\n    Priority:           <priority>\n\n    Please review and adjust these fields if needed.\n    ============================================================\n\nBefore writing the PR description, read `.github/PULL_REQUEST_TEMPLATE` and fill in every section from that file.\n\nDO NOT push to the upstream repo. Always push to the personal fork. Open PRs against `master` on the upstream repo.\n\nDO NOT force push or use `--amend` on pushed commits unless the user explicitly asks. If the remote branch has new commits, fetch and rebase before pushing.\n\nAlways get user approval before external operations such as pushing commits, creating PRs, or posting comments. Use `gh pr create` to open PRs. If `gh` is not installed, generate the GitHub PR URL for the user and recommend installing the GitHub CLI.\n\n## Project Instructions Discovery\n\nWhen exploring or working in any directory, always check for nested `AGENTS.md` files in that\ndirectory and its ancestors. Read and follow every applicable file; instructions in a more specific\ndirectory take precedence for that directory's scope.\n\nA directory may carry these instructions as `CLAUDE.md` rather than `AGENTS.md`, so check **both** names — a directory with only a `CLAUDE.md` has project instructions you must read too. Where both files exist, one is typically the real file and the other is a symlink to it, although either name may be the real file. Reading either one is sufficient. When adding instructions to a directory that has neither file, create AGENTS.md as the real file and add a CLAUDE.md symlink beside it (ln -s AGENTS.md CLAUDE.md). This keeps a single source of truth accessible under both names.\n\n## Versioning and Branch Policy\n\nWhen a change needs a version — `@since` annotations, config `.version(\"...\")` (`SQLConf` / `*Conf`), new `MimaExcludes` sections, etc. — use the version of the branch it first ships in, with `-SNAPSHOT` stripped. Determine that branch:\n\n- **PR opened against a non-`master` base branch** (e.g. a maintenance line like `branch-4.2`): use that base branch's version -- when you're checked out on it, that's just the working tree's `pom.xml`. The helper below covers only the common `master`-base case.\n- **PR opened against `master`:** most PRs merge to **both** `master` and the latest `branch-<N>.x` (the branch for the next feature release, e.g. `branch-4.x`), so use the `branch-<N>.x` version. The exception is **master-only** changes — use `master`'s version — which are only:\n  - breaking / binary-incompatible changes that can't ship in a minor release;\n  - dependency upgrades that don't fix a critical issue worth backporting.\n\nDo **not** just read `master`'s version: a normally-backported PR ships first in `branch-<N>.x`, whose version is lower than `master`'s. If unsure whether a change is master-only, ask the user.\n\n`dev/next_version_candidates.py` (no arguments) prints both candidate versions, reading from the local `apache/spark` remote (the `upstream` configured during pre-flight). It reports the mechanical facts only -- choosing between them per the rules above is the judgement call (the numbers below are illustrative and advance over time):\n\n    $ dev/next_version_candidates.py\n    master       5.0.0\n    branch-4.x   4.3.0\n\n## Security\n\nSecurity model: [SECURITY.md](./SECURITY.md)\n\nAgents that scan this repository should consult `SECURITY.md` for the project's threat model, in-scope / out-of-scope declarations, and known non-findings before reporting issues.\n","isInternal":false,"tokens":4704,"sizeBytes":19827},{"name":"CLAUDE.md","path":"CLAUDE.md","rawUrl":"https://raw.githubusercontent.com/apache/spark/HEAD/CLAUDE.md","title":"Claude Agent Guidelines & System Prompt","category":"claude-rule","format":"markdown","content":"# Apache Spark\n\n## Pre-flight Checks\n\nBefore the first code edit or running test in a session, ensure a clean working environment. DO NOT skip these checks:\n\n1. Run `git remote -v` to identify the personal fork and upstream (`apache/spark`). If unclear, ask the user to configure their remotes following the standard convention (`origin` for the fork, `upstream` for `apache/spark`).\n2. If the latest commit on `<upstream>/master` is more than a day old (check with `git log -1 --format=\"%ci\" <upstream>/master`), run `git fetch <upstream> master`.\n3. If there are uncommitted changes (check with `git status`), ask the user to stash them before proceeding.\n4. Switch to the appropriate branch:\n   - **Existing PR**: resolve the PR branch name via `gh api repos/apache/spark/pulls/<number> --jq '.head.ref'`, then look for a local branch matching that name. If found, switch to it and inform the user. If not found, ask whether to fetch it or if there is a local branch under a different name.\n   - **New edits**: ask the user to choose: create a new git worktree from `<upstream>/master` with `--no-track` and work from there (recommended), or create and switch to a new branch from `<upstream>/master` with `--no-track`.\n   - **Running tests**: use `<upstream>/master`.\n\n## Development Notes\n\nSQL golden file tests are managed by `SQLQueryTestSuite` and its variants. Read the class documentation before running or updating these tests. DO NOT edit the generated golden files (`.sql.out`) directly. Always regenerate them when needed, and carefully review the diff to make sure it's expected.\n\nSpark Connect protocol is defined in proto files under `sql/connect/common/src/main/protobuf/`. Read the README there before modifying proto definitions.\n\nWhen adding a member to an existing class or object, follow the sectioning the file already uses and put the new member with the code it belongs with. The common failure mode is dropping it wherever it is first used without checking how the file is organized, splitting a section of unrelated code in the process. Beyond grouping related code there is no prescribed order -- the Databricks Scala guide, which Spark follows, asks only that a long class group its members into logical sections with comment headers. Do not reorganize existing members unless the change requires it.\n\nAvoid introducing non-ASCII characters in code or comments. String literals may contain non-ASCII when the content requires it (error messages, test data, etc.). Identifiers are ASCII by convention. The common failure mode is typographic characters (em-dash, smart quotes, ellipsis, non-breaking space) sneaking into comments; scalastyle flags some of these. Spot-check before committing: `grep -rn -P \"[^\\x00-\\x7F]\" <files>`.\n\nKeep source lines within 100 characters — the linters enforce this for Scala, Java, and Python, and LLMs commonly overrun it in comments and long expressions. A quick scan of just the changed files catches most cases in seconds, far cheaper than a CI round trip:\n\n    { git diff --name-only --diff-filter=ACM HEAD; git ls-files --others --exclude-standard; } \\\n      | grep -E '\\.(scala|java|py)$' | sort -u \\\n      | xargs -r awk 'length>100 && $0 !~ /^[[:space:]]*(import|package) / && $0 !~ /https?:\\/\\// \\\n          {print FILENAME\":\"FNR\": \"length\" chars\"}'\n\nThis is only a hint: it approximates the linters' exemptions (imports, URLs) rather than matching them exactly, so it can over- or under-report. The linters remain the source of truth.\n\n## Scala Test Base Classes\n\nWhen writing a new Scala test suite, pick the lowest base class that provides what the test actually needs. Spark uses the `AnyFunSuite` ScalaTest style throughout, so the bases below are the chain to choose from. Each adds capability on top of the previous:\n\n    SparkFunSuite                                                           (core)\n      <- PlanTest                                                           (sql/catalyst)\n        <- QueryTest                                                        (sql/core)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain JVM/Scala — no Spark SQL | `SparkFunSuite` | `core` utilities, RDD, network, util classes, etc. Adds per-test timeout, `testRetry`, `gridTest`, thread audit, fixed timezone/locale, `withTempDir`, `withLogAppender`, `checkError`. |\n| Catalyst plan tests — no `SparkSession` | `PlanTest` | Adds `comparePlans`, `normalizePlan`, `normalizeExprIds`. For analyzer / optimizer / planner rule tests. |\n| SQL/DataFrame tests — needs a `SparkSession` | `QueryTest` | Adds `checkAnswer`, codegen-on/off helpers. `spark: SparkSession` is abstract and must be supplied by a session-providing trait (see below). |\n\n### Providing a `SparkSession` for `QueryTest`\n\n`QueryTest` declares `spark: SparkSession` abstractly via `SparkSessionProvider`, so it cannot be instantiated on its own. A concrete suite mixes in one of the session-providing traits below:\n\n    QueryTest                                                               (abstract `spark`)\n      + SharedSparkSession (sql/core)        -> classic in-process `TestSparkSession`\n      + TestHiveSingleton  (sql/hive)        -> Hive-backed `TestHive` session\n\n| Session provider | Module / location | Typical usage |\n|---|---|---|\n| `SharedSparkSession` | `sql/core` | Already extends `QueryTest` for historical reasons, but still mix in `QueryTest` explicitly, e.g. `class X extends QueryTest with SharedSparkSession`. Default for tests under `sql/core`. |\n| `TestHiveSingleton` | `sql/hive` | Mixed in alongside `QueryTest`, e.g. `class X extends QueryTest with TestHiveSingleton`. Used by tests under `sql/hive`. |\n\n## Python Test Base Classes\n\nPySpark tests use the stdlib `unittest` framework: every suite subclasses `unittest.TestCase` (run via `python/run-tests`, see below). As with Scala, pick the lowest base that provides what the test actually needs. The bases live under `python/pyspark/testing/` and each adds capability on top of the previous:\n\n    unittest.TestCase                                     (stdlib)\n      <- PySparkBaseTestCase                              (pyspark.testing.utils)\n        <- ReusedPySparkTestCase                          (pyspark.testing.utils)\n          <- ReusedSQLTestCase                            (pyspark.testing.sqlutils)\n            <- PandasOnSparkTestCase                      (pyspark.testing.pandasutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain Python — no Spark | `unittest.TestCase` | Use the stdlib base directly. `PySparkBaseTestCase` is the same thing plus a SIGTERM fault-handler dump enabled when `PYSPARK_TEST_TIMEOUT` is set; subclass it only when you want that. |\n| RDD / `SparkContext` — no `SparkSession` | `PySparkTestCase` (fresh) or `ReusedPySparkTestCase` (shared) | Both create a `SparkContext(\"local[4]\")`. `PySparkTestCase` makes a new one per test (isolation); `ReusedPySparkTestCase` shares one per class (faster, the usual choice) and adds `quiet()` and an overridable `conf()` / `master()`. |\n| SQL / DataFrame — needs a `SparkSession` | `ReusedSQLTestCase` | The workhorse for classic tests under `python/pyspark/sql`. Adds a shared `cls.spark`, sample `cls.df` / `cls.testData`, and mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. Classic (non-Connect) mode. |\n| pandas API on Spark | `PandasOnSparkTestCase` | Extends `ReusedSQLTestCase` with Arrow enabled and pandas-on-Spark assertions (`PandasOnSparkTestUtils`); `ComparisonTestBase` builds on it. |\n\n### Spark Connect test bases\n\nSpark Connect suites live in `pyspark.testing.connectutils` and are auto-skipped (via `should_test_connect`) when Connect dependencies are missing:\n\n    PySparkBaseTestCase                                   (pyspark.testing.utils)\n      <- PlanOnlyTestFixture                              (pyspark.testing.connectutils)\n      <- ReusedConnectTestCase                            (pyspark.testing.connectutils)\n           <- ReusedMixedTestCase                         (pyspark.testing.connectutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plan / proto construction — no server | `PlanOnlyTestFixture` | Uses a `MockRemoteSession`; builds and inspects plans without a running Connect server. For proto / plan-shape assertions. |\n| Connect DataFrame — real session | `ReusedConnectTestCase` | The Connect analog of `ReusedSQLTestCase`; starts a session via `.remote(...)` (honoring `SPARK_CONNECT_TESTING_REMOTE`, default `local[4]`). Mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. |\n| Classic + Connect side by side | `ReusedMixedTestCase` | Extends `ReusedConnectTestCase`. For directly comparing classic vs Connect: it exposes a classic `self.spark` and a Connect `self.connect` so a test can run the same operation on each and assert they agree (`compare_by_show`, `both_conf`). Requires JVM access. |\n\n### Mixins and helpers\n\nThese are combined with a base above rather than used on their own:\n\n- `SQLTestUtils` — context managers `sql_conf`, `table`, `temp_view`, `view`, `database`, `function`, `temp_func`, `temp_env`; assumes `self.spark`. Already mixed into `ReusedSQLTestCase` and `ReusedConnectTestCase`.\n- `PySparkErrorTestUtils` — `check_error(...)` to assert on a `PySparkException`'s error class and message parameters. The Python counterpart of Scala's `checkError`.\n- `assertDataFrameEqual` / `assertSchemaEqual` (public API, from `pyspark.testing`) — standalone assertion functions that work in any test, no particular base class required.\n- Domain bases outside the main ladder: `SparkSessionTestCase` (`pyspark.testing.mlutils`, for `ml`), `MLlibTestCase` (`pyspark.testing.mllibutils`), and `PySparkStreamingTestCase` (`pyspark.testing.streamingutils`, DStreams).\n\n## Build and Test\n\nBuild and tests can take a long time, so prefer the narrowest relevant build or test command.\n\nFor build and test setup, including how to run tests and troubleshoot common\nlocal failures, see `docs/building-spark.md`.\n\nPrefer SBT over Maven for faster incremental compilation. Module names are defined in `project/SparkBuild.scala`.\n\nCompile a single module:\n\n    build/sbt <module>/compile\n\nCompile test code for a single module:\n\n    build/sbt <module>/Test/compile\n\nRun test suites by wildcard or full class name:\n\n    build/sbt '<module>/testOnly *MySuite'\n    build/sbt '<module>/testOnly org.apache.spark.sql.MySuite'\n\nRun test cases matching a substring:\n\n    build/sbt '<module>/testOnly *MySuite -- -z \"test name\"'\n\nRun test cases in an optional module:\n\n    build/sbt -P<maven-profiles> '<module>/testOnly *MySuite'\n\nFor faster iteration, keep SBT open in interactive mode:\n\n    build/sbt\n    > project <module>\n    > testOnly *MySuite\n\n### PySpark Tests\n\nPySpark tests require building Spark with Hive support first:\n\n    build/sbt -Phive package\n\nActivate the virtual environment specified by the user, or default to `.venv`:\n\n    source <venv>/bin/activate\n\nIf the default venv does not exist, create it:\n\n    python3 -m venv .venv\n    source .venv/bin/activate\n    pip install --upgrade pip\n    pip install --group dev\n\nRun a single test suite:\n\n    python/run-tests --testnames pyspark.sql.tests.arrow.test_arrow\n\nRun a single test case:\n\n    python/run-tests --testnames \"pyspark.sql.tests.test_catalog CatalogTests.test_current_database\"\n\n## Investigating PR CI Failures\n\nEnumerate all failing check runs first, then drill into each by type. Do not assume a single failure: a PR can fail tests, linters, and the build at once, and these surface through different channels.\n\nStep 1 — Get the fork owner and the latest commit SHA of the PR:\n\n    gh api repos/apache/spark/pulls/<PR_NUMBER> --jq '{owner: .head.repo.owner.login, sha: .head.sha}'\n\nStep 2 — List every failing check run on the fork's commit. This is the complete failure set:\n\n    gh api repos/<OWNER>/spark/commits/<SHA>/check-runs --paginate \\\n      --jq '.check_runs[] | select(.conclusion == \"failure\") | {name, id: .id}'\n\nA passing (or absent) \"Report test results\" does NOT mean CI is green. That check aggregates only test-case failures; linter, license, dependency, MiMa, compile, and doc-build failures are separate check runs that produce no test annotations. Always work from the list in Step 2, not from any single check.\n\nStep 3 — Drill into each failure according to its kind:\n\n- **Test jobs** (e.g. \"Report test results\", \"Build modules: ...\"): fetch failure annotations. Each annotation contains the test class, test name, and failure message:\n\n      gh api repos/<OWNER>/spark/check-runs/<CHECK_RUN_ID>/annotations\n\n- **Non-test jobs** (e.g. \"Linters, licenses, and dependencies\", \"Build\"): find the failed step, then read only that job's log:\n\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID> \\\n        --jq '{name, steps: [.steps[] | select(.conclusion == \"failure\") | .name]}'\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID>/logs\n\nAvoid downloading the large per-shard *test* job logs — they are very large and slow; use the annotations for those. Lint, license, dependency, and build job logs are small and fine to read directly when a step fails.\n\n## Checking PR Merge Status\n\nSpark merges PRs with `dev/merge_spark_pr.py`, not the GitHub merge button, so a **merged PR shows up on GitHub as Closed, not Merged** — its `merged` / `mergedAt` are empty, and backports to maintenance branches are plain pushes. **Do not read a Closed PR as rejected**; most Closed PRs were in fact merged.\n\nTo check whether, and to which branches, a PR was merged, run `dev/pr_merge_status.py <pr-number>`:\n\n    $ dev/pr_merge_status.py 56356\n    PR #56356 (base master): [SPARK-57295][SQL] Make database location validation ...\n    merged: yes\n      master       9357bc9ae05\n      branch-4.x   72edddb358e\n\nIt lists `master` and the latest major's release branches the commit reached (e.g. `branch-4.x`, `branch-4.2`) — or reports `open` or `closed without merging`. It needs `gh` authenticated and a non-shallow checkout, and exits non-zero with a clear message if the PR is unknown (404) or the environment isn't ready (no `gh`/auth, no apache/spark remote). Older majors' branches are omitted; pass `--all-branches` to list every branch the commit reached.\n\n## Pull Request Workflow\n\nPR title format is `[SPARK-xxxx][COMPONENT] Title`. Draft, WIP, MINOR, and TRIVIAL PRs may omit the JIRA ID. The component tag comes from the `COMPONENTS` registry in `dev/merge_spark_pr.py`, the source of truth: each entry pairs a PR-title tag with its canonical JIRA component name and accepted aliases (e.g. `Documentation` → `[DOC]`, `Pandas API on Spark` → `[PS]`, `SQL` → `[SQL]`). Read that registry to pick the right tag.\n\nUse `[FOLLOWUP]` only for small PRs that directly modify or correct unreleased earlier PRs. For separately planned work, non-trivial changes, or work outside the earlier JIRA's scope, create a separate JIRA ticket for each PR and use the normal title format without `[FOLLOWUP]`.\n\nInfer the PR title from the changes. If no ticket ID is given and the PR is not draft, WIP, MINOR, or TRIVIAL, create one using `dev/create_spark_jira.py`, using the PR title (without the JIRA ID and component tag) as the ticket title.\n\n    python3 dev/create_spark_jira.py \"<title>\" -c <component> { -t <type> | -p <parent-jira-id> }\n\n- **Component** (`-c`): the exact JIRA component name (not the PR title shorthand), e.g. \"SQL\", \"Spark Core\", \"PySpark\", \"Connect\". Run `python3 dev/create_spark_jira.py --list-components` for the full list.\n- **Issue type** (`-t`): \"Bug\", \"Improvement\", \"New Feature\", \"Test\", \"Documentation\", or \"Dependency upgrade\".\n- **Parent** (`-p`): if the user mentions a parent JIRA ticket (e.g., \"this is a subtask of SPARK-12345\"), pass it instead of `-t`. The issue type is automatically \"Sub-task\".\n\nThe script sets the latest unreleased version as the default affected version.\n\nAfter creating a JIRA ticket, print a prominent notice so the user does not miss it:\n\n    ============================================================\n    JIRA ticket created: SPARK-XXXXX\n    https://issues.apache.org/jira/browse/SPARK-XXXXX\n\n    Title:              <title>\n    Component(s):       <component>\n    Issue type:         <type>\n    Affected version(s): <version>\n    Priority:           <priority>\n\n    Please review and adjust these fields if needed.\n    ============================================================\n\nBefore writing the PR description, read `.github/PULL_REQUEST_TEMPLATE` and fill in every section from that file.\n\nDO NOT push to the upstream repo. Always push to the personal fork. Open PRs against `master` on the upstream repo.\n\nDO NOT force push or use `--amend` on pushed commits unless the user explicitly asks. If the remote branch has new commits, fetch and rebase before pushing.\n\nAlways get user approval before external operations such as pushing commits, creating PRs, or posting comments. Use `gh pr create` to open PRs. If `gh` is not installed, generate the GitHub PR URL for the user and recommend installing the GitHub CLI.\n\n## Project Instructions Discovery\n\nWhen exploring or working in any directory, always check for nested `AGENTS.md` files in that\ndirectory and its ancestors. Read and follow every applicable file; instructions in a more specific\ndirectory take precedence for that directory's scope.\n\nA directory may carry these instructions as `CLAUDE.md` rather than `AGENTS.md`, so check **both** names — a directory with only a `CLAUDE.md` has project instructions you must read too. Where both files exist, one is typically the real file and the other is a symlink to it, although either name may be the real file. Reading either one is sufficient. When adding instructions to a directory that has neither file, create AGENTS.md as the real file and add a CLAUDE.md symlink beside it (ln -s AGENTS.md CLAUDE.md). This keeps a single source of truth accessible under both names.\n\n## Versioning and Branch Policy\n\nWhen a change needs a version — `@since` annotations, config `.version(\"...\")` (`SQLConf` / `*Conf`), new `MimaExcludes` sections, etc. — use the version of the branch it first ships in, with `-SNAPSHOT` stripped. Determine that branch:\n\n- **PR opened against a non-`master` base branch** (e.g. a maintenance line like `branch-4.2`): use that base branch's version -- when you're checked out on it, that's just the working tree's `pom.xml`. The helper below covers only the common `master`-base case.\n- **PR opened against `master`:** most PRs merge to **both** `master` and the latest `branch-<N>.x` (the branch for the next feature release, e.g. `branch-4.x`), so use the `branch-<N>.x` version. The exception is **master-only** changes — use `master`'s version — which are only:\n  - breaking / binary-incompatible changes that can't ship in a minor release;\n  - dependency upgrades that don't fix a critical issue worth backporting.\n\nDo **not** just read `master`'s version: a normally-backported PR ships first in `branch-<N>.x`, whose version is lower than `master`'s. If unsure whether a change is master-only, ask the user.\n\n`dev/next_version_candidates.py` (no arguments) prints both candidate versions, reading from the local `apache/spark` remote (the `upstream` configured during pre-flight). It reports the mechanical facts only -- choosing between them per the rules above is the judgement call (the numbers below are illustrative and advance over time):\n\n    $ dev/next_version_candidates.py\n    master       5.0.0\n    branch-4.x   4.3.0\n\n## Security\n\nSecurity model: [SECURITY.md](./SECURITY.md)\n\nAgents that scan this repository should consult `SECURITY.md` for the project's threat model, in-scope / out-of-scope declarations, and known non-findings before reporting issues.\n","isInternal":false,"tokens":4704,"sizeBytes":19827},{"name":"marketplace.json","path":".claude-plugin/marketplace.json","rawUrl":"https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/marketplace.json","title":"Claude Plugin Marketplace Catalog","category":"marketplace","format":"json","content":"# Claude Plugin Marketplace Catalog\n\nPath: `.claude-plugin/marketplace.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/marketplace.json)","tokens":300,"sizeBytes":191},{"name":"plugin.json","path":".claude-plugin/plugin.json","rawUrl":"https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/plugin.json","title":"claude-plugin Plugin Manifest","category":"plugin-manifest","format":"json","content":"# claude-plugin Plugin Manifest\n\nPath: `.claude-plugin/plugin.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/plugin.json)","tokens":300,"sizeBytes":177},{"name":"mcp.json","path":".cursor/mcp.json","rawUrl":"https://raw.githubusercontent.com/apache/spark/HEAD/.cursor/mcp.json","title":"Model Context Protocol (MCP) Configuration","category":"mcp-config","format":"json","content":"# Model Context Protocol (MCP) Configuration\n\nPath: `.cursor/mcp.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.cursor/mcp.json)","tokens":300,"sizeBytes":170},{"name":".cursorrules","path":".cursorrules (pyspark-etl-best-practices-cursorrules-prompt-file.mdc)","rawUrl":"https://raw.githubusercontent.com/PatrickJS/awesome-cursorrules/main/rules/pyspark-etl-best-practices-cursorrules-prompt-file.mdc","title":"Awesome Cursorrules: pyspark etl best practices","category":"cursor-rule","format":"markdown","content":"---\ndescription: \"Cursor rules for PySpark ETL development with code style, joins, window functions, map operations, and Iceberg patterns.\"\nglobs: **/*\nalwaysApply: false\n---\nYou are an expert in PySpark, Spark SQL, Apache Iceberg, and production data engineering. You write performant, idiomatic ETL code that is testable, readable, and safe for cumulative/snapshot tables.\n\nFollow these rules when generating or reviewing PySpark code.\n\n# PySpark ETL Best Practices\n\n## 1. Project Structure\n\n### ETL class scaffold\n\nCreate a base class that manages the SparkSession lifecycle. Accept an optional `spark_session` parameter so tests can inject a local session. Use an abstract method for the job logic.\n\n```python\nfrom abc import ABC, abstractmethod\nfrom pyspark.sql import SparkSession\n\nclass BaseETL(ABC):\n    def __init__(self, config, app_name=\"ETL Job\", spark_session=None):\n        self.spark = spark_session or SparkSession.builder.appName(app_name).getOrCreate()\n        self.config = config\n        self.logger = logging.getLogger(self.__class__.__name__)\n\n    @abstractmethod\n    def run_job(self): ...\n\n    def stop(self):\n        self.spark.stop()\n```\n\n### Config — use a factory function\n\nKeep the dataclass as pure data and put CLI parsing in a standalone factory function. This makes configs easy to construct in tests without touching `sys.argv`.\n\n```python\n@dataclass\nclass MyConfig:\n    read_date: int = 20200101\n\ndef create_config() -> MyConfig:\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--read_date\", type=int, default=20200101)\n    args = parser.parse_args()\n    return MyConfig(read_date=args.read_date)\n```\n\n### Pipeline composition with `.transform()`\n\nKeep `run_job` as orchestration. Each step is a named method.\n\n```python\nevents = self.read_source().transform(self.enrich).transform(self.merge_with_existing)\n```\n\n### Use a shared reader for partition-aware reads\n\nBuild a generic reader utility that handles partition mechanics (date filters, hour ranges, latest-partition lookups). Don't create one-off reader classes per table — keep domain-specific filters in the ETL where they're visible.\n\n```python\nclass PartitionedReader:\n    @staticmethod\n    def read_latest(spark, table_name, partition_col):\n        row = spark.read.table(table_name).agg(F.max(partition_col)).first()\n        if row is None or row[0] is None:\n            return spark.createDataFrame([], spark.read.table(table_name).schema)\n        return spark.read.table(table_name).filter(F.col(partition_col) == row[0])\n\n    @staticmethod\n    def read_by_date(spark, table_name, partition_col, date_value):\n        return spark.read.table(table_name).filter(F.col(partition_col) == date_value)\n\n# Reader handles partitioning\nevents = PartitionedReader.read_by_date(spark, \"catalog.my_table\", \"event_date\", 20260319)\n\n# Business filters stay in the ETL\nevents = events.filter(F.col(\"event_type\").isin(\"login\", \"purchase\"))\n```\n\n### Shared merge utilities\n\nFor simple outer-join-with-coalesce merges, build a reusable merge function that handles aliasing, join key coalescing, and per-column defaults. Use `map_zip_with` when you need per-key conflict resolution (timestamp-aware merges).\n\n## 2. Code Style\n\n### Use `F.col()` — always use the `F.` prefix\n\nImport functions as `import pyspark.sql.functions as F` and use `F.col()`, `F.when()`, `F.lit()`, etc. throughout. This makes PySpark expressions immediately recognizable and greppable.\n\nAvoid `df.colA` attribute access — it binds the column to a specific DataFrame variable, which breaks after joins or when the variable is reassigned. Use `F.col()` with `.alias()` on the DataFrame if disambiguation is needed.\n\n```python\n# BAD — binds column to a specific DataFrame variable, breaks after joins\ndf.select(F.lower(df1.colA), F.upper(df2.colB))\n\n# GOOD\ndf.select(F.lower(F.col('colA')), F.upper(F.col('colB')))\n```\n\n### Extract complex conditions into named variables\n\nLimit logic inside `.filter()` or `F.when()` to 3 expressions. Extract the rest.\n\n```python\n# BAD — redundant logic hidden in nested parentheses\nF.when((F.col('status') == 'Delivered') | (((F.datediff('date_a', 'date_b') < 0) & ...)), 'Active')\n\n# GOOD\nis_delivered = (F.col('status') == 'Delivered')\ndate_passed = (F.datediff(F.col('date_a'), F.col('date_b')) < 0)\nhas_registration = (F.col('registration').rlike('.+'))\nF.when(is_delivered | (date_passed & has_registration), 'Active')\n```\n\n### Prefer `select` over `withColumn` chains\n\n`select` specifies the output schema in one pass. `withColumn` chains create intermediate DataFrames and can degrade performance — each call triggers a new projection in the query plan.\n\n```python\n# BAD — 3 intermediate DataFrames\ndf = df.withColumn(\"a\", F.col(\"a\").cast(\"double\"))\ndf = df.withColumn(\"b\", F.upper(F.col(\"b\")))\ndf = df.withColumn(\"c\", F.lit(1))\n\n# GOOD — 1 DataFrame, explicit schema contract\ndf = df.select(\n    F.col(\"a\").cast(\"double\"),\n    F.upper(F.col(\"b\")).alias(\"b\"),\n    F.lit(1).alias(\"c\"),\n)\n```\n\n### Use `alias` over `withColumnRenamed`\n\n```python\n# BAD\ndf.select('key', 'comments').withColumnRenamed('comments', 'num_comments')\n\n# GOOD\ndf.select('key', F.col('comments').alias('num_comments'))\n```\n\n### Chaining limits\n\nMax 5 statements per chain. Separate by operation type (select/filter vs withColumn vs join).\n\n```python\n# BAD — mixed concerns in one chain\ndf = (df.select('a', 'b', 'key')\n    .filter(F.col('a') == 'x')\n    .withColumn('ratio', F.col('a') / F.col('b'))\n    .join(df2, 'key', how='inner')\n    .drop('b'))\n\n# GOOD — separated by concern\ndf = df.select('a', 'b', 'key').filter(F.col('a') == 'x')\ndf = df.withColumn('ratio', F.col('a') / F.col('b'))\ndf = df.join(df2, 'key', how='inner').drop('b')\n```\n\n## 3. Joins\n\n### Always specify `how=` explicitly\n\n```python\n# BAD\ndf.join(other, 'key')\n\n# GOOD\ndf.join(other, 'key', how='inner')\n```\n\n### Prefer left joins over right joins\n\nFlip the DataFrame order and use `left` instead of `right` for readability — the primary DataFrame stays on the left.\n\n```python\nflights = flights.join(aircraft, 'aircraft_id', how='left')\n```\n\n### Use `.alias()` for disambiguation after joins\n\n```python\n# BAD — renaming every column before join\nfor c in columns:\n    flights = flights.withColumnRenamed(c, 'flights_' + c)\n\n# GOOD — alias the whole DataFrame\nflights = flights.alias('f')\nparking = parking.alias('p')\nresult = flights.join(parking, 'code', how='left').select(\n    F.col('f.start_time').alias('flight_start'),\n    F.col('p.total_time').alias('parking_total'),\n)\n```\n\n### Broadcast small dimension tables\n\nWhen joining a large fact DataFrame with a small lookup/dimension table, wrap the small side in `F.broadcast()` to skip the shuffle on the small side.\n\nUse broadcast for tables that are small enough to fit in executor memory — typically dimension/lookup tables (category lookups, country codes, config mappings). Spark auto-broadcasts tables under 10MB by default (`spark.sql.autoBroadcastJoinThreshold`), but an explicit hint is useful when Spark can't infer the size (e.g., after filters or transformations).\n\nTo check if a table is broadcast-worthy during development:\n- **Spark UI**: after a run, check the SQL tab — scan sizes are shown per table\n- **Quick row count**: `spark.read.table(\"catalog.my_dim\").count()` (dev only, not in production code)\n- **Query plan**: `df.explain()` — Spark shows `BroadcastHashJoin` if it auto-broadcasts, `SortMergeJoin` if it doesn't\n\n```python\ndf.join(F.broadcast(category_dim), 'category_id', how='left')\n```\n\n### Never use `.dropDuplicates()` as a crutch\n\nIf duplicate rows appear, find the root cause. `.dropDuplicates()` masks the problem and adds shuffle overhead.\n\n## 4. Window Functions\n\nUse `from pyspark.sql import Window as W` alongside `import pyspark.sql.functions as F`.\n\n### Always specify an explicit frame\n\nWithout a frame, Spark picks one that changes depending on whether `orderBy` is present.\n\n```python\n# BAD — F.sum gives running sum with orderBy, total without. Surprising.\nw = W.partitionBy('key').orderBy('num')\n\n# GOOD — explicit about what you want\nw = W.partitionBy('key').orderBy('num').rowsBetween(W.unboundedPreceding, W.unboundedFollowing)\n```\n\n### `row_number` + filter vs `first` — know the difference\n\n- `row_number` + filter = **drop rows**, keep the best one\n- `first` over window = **overwrite a column value**, keep all rows\n\n### Use `ignorenulls=True` with `first` and `last`\n\nWithout it, a null in the first row gives null for the entire partition.\n\n```python\n# BAD — returns None if first row is null\nF.first('version').over(window)\n\n# GOOD\nF.first('version', ignorenulls=True).over(window)\n```\n\n### Avoid empty `partitionBy()`\n\nIt forces all data into one partition. Use `.agg()` instead for global aggregations.\n\n```python\n# BAD — single partition, kills performance\nw = W.partitionBy()\ndf.select(F.sum('num').over(w))\n\n# GOOD\ndf.agg(F.sum('num').alias('total'))\n```\n\n## 5. Map & Array Higher-Order Functions\n\n### Use `map_zip_with` when merging maps with complex logic\n\n`map_concat` is fine for simple merges with no key overlap. When you need custom logic per key (e.g., keep the newer timestamp, pick the higher value), use `map_zip_with` — it gives you a per-key merge function instead of blindly letting one side win.\n\n```python\n# BAD — no control over conflict resolution\nmap_concat(existing_map, new_map)\n\n# GOOD — keep the entry with the later timestamp\nmap_zip_with(new_map, existing_map,\n    lambda key, v1, v2: (\n        F.when(v1.isNull(), v2)\n        .when(v2.isNull(), v1)\n        .otherwise(F.when(v1.event_ts >= v2.event_ts, v1).otherwise(v2))\n    )\n)\n```\n\n### Use `transform` + `array_max` to extract from nested structs\n\n```python\n# Extract max event_ts from a map of structs\narray_max(transform(map_values(my_map), lambda x: x.event_ts))\n```\n\n### Avoid UDFs — use native Spark functions first\n\nUDFs break Catalyst optimization and add serialization overhead. Before writing one, check if a built-in Spark function or higher-order function can do the job.\n\n## 6. Cumulative / Snapshot Table Patterns\n\n### Merges must be idempotent\n\nRe-running with the same data should produce the same result, not create duplicates.\n\n### Merges must be order-independent\n\nBackfilling old data should not overwrite newer data. Use an explicit ordering criterion (e.g., event timestamp, version number, partition date) to resolve conflicts — don't rely on positional precedence like `coalesce` argument order.\n\n### Validate primary key uniqueness after writes\n\nAdd audit steps that validate primary key uniqueness and check for nulls in key columns.\n\n## 7. Data Quality & Performance\n\n### Use `F.lit(None)` for empty columns, never empty strings\n\n```python\n# BAD\ndf = df.withColumn('foo', F.lit(''))\ndf = df.withColumn('foo', F.lit('NA'))\n\n# GOOD\ndf = df.withColumn('foo', F.lit(None))\n```\n\n### Avoid `.otherwise()` as a general catch-all\n\nUnknown values silently collapse into the otherwise bucket, hiding data quality issues.\n\n```python\n# BAD — a new platform_type you didn't anticipate becomes \"Other\" silently\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n .otherwise('Other')\n\n# GOOD — unmapped values stay null, surfacing gaps in your logic\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n```\n\n### No `.show()`, `.collect()`, `.printSchema()` in production\n\nThese trigger full materialization or add unnecessary driver overhead. Use them only for local debugging, never in deployed ETL code. `.count()` is acceptable when used intentionally (e.g., logging row counts for monitoring, forcing materialization before a DAG fork).\n\n### Use `persist()` intentionally\n\nOnly persist a DataFrame when it's referenced in multiple subsequent actions — otherwise the write action will materialize it for you. `.persist()` + `.count()` is a common pattern to force materialization and log row counts for debugging; use it when needed but be aware it adds a full scan.\n\nChoose the storage level based on your use case:\n- `MEMORY_AND_DISK` — safe default, spills to disk if memory is tight\n- `MEMORY_ONLY` — faster but risks recomputation if evicted\n- `DISK_ONLY` — for very large DataFrames that don't fit in memory\n\n## 8. Iceberg Write Patterns\n\n### Use `.byName()` for schema evolution safety\n\nColumn ordering doesn't matter — Spark matches by name, not position.\n\n```python\ndf.write.byName().mode(\"overwrite\").insertInto(\"catalog.my_table\")\n```\n\n### Use `__partitions` metadata table for latest partition reads\n\nIceberg exposes a `__partitions` metadata table. Use it to find the latest snapshot instead of scanning the full table.\n\n```python\npartition_df = spark.read.table(\"catalog.my_table__partitions\").select(\n    \"partition.partition_date\", \"partition.partition_hour\"\n)\nmax_partition = partition_df.orderBy(\n    F.col(\"partition_date\").desc(), F.col(\"partition_hour\").desc()\n).first()\nif max_partition is None:\n    raise ValueError(\"No partitions found in catalog.my_table\")\nlatest_date = max_partition[\"partition_date\"]\n```\n\n### Understand `write.distribution-mode`\n\n- `\"none\"` — no re-shuffle before writing. Fastest, but output file sizes depend on upstream partitioning.\n- `\"hash\"` — redistributes data by partition key. Produces evenly sized files but adds a shuffle.\n- `\"range\"` — sorts data by partition key before writing. Good for ordered scan performance but most expensive.\n","tokens":3262,"sizeBytes":13538}],"systemPromptSnippet":"<yakaai_skills repo=\"apache/spark\">\n<!-- File: AGENTS.md (Tokens: ~4704 | Category: root-instruction) -->\n# Apache Spark\n\n## Pre-flight Checks\n\nBefore the first code edit or running test in a session, ensure a clean working environment. DO NOT skip these checks:\n\n1. Run `git remote -v` to identify the personal fork and upstream (`apache/spark`). If unclear, ask the user to configure their remotes following the standard convention (`origin` for the fork, `upstream` for `apache/spark`).\n2. If the latest commit on `<upstream>/master` is more than a day old (check with `git log -1 --format=\"%ci\" <upstream>/master`), run `git fetch <upstream> master`.\n3. If there are uncommitted changes (check with `git status`), ask the user to stash them before proceeding.\n4. Switch to the appropriate branch:\n   - **Existing PR**: resolve the PR branch name via `gh api repos/apache/spark/pulls/<number> --jq '.head.ref'`, then look for a local branch matching that name. If found, switch to it and inform the user. If not found, ask whether to fetch it or if there is a local branch under a different name.\n   - **New edits**: ask the user to choose: create a new git worktree from `<upstream>/master` with `--no-track` and work from there (recommended), or create and switch to a new branch from `<upstream>/master` with `--no-track`.\n   - **Running tests**: use `<upstream>/master`.\n\n## Development Notes\n\nSQL golden file tests are managed by `SQLQueryTestSuite` and its variants. Read the class documentation before running or updating these tests. DO NOT edit the generated golden files (`.sql.out`) directly. Always regenerate them when needed, and carefully review the diff to make sure it's expected.\n\nSpark Connect protocol is defined in proto files under `sql/connect/common/src/main/protobuf/`. Read the README there before modifying proto definitions.\n\nWhen adding a member to an existing class or object, follow the sectioning the file already uses and put the new member with the code it belongs with. The common failure mode is dropping it wherever it is first used without checking how the file is organized, splitting a section of unrelated code in the process. Beyond grouping related code there is no prescribed order -- the Databricks Scala guide, which Spark follows, asks only that a long class group its members into logical sections with comment headers. Do not reorganize existing members unless the change requires it.\n\nAvoid introducing non-ASCII characters in code or comments. String literals may contain non-ASCII when the content requires it (error messages, test data, etc.). Identifiers are ASCII by convention. The common failure mode is typographic characters (em-dash, smart quotes, ellipsis, non-breaking space) sneaking into comments; scalastyle flags some of these. Spot-check before committing: `grep -rn -P \"[^\\x00-\\x7F]\" <files>`.\n\nKeep source lines within 100 characters — the linters enforce this for Scala, Java, and Python, and LLMs commonly overrun it in comments and long expressions. A quick scan of just the changed files catches most cases in seconds, far cheaper than a CI round trip:\n\n    { git diff --name-only --diff-filter=ACM HEAD; git ls-files --others --exclude-standard; } \\\n      | grep -E '\\.(scala|java|py)$' | sort -u \\\n      | xargs -r awk 'length>100 && $0 !~ /^[[:space:]]*(import|package) / && $0 !~ /https?:\\/\\// \\\n          {print FILENAME\":\"FNR\": \"length\" chars\"}'\n\nThis is only a hint: it approximates the linters' exemptions (imports, URLs) rather than matching them exactly, so it can over- or under-report. The linters remain the source of truth.\n\n## Scala Test Base Classes\n\nWhen writing a new Scala test suite, pick the lowest base class that provides what the test actually needs. Spark uses the `AnyFunSuite` ScalaTest style throughout, so the bases below are the chain to choose from. Each adds capability on top of the previous:\n\n    SparkFunSuite                                                           (core)\n      <- PlanTest                                                           (sql/catalyst)\n        <- QueryTest                                                        (sql/core)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain JVM/Scala — no Spark SQL | `SparkFunSuite` | `core` utilities, RDD, network, util classes, etc. Adds per-test timeout, `testRetry`, `gridTest`, thread audit, fixed timezone/locale, `withTempDir`, `withLogAppender`, `checkError`. |\n| Catalyst plan tests — no `SparkSession` | `PlanTest` | Adds `comparePlans`, `normalizePlan`, `normalizeExprIds`. For analyzer / optimizer / planner rule tests. |\n| SQL/DataFrame tests — needs a `SparkSession` | `QueryTest` | Adds `checkAnswer`, codegen-on/off helpers. `spark: SparkSession` is abstract and must be supplied by a session-providing trait (see below). |\n\n### Providing a `SparkSession` for `QueryTest`\n\n`QueryTest` declares `spark: SparkSession` abstractly via `SparkSessionProvider`, so it cannot be instantiated on its own. A concrete suite mixes in one of the session-providing traits below:\n\n    QueryTest                                                               (abstract `spark`)\n      + SharedSparkSession (sql/core)        -> classic in-process `TestSparkSession`\n      + TestHiveSingleton  (sql/hive)        -> Hive-backed `TestHive` session\n\n| Session provider | Module / location | Typical usage |\n|---|---|---|\n| `SharedSparkSession` | `sql/core` | Already extends `QueryTest` for historical reasons, but still mix in `QueryTest` explicitly, e.g. `class X extends QueryTest with SharedSparkSession`. Default for tests under `sql/core`. |\n| `TestHiveSingleton` | `sql/hive` | Mixed in alongside `QueryTest`, e.g. `class X extends QueryTest with TestHiveSingleton`. Used by tests under `sql/hive`. |\n\n## Python Test Base Classes\n\nPySpark tests use the stdlib `unittest` framework: every suite subclasses `unittest.TestCase` (run via `python/run-tests`, see below). As with Scala, pick the lowest base that provides what the test actually needs. The bases live under `python/pyspark/testing/` and each adds capability on top of the previous:\n\n    unittest.TestCase                                     (stdlib)\n      <- PySparkBaseTestCase                              (pyspark.testing.utils)\n        <- ReusedPySparkTestCase                          (pyspark.testing.utils)\n          <- ReusedSQLTestCase                            (pyspark.testing.sqlutils)\n            <- PandasOnSparkTestCase                      (pyspark.testing.pandasutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain Python — no Spark | `unittest.TestCase` | Use the stdlib base directly. `PySparkBaseTestCase` is the same thing plus a SIGTERM fault-handler dump enabled when `PYSPARK_TEST_TIMEOUT` is set; subclass it only when you want that. |\n| RDD / `SparkContext` — no `SparkSession` | `PySparkTestCase` (fresh) or `ReusedPySparkTestCase` (shared) | Both create a `SparkContext(\"local[4]\")`. `PySparkTestCase` makes a new one per test (isolation); `ReusedPySparkTestCase` shares one per class (faster, the usual choice) and adds `quiet()` and an overridable `conf()` / `master()`. |\n| SQL / DataFrame — needs a `SparkSession` | `ReusedSQLTestCase` | The workhorse for classic tests under `python/pyspark/sql`. Adds a shared `cls.spark`, sample `cls.df` / `cls.testData`, and mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. Classic (non-Connect) mode. |\n| pandas API on Spark | `PandasOnSparkTestCase` | Extends `ReusedSQLTestCase` with Arrow enabled and pandas-on-Spark assertions (`PandasOnSparkTestUtils`); `ComparisonTestBase` builds on it. |\n\n### Spark Connect test bases\n\nSpark Connect suites live in `pyspark.testing.connectutils` and are auto-skipped (via `should_test_connect`) when Connect dependencies are missing:\n\n    PySparkBaseTestCase                                   (pyspark.testing.utils)\n      <- PlanOnlyTestFixture                              (pyspark.testing.connectutils)\n      <- ReusedConnectTestCase                            (pyspark.testing.connectutils)\n           <- ReusedMixedTestCase                         (pyspark.testing.connectutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plan / proto construction — no server | `PlanOnlyTestFixture` | Uses a `MockRemoteSession`; builds and inspects plans without a running Connect server. For proto / plan-shape assertions. |\n| Connect DataFrame — real session | `ReusedConnectTestCase` | The Connect analog of `ReusedSQLTestCase`; starts a session via `.remote(...)` (honoring `SPARK_CONNECT_TESTING_REMOTE`, default `local[4]`). Mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. |\n| Classic + Connect side by side | `ReusedMixedTestCase` | Extends `ReusedConnectTestCase`. For directly comparing classic vs Connect: it exposes a classic `self.spark` and a Connect `self.connect` so a test can run the same operation on each and assert they agree (`compare_by_show`, `both_conf`). Requires JVM access. |\n\n### Mixins and helpers\n\nThese are combined with a base above rather than used on their own:\n\n- `SQLTestUtils` — context managers `sql_conf`, `table`, `temp_view`, `view`, `database`, `function`, `temp_func`, `temp_env`; assumes `self.spark`. Already mixed into `ReusedSQLTestCase` and `ReusedConnectTestCase`.\n- `PySparkErrorTestUtils` — `check_error(...)` to assert on a `PySparkException`'s error class and message parameters. The Python counterpart of Scala's `checkError`.\n- `assertDataFrameEqual` / `assertSchemaEqual` (public API, from `pyspark.testing`) — standalone assertion functions that work in any test, no particular base class required.\n- Domain bases outside the main ladder: `SparkSessionTestCase` (`pyspark.testing.mlutils`, for `ml`), `MLlibTestCase` (`pyspark.testing.mllibutils`), and `PySparkStreamingTestCase` (`pyspark.testing.streamingutils`, DStreams).\n\n## Build and Test\n\nBuild and tests can take a long time, so prefer the narrowest relevant build or test command.\n\nFor build and test setup, including how to run tests and troubleshoot common\nlocal failures, see `docs/building-spark.md`.\n\nPrefer SBT over Maven for faster incremental compilation. Module names are defined in `project/SparkBuild.scala`.\n\nCompile a single module:\n\n    build/sbt <module>/compile\n\nCompile test code for a single module:\n\n    build/sbt <module>/Test/compile\n\nRun test suites by wildcard or full class name:\n\n    build/sbt '<module>/testOnly *MySuite'\n    build/sbt '<module>/testOnly org.apache.spark.sql.MySuite'\n\nRun test cases matching a substring:\n\n    build/sbt '<module>/testOnly *MySuite -- -z \"test name\"'\n\nRun test cases in an optional module:\n\n    build/sbt -P<maven-profiles> '<module>/testOnly *MySuite'\n\nFor faster iteration, keep SBT open in interactive mode:\n\n    build/sbt\n    > project <module>\n    > testOnly *MySuite\n\n### PySpark Tests\n\nPySpark tests require building Spark with Hive support first:\n\n    build/sbt -Phive package\n\nActivate the virtual environment specified by the user, or default to `.venv`:\n\n    source <venv>/bin/activate\n\nIf the default venv does not exist, create it:\n\n    python3 -m venv .venv\n    source .venv/bin/activate\n    pip install --upgrade pip\n    pip install --group dev\n\nRun a single test suite:\n\n    python/run-tests --testnames pyspark.sql.tests.arrow.test_arrow\n\nRun a single test case:\n\n    python/run-tests --testnames \"pyspark.sql.tests.test_catalog CatalogTests.test_current_database\"\n\n## Investigating PR CI Failures\n\nEnumerate all failing check runs first, then drill into each by type. Do not assume a single failure: a PR can fail tests, linters, and the build at once, and these surface through different channels.\n\nStep 1 — Get the fork owner and the latest commit SHA of the PR:\n\n    gh api repos/apache/spark/pulls/<PR_NUMBER> --jq '{owner: .head.repo.owner.login, sha: .head.sha}'\n\nStep 2 — List every failing check run on the fork's commit. This is the complete failure set:\n\n    gh api repos/<OWNER>/spark/commits/<SHA>/check-runs --paginate \\\n      --jq '.check_runs[] | select(.conclusion == \"failure\") | {name, id: .id}'\n\nA passing (or absent) \"Report test results\" does NOT mean CI is green. That check aggregates only test-case failures; linter, license, dependency, MiMa, compile, and doc-build failures are separate check runs that produce no test annotations. Always work from the list in Step 2, not from any single check.\n\nStep 3 — Drill into each failure according to its kind:\n\n- **Test jobs** (e.g. \"Report test results\", \"Build modules: ...\"): fetch failure annotations. Each annotation contains the test class, test name, and failure message:\n\n      gh api repos/<OWNER>/spark/check-runs/<CHECK_RUN_ID>/annotations\n\n- **Non-test jobs** (e.g. \"Linters, licenses, and dependencies\", \"Build\"): find the failed step, then read only that job's log:\n\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID> \\\n        --jq '{name, steps: [.steps[] | select(.conclusion == \"failure\") | .name]}'\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID>/logs\n\nAvoid downloading the large per-shard *test* job logs — they are very large and slow; use the annotations for those. Lint, license, dependency, and build job logs are small and fine to read directly when a step fails.\n\n## Checking PR Merge Status\n\nSpark merges PRs with `dev/merge_spark_pr.py`, not the GitHub merge button, so a **merged PR shows up on GitHub as Closed, not Merged** — its `merged` / `mergedAt` are empty, and backports to maintenance branches are plain pushes. **Do not read a Closed PR as rejected**; most Closed PRs were in fact merged.\n\nTo check whether, and to which branches, a PR was merged, run `dev/pr_merge_status.py <pr-number>`:\n\n    $ dev/pr_merge_status.py 56356\n    PR #56356 (base master): [SPARK-57295][SQL] Make database location validation ...\n    merged: yes\n      master       9357bc9ae05\n      branch-4.x   72edddb358e\n\nIt lists `master` and the latest major's release branches the commit reached (e.g. `branch-4.x`, `branch-4.2`) — or reports `open` or `closed without merging`. It needs `gh` authenticated and a non-shallow checkout, and exits non-zero with a clear message if the PR is unknown (404) or the environment isn't ready (no `gh`/auth, no apache/spark remote). Older majors' branches are omitted; pass `--all-branches` to list every branch the commit reached.\n\n## Pull Request Workflow\n\nPR title format is `[SPARK-xxxx][COMPONENT] Title`. Draft, WIP, MINOR, and TRIVIAL PRs may omit the JIRA ID. The component tag comes from the `COMPONENTS` registry in `dev/merge_spark_pr.py`, the source of truth: each entry pairs a PR-title tag with its canonical JIRA component name and accepted aliases (e.g. `Documentation` → `[DOC]`, `Pandas API on Spark` → `[PS]`, `SQL` → `[SQL]`). Read that registry to pick the right tag.\n\nUse `[FOLLOWUP]` only for small PRs that directly modify or correct unreleased earlier PRs. For separately planned work, non-trivial changes, or work outside the earlier JIRA's scope, create a separate JIRA ticket for each PR and use the normal title format without `[FOLLOWUP]`.\n\nInfer the PR title from the changes. If no ticket ID is given and the PR is not draft, WIP, MINOR, or TRIVIAL, create one using `dev/create_spark_jira.py`, using the PR title (without the JIRA ID and component tag) as the ticket title.\n\n    python3 dev/create_spark_jira.py \"<title>\" -c <component> { -t <type> | -p <parent-jira-id> }\n\n- **Component** (`-c`): the exact JIRA component name (not the PR title shorthand), e.g. \"SQL\", \"Spark Core\", \"PySpark\", \"Connect\". Run `python3 dev/create_spark_jira.py --list-components` for the full list.\n- **Issue type** (`-t`): \"Bug\", \"Improvement\", \"New Feature\", \"Test\", \"Documentation\", or \"Dependency upgrade\".\n- **Parent** (`-p`): if the user mentions a parent JIRA ticket (e.g., \"this is a subtask of SPARK-12345\"), pass it instead of `-t`. The issue type is automatically \"Sub-task\".\n\nThe script sets the latest unreleased version as the default affected version.\n\nAfter creating a JIRA ticket, print a prominent notice so the user does not miss it:\n\n    ============================================================\n    JIRA ticket created: SPARK-XXXXX\n    https://issues.apache.org/jira/browse/SPARK-XXXXX\n\n    Title:              <title>\n    Component(s):       <component>\n    Issue type:         <type>\n    Affected version(s): <version>\n    Priority:           <priority>\n\n    Please review and adjust these fields if needed.\n    ============================================================\n\nBefore writing the PR description, read `.github/PULL_REQUEST_TEMPLATE` and fill in every section from that file.\n\nDO NOT push to the upstream repo. Always push to the personal fork. Open PRs against `master` on the upstream repo.\n\nDO NOT force push or use `--amend` on pushed commits unless the user explicitly asks. If the remote branch has new commits, fetch and rebase before pushing.\n\nAlways get user approval before external operations such as pushing commits, creating PRs, or posting comments. Use `gh pr create` to open PRs. If `gh` is not installed, generate the GitHub PR URL for the user and recommend installing the GitHub CLI.\n\n## Project Instructions Discovery\n\nWhen exploring or working in any directory, always check for nested `AGENTS.md` files in that\ndirectory and its ancestors. Read and follow every applicable file; instructions in a more specific\ndirectory take precedence for that directory's scope.\n\nA directory may carry these instructions as `CLAUDE.md` rather than `AGENTS.md`, so check **both** names — a directory with only a `CLAUDE.md` has project instructions you must read too. Where both files exist, one is typically the real file and the other is a symlink to it, although either name may be the real file. Reading either one is sufficient. When adding instructions to a directory that has neither file, create AGENTS.md as the real file and add a CLAUDE.md symlink beside it (ln -s AGENTS.md CLAUDE.md). This keeps a single source of truth accessible under both names.\n\n## Versioning and Branch Policy\n\nWhen a change needs a version — `@since` annotations, config `.version(\"...\")` (`SQLConf` / `*Conf`), new `MimaExcludes` sections, etc. — use the version of the branch it first ships in, with `-SNAPSHOT` stripped. Determine that branch:\n\n- **PR opened against a non-`master` base branch** (e.g. a maintenance line like `branch-4.2`): use that base branch's version -- when you're checked out on it, that's just the working tree's `pom.xml`. The helper below covers only the common `master`-base case.\n- **PR opened against `master`:** most PRs merge to **both** `master` and the latest `branch-<N>.x` (the branch for the next feature release, e.g. `branch-4.x`), so use the `branch-<N>.x` version. The exception is **master-only** changes — use `master`'s version — which are only:\n  - breaking / binary-incompatible changes that can't ship in a minor release;\n  - dependency upgrades that don't fix a critical issue worth backporting.\n\nDo **not** just read `master`'s version: a normally-backported PR ships first in `branch-<N>.x`, whose version is lower than `master`'s. If unsure whether a change is master-only, ask the user.\n\n`dev/next_version_candidates.py` (no arguments) prints both candidate versions, reading from the local `apache/spark` remote (the `upstream` configured during pre-flight). It reports the mechanical facts only -- choosing between them per the rules above is the judgement call (the numbers below are illustrative and advance over time):\n\n    $ dev/next_version_candidates.py\n    master       5.0.0\n    branch-4.x   4.3.0\n\n## Security\n\nSecurity model: [SECURITY.md](./SECURITY.md)\n\nAgents that scan this repository should consult `SECURITY.md` for the project's threat model, in-scope / out-of-scope declarations, and known non-findings before reporting issues.\n\n<!-- File: CLAUDE.md (Tokens: ~4704 | Category: claude-rule) -->\n# Apache Spark\n\n## Pre-flight Checks\n\nBefore the first code edit or running test in a session, ensure a clean working environment. DO NOT skip these checks:\n\n1. Run `git remote -v` to identify the personal fork and upstream (`apache/spark`). If unclear, ask the user to configure their remotes following the standard convention (`origin` for the fork, `upstream` for `apache/spark`).\n2. If the latest commit on `<upstream>/master` is more than a day old (check with `git log -1 --format=\"%ci\" <upstream>/master`), run `git fetch <upstream> master`.\n3. If there are uncommitted changes (check with `git status`), ask the user to stash them before proceeding.\n4. Switch to the appropriate branch:\n   - **Existing PR**: resolve the PR branch name via `gh api repos/apache/spark/pulls/<number> --jq '.head.ref'`, then look for a local branch matching that name. If found, switch to it and inform the user. If not found, ask whether to fetch it or if there is a local branch under a different name.\n   - **New edits**: ask the user to choose: create a new git worktree from `<upstream>/master` with `--no-track` and work from there (recommended), or create and switch to a new branch from `<upstream>/master` with `--no-track`.\n   - **Running tests**: use `<upstream>/master`.\n\n## Development Notes\n\nSQL golden file tests are managed by `SQLQueryTestSuite` and its variants. Read the class documentation before running or updating these tests. DO NOT edit the generated golden files (`.sql.out`) directly. Always regenerate them when needed, and carefully review the diff to make sure it's expected.\n\nSpark Connect protocol is defined in proto files under `sql/connect/common/src/main/protobuf/`. Read the README there before modifying proto definitions.\n\nWhen adding a member to an existing class or object, follow the sectioning the file already uses and put the new member with the code it belongs with. The common failure mode is dropping it wherever it is first used without checking how the file is organized, splitting a section of unrelated code in the process. Beyond grouping related code there is no prescribed order -- the Databricks Scala guide, which Spark follows, asks only that a long class group its members into logical sections with comment headers. Do not reorganize existing members unless the change requires it.\n\nAvoid introducing non-ASCII characters in code or comments. String literals may contain non-ASCII when the content requires it (error messages, test data, etc.). Identifiers are ASCII by convention. The common failure mode is typographic characters (em-dash, smart quotes, ellipsis, non-breaking space) sneaking into comments; scalastyle flags some of these. Spot-check before committing: `grep -rn -P \"[^\\x00-\\x7F]\" <files>`.\n\nKeep source lines within 100 characters — the linters enforce this for Scala, Java, and Python, and LLMs commonly overrun it in comments and long expressions. A quick scan of just the changed files catches most cases in seconds, far cheaper than a CI round trip:\n\n    { git diff --name-only --diff-filter=ACM HEAD; git ls-files --others --exclude-standard; } \\\n      | grep -E '\\.(scala|java|py)$' | sort -u \\\n      | xargs -r awk 'length>100 && $0 !~ /^[[:space:]]*(import|package) / && $0 !~ /https?:\\/\\// \\\n          {print FILENAME\":\"FNR\": \"length\" chars\"}'\n\nThis is only a hint: it approximates the linters' exemptions (imports, URLs) rather than matching them exactly, so it can over- or under-report. The linters remain the source of truth.\n\n## Scala Test Base Classes\n\nWhen writing a new Scala test suite, pick the lowest base class that provides what the test actually needs. Spark uses the `AnyFunSuite` ScalaTest style throughout, so the bases below are the chain to choose from. Each adds capability on top of the previous:\n\n    SparkFunSuite                                                           (core)\n      <- PlanTest                                                           (sql/catalyst)\n        <- QueryTest                                                        (sql/core)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain JVM/Scala — no Spark SQL | `SparkFunSuite` | `core` utilities, RDD, network, util classes, etc. Adds per-test timeout, `testRetry`, `gridTest`, thread audit, fixed timezone/locale, `withTempDir`, `withLogAppender`, `checkError`. |\n| Catalyst plan tests — no `SparkSession` | `PlanTest` | Adds `comparePlans`, `normalizePlan`, `normalizeExprIds`. For analyzer / optimizer / planner rule tests. |\n| SQL/DataFrame tests — needs a `SparkSession` | `QueryTest` | Adds `checkAnswer`, codegen-on/off helpers. `spark: SparkSession` is abstract and must be supplied by a session-providing trait (see below). |\n\n### Providing a `SparkSession` for `QueryTest`\n\n`QueryTest` declares `spark: SparkSession` abstractly via `SparkSessionProvider`, so it cannot be instantiated on its own. A concrete suite mixes in one of the session-providing traits below:\n\n    QueryTest                                                               (abstract `spark`)\n      + SharedSparkSession (sql/core)        -> classic in-process `TestSparkSession`\n      + TestHiveSingleton  (sql/hive)        -> Hive-backed `TestHive` session\n\n| Session provider | Module / location | Typical usage |\n|---|---|---|\n| `SharedSparkSession` | `sql/core` | Already extends `QueryTest` for historical reasons, but still mix in `QueryTest` explicitly, e.g. `class X extends QueryTest with SharedSparkSession`. Default for tests under `sql/core`. |\n| `TestHiveSingleton` | `sql/hive` | Mixed in alongside `QueryTest`, e.g. `class X extends QueryTest with TestHiveSingleton`. Used by tests under `sql/hive`. |\n\n## Python Test Base Classes\n\nPySpark tests use the stdlib `unittest` framework: every suite subclasses `unittest.TestCase` (run via `python/run-tests`, see below). As with Scala, pick the lowest base that provides what the test actually needs. The bases live under `python/pyspark/testing/` and each adds capability on top of the previous:\n\n    unittest.TestCase                                     (stdlib)\n      <- PySparkBaseTestCase                              (pyspark.testing.utils)\n        <- ReusedPySparkTestCase                          (pyspark.testing.utils)\n          <- ReusedSQLTestCase                            (pyspark.testing.sqlutils)\n            <- PandasOnSparkTestCase                      (pyspark.testing.pandasutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plain Python — no Spark | `unittest.TestCase` | Use the stdlib base directly. `PySparkBaseTestCase` is the same thing plus a SIGTERM fault-handler dump enabled when `PYSPARK_TEST_TIMEOUT` is set; subclass it only when you want that. |\n| RDD / `SparkContext` — no `SparkSession` | `PySparkTestCase` (fresh) or `ReusedPySparkTestCase` (shared) | Both create a `SparkContext(\"local[4]\")`. `PySparkTestCase` makes a new one per test (isolation); `ReusedPySparkTestCase` shares one per class (faster, the usual choice) and adds `quiet()` and an overridable `conf()` / `master()`. |\n| SQL / DataFrame — needs a `SparkSession` | `ReusedSQLTestCase` | The workhorse for classic tests under `python/pyspark/sql`. Adds a shared `cls.spark`, sample `cls.df` / `cls.testData`, and mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. Classic (non-Connect) mode. |\n| pandas API on Spark | `PandasOnSparkTestCase` | Extends `ReusedSQLTestCase` with Arrow enabled and pandas-on-Spark assertions (`PandasOnSparkTestUtils`); `ComparisonTestBase` builds on it. |\n\n### Spark Connect test bases\n\nSpark Connect suites live in `pyspark.testing.connectutils` and are auto-skipped (via `should_test_connect`) when Connect dependencies are missing:\n\n    PySparkBaseTestCase                                   (pyspark.testing.utils)\n      <- PlanOnlyTestFixture                              (pyspark.testing.connectutils)\n      <- ReusedConnectTestCase                            (pyspark.testing.connectutils)\n           <- ReusedMixedTestCase                         (pyspark.testing.connectutils)\n\n| Test scope | Base | Notes |\n|------------|------|-------|\n| Plan / proto construction — no server | `PlanOnlyTestFixture` | Uses a `MockRemoteSession`; builds and inspects plans without a running Connect server. For proto / plan-shape assertions. |\n| Connect DataFrame — real session | `ReusedConnectTestCase` | The Connect analog of `ReusedSQLTestCase`; starts a session via `.remote(...)` (honoring `SPARK_CONNECT_TESTING_REMOTE`, default `local[4]`). Mixes in `SQLTestUtils` + `PySparkErrorTestUtils`. |\n| Classic + Connect side by side | `ReusedMixedTestCase` | Extends `ReusedConnectTestCase`. For directly comparing classic vs Connect: it exposes a classic `self.spark` and a Connect `self.connect` so a test can run the same operation on each and assert they agree (`compare_by_show`, `both_conf`). Requires JVM access. |\n\n### Mixins and helpers\n\nThese are combined with a base above rather than used on their own:\n\n- `SQLTestUtils` — context managers `sql_conf`, `table`, `temp_view`, `view`, `database`, `function`, `temp_func`, `temp_env`; assumes `self.spark`. Already mixed into `ReusedSQLTestCase` and `ReusedConnectTestCase`.\n- `PySparkErrorTestUtils` — `check_error(...)` to assert on a `PySparkException`'s error class and message parameters. The Python counterpart of Scala's `checkError`.\n- `assertDataFrameEqual` / `assertSchemaEqual` (public API, from `pyspark.testing`) — standalone assertion functions that work in any test, no particular base class required.\n- Domain bases outside the main ladder: `SparkSessionTestCase` (`pyspark.testing.mlutils`, for `ml`), `MLlibTestCase` (`pyspark.testing.mllibutils`), and `PySparkStreamingTestCase` (`pyspark.testing.streamingutils`, DStreams).\n\n## Build and Test\n\nBuild and tests can take a long time, so prefer the narrowest relevant build or test command.\n\nFor build and test setup, including how to run tests and troubleshoot common\nlocal failures, see `docs/building-spark.md`.\n\nPrefer SBT over Maven for faster incremental compilation. Module names are defined in `project/SparkBuild.scala`.\n\nCompile a single module:\n\n    build/sbt <module>/compile\n\nCompile test code for a single module:\n\n    build/sbt <module>/Test/compile\n\nRun test suites by wildcard or full class name:\n\n    build/sbt '<module>/testOnly *MySuite'\n    build/sbt '<module>/testOnly org.apache.spark.sql.MySuite'\n\nRun test cases matching a substring:\n\n    build/sbt '<module>/testOnly *MySuite -- -z \"test name\"'\n\nRun test cases in an optional module:\n\n    build/sbt -P<maven-profiles> '<module>/testOnly *MySuite'\n\nFor faster iteration, keep SBT open in interactive mode:\n\n    build/sbt\n    > project <module>\n    > testOnly *MySuite\n\n### PySpark Tests\n\nPySpark tests require building Spark with Hive support first:\n\n    build/sbt -Phive package\n\nActivate the virtual environment specified by the user, or default to `.venv`:\n\n    source <venv>/bin/activate\n\nIf the default venv does not exist, create it:\n\n    python3 -m venv .venv\n    source .venv/bin/activate\n    pip install --upgrade pip\n    pip install --group dev\n\nRun a single test suite:\n\n    python/run-tests --testnames pyspark.sql.tests.arrow.test_arrow\n\nRun a single test case:\n\n    python/run-tests --testnames \"pyspark.sql.tests.test_catalog CatalogTests.test_current_database\"\n\n## Investigating PR CI Failures\n\nEnumerate all failing check runs first, then drill into each by type. Do not assume a single failure: a PR can fail tests, linters, and the build at once, and these surface through different channels.\n\nStep 1 — Get the fork owner and the latest commit SHA of the PR:\n\n    gh api repos/apache/spark/pulls/<PR_NUMBER> --jq '{owner: .head.repo.owner.login, sha: .head.sha}'\n\nStep 2 — List every failing check run on the fork's commit. This is the complete failure set:\n\n    gh api repos/<OWNER>/spark/commits/<SHA>/check-runs --paginate \\\n      --jq '.check_runs[] | select(.conclusion == \"failure\") | {name, id: .id}'\n\nA passing (or absent) \"Report test results\" does NOT mean CI is green. That check aggregates only test-case failures; linter, license, dependency, MiMa, compile, and doc-build failures are separate check runs that produce no test annotations. Always work from the list in Step 2, not from any single check.\n\nStep 3 — Drill into each failure according to its kind:\n\n- **Test jobs** (e.g. \"Report test results\", \"Build modules: ...\"): fetch failure annotations. Each annotation contains the test class, test name, and failure message:\n\n      gh api repos/<OWNER>/spark/check-runs/<CHECK_RUN_ID>/annotations\n\n- **Non-test jobs** (e.g. \"Linters, licenses, and dependencies\", \"Build\"): find the failed step, then read only that job's log:\n\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID> \\\n        --jq '{name, steps: [.steps[] | select(.conclusion == \"failure\") | .name]}'\n      gh api repos/<OWNER>/spark/actions/jobs/<JOB_ID>/logs\n\nAvoid downloading the large per-shard *test* job logs — they are very large and slow; use the annotations for those. Lint, license, dependency, and build job logs are small and fine to read directly when a step fails.\n\n## Checking PR Merge Status\n\nSpark merges PRs with `dev/merge_spark_pr.py`, not the GitHub merge button, so a **merged PR shows up on GitHub as Closed, not Merged** — its `merged` / `mergedAt` are empty, and backports to maintenance branches are plain pushes. **Do not read a Closed PR as rejected**; most Closed PRs were in fact merged.\n\nTo check whether, and to which branches, a PR was merged, run `dev/pr_merge_status.py <pr-number>`:\n\n    $ dev/pr_merge_status.py 56356\n    PR #56356 (base master): [SPARK-57295][SQL] Make database location validation ...\n    merged: yes\n      master       9357bc9ae05\n      branch-4.x   72edddb358e\n\nIt lists `master` and the latest major's release branches the commit reached (e.g. `branch-4.x`, `branch-4.2`) — or reports `open` or `closed without merging`. It needs `gh` authenticated and a non-shallow checkout, and exits non-zero with a clear message if the PR is unknown (404) or the environment isn't ready (no `gh`/auth, no apache/spark remote). Older majors' branches are omitted; pass `--all-branches` to list every branch the commit reached.\n\n## Pull Request Workflow\n\nPR title format is `[SPARK-xxxx][COMPONENT] Title`. Draft, WIP, MINOR, and TRIVIAL PRs may omit the JIRA ID. The component tag comes from the `COMPONENTS` registry in `dev/merge_spark_pr.py`, the source of truth: each entry pairs a PR-title tag with its canonical JIRA component name and accepted aliases (e.g. `Documentation` → `[DOC]`, `Pandas API on Spark` → `[PS]`, `SQL` → `[SQL]`). Read that registry to pick the right tag.\n\nUse `[FOLLOWUP]` only for small PRs that directly modify or correct unreleased earlier PRs. For separately planned work, non-trivial changes, or work outside the earlier JIRA's scope, create a separate JIRA ticket for each PR and use the normal title format without `[FOLLOWUP]`.\n\nInfer the PR title from the changes. If no ticket ID is given and the PR is not draft, WIP, MINOR, or TRIVIAL, create one using `dev/create_spark_jira.py`, using the PR title (without the JIRA ID and component tag) as the ticket title.\n\n    python3 dev/create_spark_jira.py \"<title>\" -c <component> { -t <type> | -p <parent-jira-id> }\n\n- **Component** (`-c`): the exact JIRA component name (not the PR title shorthand), e.g. \"SQL\", \"Spark Core\", \"PySpark\", \"Connect\". Run `python3 dev/create_spark_jira.py --list-components` for the full list.\n- **Issue type** (`-t`): \"Bug\", \"Improvement\", \"New Feature\", \"Test\", \"Documentation\", or \"Dependency upgrade\".\n- **Parent** (`-p`): if the user mentions a parent JIRA ticket (e.g., \"this is a subtask of SPARK-12345\"), pass it instead of `-t`. The issue type is automatically \"Sub-task\".\n\nThe script sets the latest unreleased version as the default affected version.\n\nAfter creating a JIRA ticket, print a prominent notice so the user does not miss it:\n\n    ============================================================\n    JIRA ticket created: SPARK-XXXXX\n    https://issues.apache.org/jira/browse/SPARK-XXXXX\n\n    Title:              <title>\n    Component(s):       <component>\n    Issue type:         <type>\n    Affected version(s): <version>\n    Priority:           <priority>\n\n    Please review and adjust these fields if needed.\n    ============================================================\n\nBefore writing the PR description, read `.github/PULL_REQUEST_TEMPLATE` and fill in every section from that file.\n\nDO NOT push to the upstream repo. Always push to the personal fork. Open PRs against `master` on the upstream repo.\n\nDO NOT force push or use `--amend` on pushed commits unless the user explicitly asks. If the remote branch has new commits, fetch and rebase before pushing.\n\nAlways get user approval before external operations such as pushing commits, creating PRs, or posting comments. Use `gh pr create` to open PRs. If `gh` is not installed, generate the GitHub PR URL for the user and recommend installing the GitHub CLI.\n\n## Project Instructions Discovery\n\nWhen exploring or working in any directory, always check for nested `AGENTS.md` files in that\ndirectory and its ancestors. Read and follow every applicable file; instructions in a more specific\ndirectory take precedence for that directory's scope.\n\nA directory may carry these instructions as `CLAUDE.md` rather than `AGENTS.md`, so check **both** names — a directory with only a `CLAUDE.md` has project instructions you must read too. Where both files exist, one is typically the real file and the other is a symlink to it, although either name may be the real file. Reading either one is sufficient. When adding instructions to a directory that has neither file, create AGENTS.md as the real file and add a CLAUDE.md symlink beside it (ln -s AGENTS.md CLAUDE.md). This keeps a single source of truth accessible under both names.\n\n## Versioning and Branch Policy\n\nWhen a change needs a version — `@since` annotations, config `.version(\"...\")` (`SQLConf` / `*Conf`), new `MimaExcludes` sections, etc. — use the version of the branch it first ships in, with `-SNAPSHOT` stripped. Determine that branch:\n\n- **PR opened against a non-`master` base branch** (e.g. a maintenance line like `branch-4.2`): use that base branch's version -- when you're checked out on it, that's just the working tree's `pom.xml`. The helper below covers only the common `master`-base case.\n- **PR opened against `master`:** most PRs merge to **both** `master` and the latest `branch-<N>.x` (the branch for the next feature release, e.g. `branch-4.x`), so use the `branch-<N>.x` version. The exception is **master-only** changes — use `master`'s version — which are only:\n  - breaking / binary-incompatible changes that can't ship in a minor release;\n  - dependency upgrades that don't fix a critical issue worth backporting.\n\nDo **not** just read `master`'s version: a normally-backported PR ships first in `branch-<N>.x`, whose version is lower than `master`'s. If unsure whether a change is master-only, ask the user.\n\n`dev/next_version_candidates.py` (no arguments) prints both candidate versions, reading from the local `apache/spark` remote (the `upstream` configured during pre-flight). It reports the mechanical facts only -- choosing between them per the rules above is the judgement call (the numbers below are illustrative and advance over time):\n\n    $ dev/next_version_candidates.py\n    master       5.0.0\n    branch-4.x   4.3.0\n\n## Security\n\nSecurity model: [SECURITY.md](./SECURITY.md)\n\nAgents that scan this repository should consult `SECURITY.md` for the project's threat model, in-scope / out-of-scope declarations, and known non-findings before reporting issues.\n\n<!-- File: .claude-plugin/marketplace.json (Tokens: ~300 | Category: marketplace) -->\n# Claude Plugin Marketplace Catalog\n\nPath: `.claude-plugin/marketplace.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/marketplace.json)\n\n<!-- File: .claude-plugin/plugin.json (Tokens: ~300 | Category: plugin-manifest) -->\n# claude-plugin Plugin Manifest\n\nPath: `.claude-plugin/plugin.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.claude-plugin/plugin.json)\n\n<!-- File: .cursor/mcp.json (Tokens: ~300 | Category: mcp-config) -->\n# Model Context Protocol (MCP) Configuration\n\nPath: `.cursor/mcp.json`\n\n[View Raw Content on GitHub](https://raw.githubusercontent.com/apache/spark/HEAD/.cursor/mcp.json)\n\n<!-- File: .cursorrules (pyspark-etl-best-practices-cursorrules-prompt-file.mdc) (Tokens: ~3262 | Category: cursor-rule) -->\n---\ndescription: \"Cursor rules for PySpark ETL development with code style, joins, window functions, map operations, and Iceberg patterns.\"\nglobs: **/*\nalwaysApply: false\n---\nYou are an expert in PySpark, Spark SQL, Apache Iceberg, and production data engineering. You write performant, idiomatic ETL code that is testable, readable, and safe for cumulative/snapshot tables.\n\nFollow these rules when generating or reviewing PySpark code.\n\n# PySpark ETL Best Practices\n\n## 1. Project Structure\n\n### ETL class scaffold\n\nCreate a base class that manages the SparkSession lifecycle. Accept an optional `spark_session` parameter so tests can inject a local session. Use an abstract method for the job logic.\n\n```python\nfrom abc import ABC, abstractmethod\nfrom pyspark.sql import SparkSession\n\nclass BaseETL(ABC):\n    def __init__(self, config, app_name=\"ETL Job\", spark_session=None):\n        self.spark = spark_session or SparkSession.builder.appName(app_name).getOrCreate()\n        self.config = config\n        self.logger = logging.getLogger(self.__class__.__name__)\n\n    @abstractmethod\n    def run_job(self): ...\n\n    def stop(self):\n        self.spark.stop()\n```\n\n### Config — use a factory function\n\nKeep the dataclass as pure data and put CLI parsing in a standalone factory function. This makes configs easy to construct in tests without touching `sys.argv`.\n\n```python\n@dataclass\nclass MyConfig:\n    read_date: int = 20200101\n\ndef create_config() -> MyConfig:\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--read_date\", type=int, default=20200101)\n    args = parser.parse_args()\n    return MyConfig(read_date=args.read_date)\n```\n\n### Pipeline composition with `.transform()`\n\nKeep `run_job` as orchestration. Each step is a named method.\n\n```python\nevents = self.read_source().transform(self.enrich).transform(self.merge_with_existing)\n```\n\n### Use a shared reader for partition-aware reads\n\nBuild a generic reader utility that handles partition mechanics (date filters, hour ranges, latest-partition lookups). Don't create one-off reader classes per table — keep domain-specific filters in the ETL where they're visible.\n\n```python\nclass PartitionedReader:\n    @staticmethod\n    def read_latest(spark, table_name, partition_col):\n        row = spark.read.table(table_name).agg(F.max(partition_col)).first()\n        if row is None or row[0] is None:\n            return spark.createDataFrame([], spark.read.table(table_name).schema)\n        return spark.read.table(table_name).filter(F.col(partition_col) == row[0])\n\n    @staticmethod\n    def read_by_date(spark, table_name, partition_col, date_value):\n        return spark.read.table(table_name).filter(F.col(partition_col) == date_value)\n\n# Reader handles partitioning\nevents = PartitionedReader.read_by_date(spark, \"catalog.my_table\", \"event_date\", 20260319)\n\n# Business filters stay in the ETL\nevents = events.filter(F.col(\"event_type\").isin(\"login\", \"purchase\"))\n```\n\n### Shared merge utilities\n\nFor simple outer-join-with-coalesce merges, build a reusable merge function that handles aliasing, join key coalescing, and per-column defaults. Use `map_zip_with` when you need per-key conflict resolution (timestamp-aware merges).\n\n## 2. Code Style\n\n### Use `F.col()` — always use the `F.` prefix\n\nImport functions as `import pyspark.sql.functions as F` and use `F.col()`, `F.when()`, `F.lit()`, etc. throughout. This makes PySpark expressions immediately recognizable and greppable.\n\nAvoid `df.colA` attribute access — it binds the column to a specific DataFrame variable, which breaks after joins or when the variable is reassigned. Use `F.col()` with `.alias()` on the DataFrame if disambiguation is needed.\n\n```python\n# BAD — binds column to a specific DataFrame variable, breaks after joins\ndf.select(F.lower(df1.colA), F.upper(df2.colB))\n\n# GOOD\ndf.select(F.lower(F.col('colA')), F.upper(F.col('colB')))\n```\n\n### Extract complex conditions into named variables\n\nLimit logic inside `.filter()` or `F.when()` to 3 expressions. Extract the rest.\n\n```python\n# BAD — redundant logic hidden in nested parentheses\nF.when((F.col('status') == 'Delivered') | (((F.datediff('date_a', 'date_b') < 0) & ...)), 'Active')\n\n# GOOD\nis_delivered = (F.col('status') == 'Delivered')\ndate_passed = (F.datediff(F.col('date_a'), F.col('date_b')) < 0)\nhas_registration = (F.col('registration').rlike('.+'))\nF.when(is_delivered | (date_passed & has_registration), 'Active')\n```\n\n### Prefer `select` over `withColumn` chains\n\n`select` specifies the output schema in one pass. `withColumn` chains create intermediate DataFrames and can degrade performance — each call triggers a new projection in the query plan.\n\n```python\n# BAD — 3 intermediate DataFrames\ndf = df.withColumn(\"a\", F.col(\"a\").cast(\"double\"))\ndf = df.withColumn(\"b\", F.upper(F.col(\"b\")))\ndf = df.withColumn(\"c\", F.lit(1))\n\n# GOOD — 1 DataFrame, explicit schema contract\ndf = df.select(\n    F.col(\"a\").cast(\"double\"),\n    F.upper(F.col(\"b\")).alias(\"b\"),\n    F.lit(1).alias(\"c\"),\n)\n```\n\n### Use `alias` over `withColumnRenamed`\n\n```python\n# BAD\ndf.select('key', 'comments').withColumnRenamed('comments', 'num_comments')\n\n# GOOD\ndf.select('key', F.col('comments').alias('num_comments'))\n```\n\n### Chaining limits\n\nMax 5 statements per chain. Separate by operation type (select/filter vs withColumn vs join).\n\n```python\n# BAD — mixed concerns in one chain\ndf = (df.select('a', 'b', 'key')\n    .filter(F.col('a') == 'x')\n    .withColumn('ratio', F.col('a') / F.col('b'))\n    .join(df2, 'key', how='inner')\n    .drop('b'))\n\n# GOOD — separated by concern\ndf = df.select('a', 'b', 'key').filter(F.col('a') == 'x')\ndf = df.withColumn('ratio', F.col('a') / F.col('b'))\ndf = df.join(df2, 'key', how='inner').drop('b')\n```\n\n## 3. Joins\n\n### Always specify `how=` explicitly\n\n```python\n# BAD\ndf.join(other, 'key')\n\n# GOOD\ndf.join(other, 'key', how='inner')\n```\n\n### Prefer left joins over right joins\n\nFlip the DataFrame order and use `left` instead of `right` for readability — the primary DataFrame stays on the left.\n\n```python\nflights = flights.join(aircraft, 'aircraft_id', how='left')\n```\n\n### Use `.alias()` for disambiguation after joins\n\n```python\n# BAD — renaming every column before join\nfor c in columns:\n    flights = flights.withColumnRenamed(c, 'flights_' + c)\n\n# GOOD — alias the whole DataFrame\nflights = flights.alias('f')\nparking = parking.alias('p')\nresult = flights.join(parking, 'code', how='left').select(\n    F.col('f.start_time').alias('flight_start'),\n    F.col('p.total_time').alias('parking_total'),\n)\n```\n\n### Broadcast small dimension tables\n\nWhen joining a large fact DataFrame with a small lookup/dimension table, wrap the small side in `F.broadcast()` to skip the shuffle on the small side.\n\nUse broadcast for tables that are small enough to fit in executor memory — typically dimension/lookup tables (category lookups, country codes, config mappings). Spark auto-broadcasts tables under 10MB by default (`spark.sql.autoBroadcastJoinThreshold`), but an explicit hint is useful when Spark can't infer the size (e.g., after filters or transformations).\n\nTo check if a table is broadcast-worthy during development:\n- **Spark UI**: after a run, check the SQL tab — scan sizes are shown per table\n- **Quick row count**: `spark.read.table(\"catalog.my_dim\").count()` (dev only, not in production code)\n- **Query plan**: `df.explain()` — Spark shows `BroadcastHashJoin` if it auto-broadcasts, `SortMergeJoin` if it doesn't\n\n```python\ndf.join(F.broadcast(category_dim), 'category_id', how='left')\n```\n\n### Never use `.dropDuplicates()` as a crutch\n\nIf duplicate rows appear, find the root cause. `.dropDuplicates()` masks the problem and adds shuffle overhead.\n\n## 4. Window Functions\n\nUse `from pyspark.sql import Window as W` alongside `import pyspark.sql.functions as F`.\n\n### Always specify an explicit frame\n\nWithout a frame, Spark picks one that changes depending on whether `orderBy` is present.\n\n```python\n# BAD — F.sum gives running sum with orderBy, total without. Surprising.\nw = W.partitionBy('key').orderBy('num')\n\n# GOOD — explicit about what you want\nw = W.partitionBy('key').orderBy('num').rowsBetween(W.unboundedPreceding, W.unboundedFollowing)\n```\n\n### `row_number` + filter vs `first` — know the difference\n\n- `row_number` + filter = **drop rows**, keep the best one\n- `first` over window = **overwrite a column value**, keep all rows\n\n### Use `ignorenulls=True` with `first` and `last`\n\nWithout it, a null in the first row gives null for the entire partition.\n\n```python\n# BAD — returns None if first row is null\nF.first('version').over(window)\n\n# GOOD\nF.first('version', ignorenulls=True).over(window)\n```\n\n### Avoid empty `partitionBy()`\n\nIt forces all data into one partition. Use `.agg()` instead for global aggregations.\n\n```python\n# BAD — single partition, kills performance\nw = W.partitionBy()\ndf.select(F.sum('num').over(w))\n\n# GOOD\ndf.agg(F.sum('num').alias('total'))\n```\n\n## 5. Map & Array Higher-Order Functions\n\n### Use `map_zip_with` when merging maps with complex logic\n\n`map_concat` is fine for simple merges with no key overlap. When you need custom logic per key (e.g., keep the newer timestamp, pick the higher value), use `map_zip_with` — it gives you a per-key merge function instead of blindly letting one side win.\n\n```python\n# BAD — no control over conflict resolution\nmap_concat(existing_map, new_map)\n\n# GOOD — keep the entry with the later timestamp\nmap_zip_with(new_map, existing_map,\n    lambda key, v1, v2: (\n        F.when(v1.isNull(), v2)\n        .when(v2.isNull(), v1)\n        .otherwise(F.when(v1.event_ts >= v2.event_ts, v1).otherwise(v2))\n    )\n)\n```\n\n### Use `transform` + `array_max` to extract from nested structs\n\n```python\n# Extract max event_ts from a map of structs\narray_max(transform(map_values(my_map), lambda x: x.event_ts))\n```\n\n### Avoid UDFs — use native Spark functions first\n\nUDFs break Catalyst optimization and add serialization overhead. Before writing one, check if a built-in Spark function or higher-order function can do the job.\n\n## 6. Cumulative / Snapshot Table Patterns\n\n### Merges must be idempotent\n\nRe-running with the same data should produce the same result, not create duplicates.\n\n### Merges must be order-independent\n\nBackfilling old data should not overwrite newer data. Use an explicit ordering criterion (e.g., event timestamp, version number, partition date) to resolve conflicts — don't rely on positional precedence like `coalesce` argument order.\n\n### Validate primary key uniqueness after writes\n\nAdd audit steps that validate primary key uniqueness and check for nulls in key columns.\n\n## 7. Data Quality & Performance\n\n### Use `F.lit(None)` for empty columns, never empty strings\n\n```python\n# BAD\ndf = df.withColumn('foo', F.lit(''))\ndf = df.withColumn('foo', F.lit('NA'))\n\n# GOOD\ndf = df.withColumn('foo', F.lit(None))\n```\n\n### Avoid `.otherwise()` as a general catch-all\n\nUnknown values silently collapse into the otherwise bucket, hiding data quality issues.\n\n```python\n# BAD — a new platform_type you didn't anticipate becomes \"Other\" silently\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n .otherwise('Other')\n\n# GOOD — unmapped values stay null, surfacing gaps in your logic\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n```\n\n### No `.show()`, `.collect()`, `.printSchema()` in production\n\nThese trigger full materialization or add unnecessary driver overhead. Use them only for local debugging, never in deployed ETL code. `.count()` is acceptable when used intentionally (e.g., logging row counts for monitoring, forcing materialization before a DAG fork).\n\n### Use `persist()` intentionally\n\nOnly persist a DataFrame when it's referenced in multiple subsequent actions — otherwise the write action will materialize it for you. `.persist()` + `.count()` is a common pattern to force materialization and log row counts for debugging; use it when needed but be aware it adds a full scan.\n\nChoose the storage level based on your use case:\n- `MEMORY_AND_DISK` — safe default, spills to disk if memory is tight\n- `MEMORY_ONLY` — faster but risks recomputation if evicted\n- `DISK_ONLY` — for very large DataFrames that don't fit in memory\n\n## 8. Iceberg Write Patterns\n\n### Use `.byName()` for schema evolution safety\n\nColumn ordering doesn't matter — Spark matches by name, not position.\n\n```python\ndf.write.byName().mode(\"overwrite\").insertInto(\"catalog.my_table\")\n```\n\n### Use `__partitions` metadata table for latest partition reads\n\nIceberg exposes a `__partitions` metadata table. Use it to find the latest snapshot instead of scanning the full table.\n\n```python\npartition_df = spark.read.table(\"catalog.my_table__partitions\").select(\n    \"partition.partition_date\", \"partition.partition_hour\"\n)\nmax_partition = partition_df.orderBy(\n    F.col(\"partition_date\").desc(), F.col(\"partition_hour\").desc()\n).first()\nif max_partition is None:\n    raise ValueError(\"No partitions found in catalog.my_table\")\nlatest_date = max_partition[\"partition_date\"]\n```\n\n### Understand `write.distribution-mode`\n\n- `\"none\"` — no re-shuffle before writing. Fastest, but output file sizes depend on upstream partitioning.\n- `\"hash\"` — redistributes data by partition key. Produces evenly sized files but adds a shuffle.\n- `\"range\"` — sorts data by partition key before writing. Good for ordered scan performance but most expensive.\n</yakaai_skills>"}