[DRAFT][SPARK-51705][CONNECT][PYTHON] Support SparkSession.broadcast() (broadcast variables over Spark Connect) - #1
Open
Tagar wants to merge 11 commits into
Open
[DRAFT][SPARK-51705][CONNECT][PYTHON] Support SparkSession.broadcast() (broadcast variables over Spark Connect)#1Tagar wants to merge 11 commits into
Tagar wants to merge 11 commits into
Conversation
Tagar
marked this pull request as ready for review
July 20, 2026 23:00
Tagar
force-pushed
the
broadcast-connect-python-v1
branch
from
July 20, 2026 23:10
62b1847 to
dc11a44
Compare
Tagar
marked this pull request as draft
July 20, 2026 23:10
Tagar
force-pushed
the
broadcast-connect-python-v1
branch
5 times, most recently
from
July 23, 2026 18:22
bc20d9f to
f4b6181
Compare
Tagar
force-pushed
the
broadcast-connect-python-v1
branch
3 times, most recently
from
July 24, 2026 20:32
352504e to
bb7358d
Compare
Tagar
marked this pull request as ready for review
July 27, 2026 06:19
Tagar
force-pushed
the
broadcast-connect-python-v1
branch
from
July 27, 2026 06:22
bb7358d to
3248e82
Compare
### What changes were proposed in this pull request? Fixes the grammar of the null-metadata `require(...)` message in `HDFSMetadataLog` and `AsyncOffsetSeqLog`: "'null' metadata cannot written to a metadata log" becomes "... cannot be written ...". ### Why are the changes needed? The message was missing the word "be". The sibling `AsyncCommitLog` already uses the correct phrasing, so this brings the two outliers in line. The message is a plain `require()` string, not part of the error-condition framework, so it is not golden-tested. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Message-text-only change; no behavior change. No tests needed. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Closes apache#57546 from uros-b/error-metadatalog-cannot-be-written. Lead-authored-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com> Co-authored-by: Uros Bojanic <uros.bojanic@databricks.com> Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com>
…deterministic ### What changes were proposed in this pull request? `TriggeredGraphExecution.getRunTerminationReason` decided which failed flow's reason to report by calling `collectFirst` over `failureTracker`, a `ConcurrentHashMap` whose iteration order is unspecified. When more than one flow exhausts its retries (a non-retryable `StopFlowExecution`), the flow whose reason gets surfaced therefore varied from run to run. This extracts a small pure helper, `chooseRunTerminationReason`, that considers only the stopped flows and picks the earliest one by `(lastFailTimestamp, flowName)`, so the reported reason is stable across otherwise-identical runs. `getRunTerminationReason` now calls it and falls back to `UnexpectedRunFailure()`. The previous code also computed `graphForExecution.flow(...)` and `lastException` only to discard them; those are dropped. ### Why are the changes needed? Two runs that fail the same way could report different termination reasons (different flow name and cause), which is confusing in logs and events and makes the outcome non-reproducible. ### Does this PR introduce _any_ user-facing change? No. The reported reason was already one of the failing flows; it is now chosen deterministically. ### How was this patch tested? Added unit tests for `chooseRunTerminationReason` in `TriggeredGraphExecutionSuite`: the earliest failure wins regardless of iteration order, ties are broken by flow name, and retryable failures are ignored. They fail against the previous order-dependent selection and pass with this change. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 4.8) Closes apache#56698 from LuciferYang/sdp-deterministic-termination-reason. Authored-by: YangJie <yangjie01@baidu.com> Signed-off-by: yangjie01 <yangjie01@baidu.com>
…esolution ### What changes were proposed in this pull request? `DataflowGraphTransformer.transformDownNodes` resolves flows on a bounded thread pool and drives them from a `while` loop that, each pass, partitioned the in-flight futures with the non-blocking `future.isDone`, reaped the completed ones, and scheduled a new flow if a slot was free. When all slots were in flight (or the queue was drained and only the last futures remained) and none had completed, the pass reaped nothing and scheduled nothing, then looped again immediately - busy-spinning on `isDone` and pinning a core for the duration of resolution. This drives the loop with an `ExecutorCompletionService` instead: completed tasks are drained with the non-blocking `poll()`, and when nothing can be scheduled but tasks are still running, the loop blocks on `take()` until the next one finishes rather than spinning. Behavior is otherwise unchanged - the same flows are scheduled in the same order, exceptions are still propagated via `Future.get()`, and an `outstanding` counter replaces the `ArrayBuffer[Future]` for slot bookkeeping. ### Why are the changes needed? Resolving a graph with more flows than the parallelism (10) kept one CPU core busy at 100% doing no useful work for the whole resolution, which is wasteful and shows up as unexplained driver CPU. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Two new cases in `ConnectValidPipelineSuite` cover the regime this PR changes - more flows than `parallelism` (10), so the slots fill and the loop reaches the blocking `take()` branch that replaces the busy-wait. The small graphs in the existing suites never get there. - `resolution terminates and resolves all flows when flow count exceeds parallelism` - 25 independent flows. - `resolution re-queues retryable flows under load when consumers exceed parallelism` - 20 consumers registered before their source `src`, so the first batch throws `TransformNodeRetryableException`, parks as dependents of `src`, and is re-queued once `src` resolves; this exercises the retryable re-queue path together with the blocking branch. Both assert only the outcome (every flow resolves and the call returns), so they are deterministic and have no timing dependence - a regression that deadlocked would hang until the suite times out. Asserting the absence of a busy-wait directly is not included, since that requires CPU-time or timing measurements that are flaky in CI. Existing graph-resolution suites (`ConnectValidPipelineSuite`, `ConnectInvalidPipelineSuite`, `SqlPipelineSuite`, `TriggeredGraphExecutionSuite`, `MaterializeTablesSuite`) still pass; the change only affects how the loop waits, not what it resolves. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 4.8) Closes apache#56700 from LuciferYang/sdp-resolution-busy-wait. Authored-by: YangJie <yangjie01@baidu.com> Signed-off-by: yangjie01 <yangjie01@baidu.com>
…_TEMP_2214-2219
### What changes were proposed in this pull request?
The catalog-plugin loading failures in `Catalogs.load` were reported with six placeholder error conditions `_LEGACY_ERROR_TEMP_2214` through `_LEGACY_ERROR_TEMP_2219`, none of which carried a SQLSTATE.
Group the six into a single umbrella error condition `CANNOT_LOAD_CATALOG` (SQLSTATE `46103`), with one subclass per failure mode in the `Catalogs.load` try/catch:
- `NOT_A_CATALOG_PLUGIN` — the class does not implement `CatalogPlugin`
- `PLUGIN_CLASS_NOT_FOUND` — `ClassNotFoundException`
- `CONSTRUCTOR_NOT_FOUND` — `NoSuchMethodException`
- `CONSTRUCTOR_NOT_ACCESSIBLE` — `IllegalAccessException`
- `ABSTRACT_CLASS` — `InstantiationException`
- `CONSTRUCTOR_FAILURE` — `InvocationTargetException`
This also drops a stray trailing `)` in the legacy `_2216`/`_2217` messages.
### Why are the changes needed?
The error-conditions README disallows new `_LEGACY_ERROR_TEMP_*` entries and asks existing ones to be resolved. This resolves six of them. SQLSTATE `46103` ("Java Error") is consistent with the sibling `CANNOT_LOAD_FUNCTION_CLASS`; the failures are triggered by a user-supplied `spark.sql.catalog.<name>` plugin class, so they are user-actionable rather than system errors.
### Does this PR introduce _any_ user-facing change?
No. The `_LEGACY_ERROR_TEMP_*` names are not part of the public API. Message text is preserved (except the stray `)` typo removal), now rendered under the umbrella prefix.
### How was this patch tested?
Updated the existing `checkError`/condition assertions in `CatalogLoadingSuite` and `SupportsCatalogOptionsSuite`, and added tests for the previously-uncovered `CONSTRUCTOR_NOT_FOUND` and `ABSTRACT_CLASS` subclasses. `build/sbt "catalyst/testOnly *CatalogLoadingSuite" "core/testOnly org.apache.spark.SparkThrowableSuite"` passes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes apache#57556 from LuciferYang/assign-name-legacy-2214-2219.
Authored-by: YangJie <yangjie01@baidu.com>
Signed-off-by: yangjie01 <yangjie01@baidu.com>
### What changes were proposed in this pull request? This PR upgrades the `json` gem from 2.12.2 to 2.21.1 in `docs/Gemfile.lock`. It is a transitive dependency (pulled in by `jekyll`, which requires `json (~> 2.6)`), and 2.21.1 satisfies that constraint, so only the locked spec version changes. `json` has no runtime dependencies, so no other lock entries change and `docs/Gemfile` does not need to be touched. ### Why are the changes needed? 2.21.1 includes the fix for a security advisory that affects all versions in `>= 2.9.0, < 2.19.9`: - [GHSA-x2f5-4prf-w687](GHSA-x2f5-4prf-w687) / CVE-2026-54696 (low): heap out-of-bounds write in the JSON generator when streaming to an IO. On the IO path, `fbuffer_do_inc_capa()` compared the requested size against the buffer's total capacity instead of its remaining capacity, so `JSON.dump(obj, io)` or `JSON::State#generate(obj, io)` could write past the buffer when serializing a string near 16 KB. Fixed in 2.19.9. Following the same pattern as SPARK-57633 (`concurrent-ruby` 1.3.7), this picks up the latest release rather than the minimum patched version. ### Does this PR introduce _any_ user-facing change? No. This only affects the documentation build toolchain. ### How was this patch tested? Manually verified in a scratch copy of `docs/Gemfile` and `docs/Gemfile.lock`: 1. Lock resolution matches. Running `bundle lock --update=json` produces a lock file byte-identical to the hand-edited one, confirming 2.21.1 satisfies jekyll's `json (~> 2.6)` and that no other locked spec is affected. 2. Frozen install succeeds. `BUNDLE_FROZEN=true bundle install` installs all 36 gems without modifying the lock file, and `bundle list` reports `json (2.21.1)`. 3. The advisory's PoC no longer crashes: ```ruby require "json" require "stringio" io = StringIO.new big = "a" * 16385 big[16382] = '"' # escapable byte near the buffer boundary JSON.dump([big], io) ``` ```text json version: 2.21.1 PoC (JSON.dump with IO): no crash, output bytes = 16390 roundtrip ok: true ``` 4. The docs site builds: ``` $ cd docs && SKIP_API=1 bundle exec jekyll build Configuration file: .../docs/_config.yml ************************ * Building error docs. * ************************ Generated: docs/_generated/error-conditions.html Source: .../docs Destination: .../docs/_site Incremental build: disabled. Enable with --incremental Generating... Warning: Tolerating missing API files because the following skip flags are set: SKIP_API done in 3.125 seconds. Auto-regeneration: disabled. Use --watch to enable. ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) Closes apache#57565 from LuciferYang/docs-json-gem-2.21.1. Authored-by: YangJie <yangjie01@baidu.com> Signed-off-by: yangjie01 <yangjie01@baidu.com>
### What changes were proposed in this pull request? This PR adds a new module `common/config` which introduces a framework to define configs in prototext files. When the Spark application starts, we load config definitions from all prototext files and put them in a map with config name as the key, and the proto binary of the config definition as the value. These proto-backed configs are also registered in `ConfigEntry.knownConfigs` via a wrapper `ProtoBackedConfigEntry`, together with existing Spark configs. The proto schema defines each config entry with: key, value type, default value, scope (cluster vs session), mutability (static vs dynamic, i.e. whether the config can be changed after system initialization), visibility (public vs internal), binding policy (session/persisted/not_applicable for SQL views, UDFs, and procedures), documentation, and version. Enum values follow proto3 naming conventions with type-name prefixes (e.g. `SCOPE_CLUSTER`, `VALUE_TYPE_BOOL`, `VISIBILITY_PUBLIC`, `BINDING_POLICY_SESSION`, `MUTABILITY_STATIC`) to avoid namespace collisions. The `BindingPolicy` field maps to the existing Scala `ConfigBindingPolicy` enum. The `Mutability` field is used by `SQLConf.isStaticConfigKey` to determine whether a config can be changed at runtime. The new framework will co-exist with the existing config framework, until we migrate all existing configs to this new framework. For simple configs that are only accessed in one place, we can hardcode the config name in the place that accesses it, with a new API `SQLConf#getConfByKeyStrict` to avoid typo. Example in this PR: `spark.sql.optimizer.datasourceV2ExprFolding` For configs that are accessed in multiple places and we want to avoid hardcoding config name, or configs that need custom validation, we can still have an entry in `object SQLConf` to reference the config definition. Examples in this PR: `spark.sql.optimizer.maxIterations` and `spark.sql.shuffledHashJoinFactor`. As part of the migration, the 4 example configs are removed from the binding-policy exceptions allowlist (`configs-without-binding-policy-exceptions`) and assigned `NOT_APPLICABLE`, aligning them with standard binding-policy handling. Since `NOT_APPLICABLE` configs do not affect view/UDF/procedure resolution results, no resolved plans change. ### Why are the changes needed? Defining configs in various Scala objects is a bad practice: - Configs are registered when the Scala objects are loaded. To list all configs we must know all these Scala objects and load them. - It's hard to audit configs as they spread all over the codebase. - We will hit JVM limitation one day sooner or later, as defining configs in a Scala object is basically doing heavy work in the constructor, which has limitation of 64 kb bytecode size. ### Does this PR introduce _any_ user-facing change? No, it's developer facing ### How was this patch tested? new tests ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code Closes apache#53488 from cloud-fan/conf. Authored-by: Wenchen Fan <wenchen@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
…w construction ### What changes were proposed in this pull request? This PR replaces apache#57490, which had to be abandoned due to severe merge conflicts. It is an exact cherry-pick of the previous commits, plus a fix for the test cases that failed due to the merge conflicts. An SCD2 AutoCDC flow can restrict which columns define a "run" via TRACK HISTORY ON (...), which populate ChangeArgs.trackHistorySelection. Until now, that selection was only resolved when the first microbatch ran, inside Scd2BatchProcessor.computeTrackedHistoryColumns during reconciliation. An unresolvable or ineligible tracking column — one that is absent from the source, is a key, is a reserved framework column, or was dropped by the flow's column_list — therefore surfaced mid-stream rather than at flow construction, unlike every other AutoCDC misconfiguration (keys, column selection, reserved names), which fail eagerly. This PR validates trackHistorySelection at AutoCdcMergeFlow construction time, mirroring the existing requireKeysPresentInSelectedSchema check: - The eligibility + resolution logic is extracted from Scd2BatchProcessor.computeTrackedHistoryColumns into a schema-based companion helper Scd2BatchProcessor.computeTrackedHistoryColumns(schema, changeArgs, caseSensitive). Both the per-microbatch runtime path and the new construction-time validator call it, so the two can never diverge. The refactor is behavior-preserving. - AutoCdcMergeFlow gains requireTrackHistoryColumnsResolvableInSelectedSchema, invoked when deriving the user-selected schema (right after the key-presence check). It runs before the flow's schema is forced, so the actionable error surfaces ahead of the temporary AUTOCDC_SCD2_NOT_SUPPORTED gate and remains correct once SCD2 support lands. - No new error condition: an unresolvable selection reuses the existing AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA (schema name trackHistorySelection). - The check is a no-op when trackHistorySelection is None, which covers all SCD1 flows (enforced by ChangeArgs) and SCD2 flows that do not restrict tracking. ### Why are the changes needed? Deferring this validation to reconciliation means a simple typo or misconfiguration (TRACK HISTORY ON (typo), or tracking a key/excluded column) is not reported at graph analysis time; it only fails once data flows, with an error raised deep in the SCD2 batch processor. Validating at flow construction gives a fail-fast, user-actionable error consistent with the rest of the AutoCDC configuration surface (keys, column_list, reserved names). ### Does this PR introduce any user-facing change? Yes. An AutoCDC SCD2 flow whose TRACK HISTORY ON (...) references a column that is not an eligible history-tracking column (absent, a key, a framework column, or excluded by column_list) now fails at flow construction with AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA instead of failing when the first microbatch runs. Valid selections are unaffected, and there is no change for SCD1 flows. (Note: SCD2 AutoCDC flows are not yet generally supported on master — still gated by AUTOCDC_SCD2_NOT_SUPPORTED — so no released behavior changes.) ### How was this patch tested? New unit tests in AutoCdcFlowSuite covering: an SCD2 flow tracking a non-existent column, a key column (ineligible), and a column dropped by columnSelection are each rejected at construction with AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA; a resolvable selection passes the check (falling through to the SCD2-not-supported gate); and case-sensitive/insensitive resolution behavior. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Closes apache#57524 from anew/spark-58313-track-history-v2. Authored-by: Andreas Neumann <andreas.neumann@databricks.com> Signed-off-by: Jose Torres <jtorres@apache.org>
### What changes were proposed in this pull request? Introduce a new optimizer rule `PullOutVariantExtractions` that hoists `variant_get` / `Cast(variant)` extractions out of three operator types that the existing `PushVariantIntoScan` / `V2ScanRelationPushDown` rules cannot see through: - **Aggregate function arguments** – e.g. `max(variant_get(v, '$.price', 'int'))`: the extraction is moved into a `Project` directly below the `Aggregate` and the aggregate references the resulting alias. The bare variant column is suppressed unless it is also needed raw (e.g. as a `GROUP BY` key), so no redundant full-variant slot is generated. - **Sort order keys** – e.g. `ORDER BY variant_get(v, '$.price', 'int')`: matched as `Project → Sort`; the extraction is hoisted below the `Sort` and the original `Project` is reproduced to prevent the alias from leaking into the output. - **Join conditions and projections above joins** – matched as `Project → Join`; extractions in both the join condition and the outer `Project` are routed to the owning join side. A `pushSideAliases` helper then pushes the aliases *through* any depth of chained joins so they land in a `Project` directly above the scan (where `PhysicalOperation` collapses them with the scan, making them visible to the pushdown). This is necessary because `PhysicalOperation` stops at a `Join` node. A `Sort` sitting over a `Join` is handled by fusing the two cases: the order-key aliases are pushed through the join tree, not left in a `Project` above it. The rule is gated by a new internal config `spark.sql.variant.pushVariantIntoScan.pullOutExtractions` (default `true`) and is a no-op unless `spark.sql.variant.pushVariantIntoScan` is also enabled. Non-variant plans are untouched. The rule is registered as the first rule in `SparkOptimizer.earlyScanPushDownRules`, before `SchemaPruning` and the V2 scan pushdown rules. ### Why are the changes needed? Before this change, a `variant_get` inside an aggregate function argument, sort key, or join condition caused the whole variant column to be read raw (or shredded with a redundant full-variant slot). For example: ```sql SELECT name, max(variant_get(v, '$.price', 'int')) FROM T GROUP BY name ``` read the entire v column even though only the price field was needed. After this change, Spark shreds only the requested typed fields, avoiding the full-variant I/O. The change improves query performance for variant referenced in aggregrate, join, sort. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Added new units. Also run correctness tests against some known workload. #### Performance result Test framework: https://github.com/cloudera-labs/variant-conformance-benchmark Dataset: tpc-ds dataset (SF=5), spark native parquet table, payload in variant or json. Run setup: pre-warm jvm, three runs, median query timing reported by spark. Run A with pullout rule enabled vs B with the rule disabled. ``` Compare: tpcds-flat-pullout-20260714 vs tpcds-flat-no-pullout-20260714 (metric: query_median) Run A: /Users/qlong/opensources/variant-conformance-benchmark/results/tpcds-flat-pullout-20260714/tpcds-flat/timings-variant.csv Run B: /Users/qlong/opensources/variant-conformance-benchmark/results/tpcds-flat-no-pullout-20260714/tpcds-flat/timings-variant.csv query median_A median_B delta_s delta_% ------ -------- -------- ------- ------- q07 1.40 3.28 -1.88 -57.2% q12 0.07 0.07 -0.00 -2.8% q19 0.06 0.07 -0.01 -9.9% q26 1.14 3.40 -2.26 -66.4% q42 0.56 2.59 -2.03 -78.2% q52 0.61 3.04 -2.43 -80.0% q55 0.54 2.68 -2.14 -79.9% q63 0.58 2.95 -2.37 -80.4% q68 1.00 3.42 -2.42 -70.8% q73 0.57 2.88 -2.31 -80.1% q79 0.85 3.26 -2.41 -73.9% q98 0.65 3.32 -2.67 -80.5% Summary: 12 queries, 12 comparable Geo-mean delta: -69.5% (Run A faster) Total (query_median): 8.0s vs 31.0s ``` Run A with pullout rule enabled vs B using json for payload, this test shows the **performance advantage of variant over json**. ``` Compare: tpcds-flat-pullout-20260714 vs tpcds-flat-pullout-20260714 (metric: query_median) Run A: /Users/qlong/opensources/variant-conformance-benchmark/results/tpcds-flat-pullout-20260714/tpcds-flat/timings-variant.csv Run B: /Users/qlong/opensources/variant-conformance-benchmark/results/tpcds-flat-pullout-20260714/tpcds-flat/timings-string_json.csv query median_A median_B delta_s delta_% ------ -------- -------- ------- ------- q07 1.40 3.26 -1.85 -56.9% q12 0.07 0.13 -0.06 -47.0% q19 0.06 0.23 -0.17 -72.2% q26 1.14 2.17 -1.03 -47.5% q42 0.56 0.56 +0.01 +1.1% q52 0.61 1.09 -0.48 -44.1% q55 0.54 0.47 +0.07 +15.2% q63 0.58 0.88 -0.30 -34.2% q68 1.00 1.34 -0.34 -25.1% q73 0.57 0.70 -0.13 -18.0% q79 0.85 1.62 -0.77 -47.4% q98 0.65 1.19 -0.54 -45.4% Summary: 12 queries, 12 comparable Geo-mean delta: -39.3% (Run A faster) Total (query_median): 8.0s vs 13.6s ``` ### Was this patch authored or co-authored using generative AI tooling? Co-authored with Claude Code Closes apache#57190 from qlong/variant-pushout-aggregate. Authored-by: Qiegang Long <qlong@users.noreply.github.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
…C flow construction ### What changes were proposed in this pull request? What changes were proposed in this pull request? This PR replaces apache#57488, which had to be abandoned due to severe semantic merge conflicts. It is an exact cherry-pick of the previous commits, plus a fix for the test cases that failed due to the merge conflicts. What were the merge conflicts? Two PRs disagreeing on the semantics when a source contains a reserved column but does not include that column in the column selection. This PR rejects that because it validates pre-column selection. The other PR (SPARK-58313) allowed that because it was doing the validation port column selection. I decided to fall back to the original behavior that already existed for SCD type 1 validate pre column selection. And we have SPARK-58325 to reconsider whether we want to do validation post column selection. AutoCdcMergeFlow validates at construction time that a flow's source change-data feed does not carry columns that collide with AutoCDC's internal columns. Until now, that validation (requireReservedPrefixAbsentInSourceColumns) only rejected column names starting with the reserved prefix __spark_autocdc_. SCD2, however, persists two framework columns to the target that do not carry that prefix — __START_AT and __END_AT (see Scd2BatchProcessor.reservedFrameworkColNames). A source column named __START_AT or __END_AT therefore slipped past the guard and would be silently overwritten during microbatch preprocessing. This PR closes that gap (the TODO(SPARK-57251) in Scd2BatchProcessor): - Adds requireReservedFrameworkColumnsAbsentInSourceColumns() to the AutoCdcMergeFlow constructor. For SCD2 flows it rejects any source column whose name collides (by exact name, resolver-aware so it respects spark.sql.caseSensitive) with a non-prefixed reserved framework column. SCD1 targets carry no such columns, so the check is a no-op for SCD1. - Runs the check before the flow's schema val is forced, so the actionable reserved-name error surfaces ahead of the temporary AUTOCDC_SCD2_NOT_SUPPORTED gate, and the check remains correct once SCD2 support lands. - Adds a new error condition AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT (SQLSTATE 42710), distinct from the existing prefix-based AUTOCDC_RESERVED_COLUMN_NAME_PREFIX_CONFLICT, since this collision is by exact name rather than by prefix. - Widens the visibility of Scd2BatchProcessor.reservedFrameworkColNames to private[pipelines] so the flow layer can validate against the single source of truth. ### Why are the changes needed? Without this check, a user whose CDC source happens to contain a __START_AT or __END_AT column would have that data silently overwritten by AutoCDC's SCD2 framework columns, with no error and no diagnostic — a data-correctness footgun. Failing fast at flow construction with a user-actionable error ("rename or remove the column") is the intended UX, consistent with the existing reserved-prefix guard. ### Does this PR introduce _any_ user-facing change? Yes. An AutoCDC SCD2 flow whose source change-data feed contains a column named __START_AT or __END_AT (subject to case-sensitivity settings) now fails at flow construction with AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT instead of silently overwriting the column. There is no change for SCD1 flows, and no change for SCD2 flows whose sources do not use these names. (Note: SCD2 AutoCDC flows are not yet generally supported on master — they are still gated by AUTOCDC_SCD2_NOT_SUPPORTED — so no released behavior changes.) ### How was this patch tested? New unit tests in AutoCdcFlowSuite covering: - an SCD2 flow with a __START_AT/__END_AT source column is rejected with AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT; - the reserved-name check fires before the AUTOCDC_SCD2_NOT_SUPPORTED gate; - an SCD1 flow with the same column name is allowed and the column survives into the flow schema; - case-sensitivity behavior (spark.sql.caseSensitive true/false); - a guard test asserting the set of non-prefixed reserved names is exactly {__START_AT, __END_AT}, so a future rename can't silently un-cover the validation. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Closes apache#57527 from anew/spark-57251-reserved-columns-v2. Authored-by: Andreas Neumann <andreas.neumann@databricks.com> Signed-off-by: Jose Torres <jtorres@apache.org>
…class ### What changes were proposed in this pull request? Remove the unused `ArrowStreamPandasSerializer` base class. After all Pandas UDF serializer subclasses were removed (`ArrowStreamPandasUDFSerializer`, `GroupPandasUDFSerializer`, `ArrowStreamAggPandasUDFSerializer`, `ApplyInPandasWithStateSerializer`, `CogroupPandasUDFSerializer`, `TransformWithStateInPandasSerializer`), the base class became an orphan with no subclasses, no instantiations, and no other references anywhere in the Python tree. This also drops the now-orphaned `_normalize_packed` helper (only used by that class) and the imports that are no longer needed. ### Why are the changes needed? Dead code cleanup. The class and its sole helper are unreachable. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing tests. No behavior change. ### Was this patch authored or co-authored using generative AI tooling? No. Closes apache#57544 from Yicong-Huang/SPARK-58354. Authored-by: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com> Signed-off-by: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com>
…) over Spark Connect Candidate-A pickle lane, Python-only v1: client cloudpickles the value and uploads it via the existing cache/<sha256> artifact channel; new CreateBroadcastCommand/UnpersistBroadcastCommand on ExecutePlan materialize a server-side Broadcast[PythonBroadcast] on the live driver SparkContext in SessionHolder; a new PythonUDF.broadcast_ids field (6) is resolved by SparkConnectPlanner into SimplePythonFunction.broadcastVars. Executor/worker path reused unchanged. broadcast_id == driver-side Broadcast.id so the worker's _broadcastRegistry/_from_id resolves it. DRAFT / DO-NOT-MERGE: proto _pb2.py stubs still need regen via dev/connect-gen-protos.sh (buf); BROADCAST_VALUE_TOO_LARGE quota not yet enforced; Scala/ScalarScalaUDF deferred (see companion spike branch). Co-authored-by: Isaac
Tagar
force-pushed
the
broadcast-connect-python-v1
branch
from
July 27, 2026 16:37
3248e82 to
d369c07
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
7 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
DRAFT / DO-NOT-MERGE — design preview to anchor the SPARK-51705 discussion. Not for merge in current form.
This PR adds a broadcast-variable API to Spark Connect (Python-only v1), closing the gap where a Connect/Serverless client cannot create a broadcast variable and reference it from a Python UDF (today this fails at the worker with
BROADCAST_VARIABLE_NOT_LOADED).Design: Candidate A — server-mediated pickle lane reusing classic
Broadcast[T]/TorrentBroadcast.The transport reuses the existing
cache/<sha256>artifact channel (content-addressing, dedup, per-sessionCacheId(sessionUUID,hash)isolation, ref-counted GC, transparent at-rest encryption) rather than adding a newbroadcasts/prefix. The typed handle comes fromCreateBroadcastResult, not the artifact prefix.Why are the changes needed?
JIRA: https://issues.apache.org/jira/browse/SPARK-51705 (
[CONNECT] Support sc.broadcast over Spark Connect, open, unassigned). Martin Grund endorsed "lift broadcast to the SparkSession" on-list. Broadcast is a core PySpark primitive with no Connect equivalent; Databricks Serverless / shared clusters run Connect, so the current guidance is to abandon shared/serverless for single-user clusters — a governance regression this closes.Carrier decision (re-verified against current master, 2026-07-20)
broadcast_idsrides the legacyPythonUDFproto message (field 6), resolved inSparkConnectPlanner.transformPythonFunctionintoSimplePythonFunction.broadcastVars. This was the #1 gate to re-check:udf/worker/module +ExternalUserDefinedFunction/ExternalUDFExecunder a default-OFF experimental SQLConf (spark.sql.execution.udf.unified.execution.enabled, v4.2.0). It is not wired into Spark Connect (0 references undersql/connect), added no.proto, and has no broadcast handling at all. So the legacyPythonUDFcarrier is correct and unobstructed for v1. Do NOT putbroadcast_idson anyudf/worker/proto/*message.PythonWorkerUtils.writeBroadcastsunchanged;core/broadcast.pyunchanged). Note: the Python worker decode was refactored (SPARK-56519 / SPARK-56324) —worker_util.setup_broadcastsnow delegates topyspark.worker_message.BroadcastInfo.from_stream— but this is Python-internal and format-preserving, so the Scala-side Connect changes need no worker edits.How was this patch tested?
(To be added; DRAFT skeleton.)
broadcastout oftest_connect_compatibility.py::expected_missing_connect_methods.spark.broadcast(dict_10k)referenced in a UDF over a >=1M-row DataFrame returns correct results (the exact case that fails today).spark.io.encryption.enabled=true): value re-encrypted at rest on server, worker reads viaEncryptedPythonBroadcastServer.unpersist()/destroy()frees driver+executor blocks and releases theRefCountedCacheId;SessionHolder.close()sweeps.broadcast_id->BROADCAST_NOT_FOUND(loud), not a silent empty list.Caveats / scope
Local verification status (2026-07-20)
protobuf==6.33.5,mypy-protobuf==3.3.0,bufremote pluginsprotocolbuffers/python:v33.5) viadev/connect-gen-protos.sh— the_pb2.py/_pb2.pyidiffs are limited to the new broadcast fields/messages; runtime import + field round-trip verified under protobuf 6.33.5.ruff check(pinnedruff==0.14.8) passes;ruff format --checkclean on all new/modified files.build/sbt), fullmypy(needs numpy/pandas-stubs), and the Connect E2E test suite (needs a running Connect server). These will run under Apache CI.Scope / caveats
transformPythonTableFunction) and DataSource (transformPythonDataSource) paths intentionally keep empty broadcasts; Scala/ScalarScalaUDFdeferred to a separate JIRA.ExternalUserDefinedFunction(opaque payload, no broadcast field) and the SQLConf flips to default-on, broadcasts need re-plumbing on that carrier. Tracked, not a v1 blocker.This pull request and its description were written by Isaac.
Proto diff
Companion: Scala UDF spike (
broadcast-connect-scala-spike)The anchor customer workload consumes its broadcast inside a Scala UDF, which Python-only v1 does not cover. The companion branch
broadcast-connect-scala-spikeis an exploratory spike (not for merge) documenting the path (seeSCALA-SPIKE-FINDINGS.md):Broadcast[T]object graph, so a broadcast id must be resolved during closure deserialization — unlike Python, where the id ridesPythonUDF.broadcast_idsand is injected intoSimplePythonFunction.broadcastVarswithout touching the closure.ConnectBroadcast[T].writeReplaceemits an id-only token; serverreadResolveswaps the id for the registryBroadcast[_]via a thread-local aroundunpackScalaUDF. Stacks on this PR'sSessionHolderregistry.SPARK-46032(closure deserialization on Connect) is a related reliability risk, not a strict blocker.Provenance
Design of record and cited impact analysis were developed in a companion research effort; this PR implements the "Candidate A" pickle lane. Note: the fork's base history is
apache/sparkand contains Apache's own SSL test-fixture PEM keys (core/src/test/resources/key.pem, etc.) — those are upstream fixtures, not introduced by these commits.