Connect Pool - #30
Open
ericm-db wants to merge 16 commits into
Open
Conversation
|
ericm-db
force-pushed
the
local-connect-reuse
branch
from
July 30, 2026 01:01
f4d9735 to
5f1f97b
Compare
…ROR_TEMP_3201-3205`: `MALFORMED_EXPRESSION_INFO`
### What changes were proposed in this pull request?
This PR proposes to assign a proper error condition for the legacy error conditions `_LEGACY_ERROR_TEMP_3201`, `_3202`, `_3203`, `_3204` and `_3205`, which are all thrown from the constructor of `ExpressionInfo` when the metadata describing an expression is malformed.
The five legacy conditions are folded into a single umbrella condition `MALFORMED_EXPRESSION_INFO` with five subclasses, one per validated field:
| Legacy | New condition | Field |
|---|---|---|
| `_LEGACY_ERROR_TEMP_3201` | `MALFORMED_EXPRESSION_INFO.NOTE` | `note` |
| `_LEGACY_ERROR_TEMP_3202` | `MALFORMED_EXPRESSION_INFO.GROUP` | `group` |
| `_LEGACY_ERROR_TEMP_3203` | `MALFORMED_EXPRESSION_INFO.SOURCE` | `source` |
| `_LEGACY_ERROR_TEMP_3204` | `MALFORMED_EXPRESSION_INFO.SINCE` | `since` |
| `_LEGACY_ERROR_TEMP_3205` | `MALFORMED_EXPRESSION_INFO.DEPRECATED` | `deprecated` |
The shared umbrella message names the offending field and expression (`'<fieldName>' is malformed in the expression [<exprName>]:`), and each subclass carries the field-specific detail. `fieldName` is a new message parameter, so `getMessageParameters()` for these errors now carries one extra key.
The assigned SQLSTATE is `22023` (invalid parameter value), consistent with the sibling `MALFORMED_*` conditions that validate a value against an allowed set or format (`MALFORMED_RECORD_IN_PARSING`, `MALFORMED_VARIANT`, and the `INVALID_PARAMETER_VALUE` archetype all use `22023`).
**On reachability, and why these get a proper name rather than `INTERNAL_ERROR`:** none of the five throw sites is reachable from a user query. `FunctionRegistryBase.expressionInfo` reads the compile-time `ExpressionDescription` annotation, and `SessionCatalog.makeExprInfoForHiveFunction` / `SQLFunction.toExpressionInfo` pass constants (`""` / `"hive"` / `"sql_udf"`). A malformed value can only come from extension or third-party code: constructing `new ExpressionInfo(...)` directly (which is what `SparkSessionExtensions.injectFunction` takes, see the example in `SparkSessionExtensionsProvider`), or calling `FunctionRegistryBase.createOrReplaceTempFunction(name, builder, source)` with an arbitrary `source`. That makes these developer-facing rather than engine-internal invariants: the person who triggers the error is the one who can fix it, so a named, actionable condition fits better than an internal error. This mirrors existing extension/configuration-author-facing conditions such as `CANNOT_LOAD_CATALOG` and `CANNOT_LOAD_FUNCTION_CLASS`, which also carry standard SQLSTATEs.
### Why are the changes needed?
`_LEGACY_ERROR_TEMP_*` conditions are placeholders that should be replaced with proper, named error conditions per the guideline in `common/utils/src/main/resources/error/README.md`. This is part of the ongoing effort to migrate legacy error conditions to the structured error framework.
### Does this PR introduce _any_ user-facing change?
No. As described above, these conditions are only reachable by extension or third-party code that builds an `ExpressionInfo` itself, and the `_LEGACY_ERROR_TEMP_*` names were never part of the public API.
For completeness, the message text is preserved with four deliberate changes:
- the `GROUP` value is now bracketed (`however, got <group>.` -> `however, got [<group>].`) for consistency with the other four subclasses;
- the split between the field and the detail moved from `.` to `:`, and the detail therefore starts with a lowercase `it should` instead of `It should`;
- the rendered message now carries the `[MALFORMED_EXPRESSION_INFO.<SUBCLASS>] ` prefix, which `SparkThrowableHelper.formatErrorMessage` omits only for `_LEGACY_ERROR_`-prefixed names;
- the rendered message now carries a ` SQLSTATE: 22023` suffix, since these five legacy entries previously had no `sqlState` at all.
Before and after, for `GROUP`:
```
OLD: 'group' is malformed in the expression [testName]. It should be a value in [...]; however, got invalid_group_funcs.
NEW: [MALFORMED_EXPRESSION_INFO.GROUP] 'group' is malformed in the expression [testName]: it should be a value in [...]; however, got [invalid_group_funcs]. SQLSTATE: 22023
```
### How was this patch tested?
Updated the existing assertions in `ExpressionInfoSuite` to check the new conditions and parameters, and added `sqlState = Some("22023")` to each so the assigned SQLSTATE is pinned by a test. Ran:
- `ExpressionInfoSuite` - 10/10 passed
- `SparkThrowableSuite` - 34/34 passed (JSON validity, alphabetical ordering, mandatory SQLSTATE, round-trip)
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes apache#57604 from LuciferYang/assign-name-legacy-3201-3205.
Authored-by: YangJie <yangjie01@baidu.com>
Signed-off-by: yangjie01 <yangjie01@baidu.com>
…ct when parse options are set
### What changes were proposed in this pull request?
The `CreateNamedStruct` branch of `OptimizeCsvJsonExprs` prunes a `from_json` schema down to the fields the struct selects, but unlike the sibling `GetStructField` and `GetArrayStructFields` branches it does not require the parse options to be empty. This adds the same `options.isEmpty` guard, so the rewrite is skipped whenever any option is set.
### Why are the changes needed?
Pruning the schema stops the parser from converting the dropped fields, so a malformed value in one of them is never reported. Under `mode=FAILFAST` that turns a query that should fail into one that silently returns a row:
```sql
SELECT named_struct(
'a', from_json(value, 'a int, b int, c int', map('mode', 'FAILFAST')).a,
'b', from_json(value, 'a int, b int, c int', map('mode', 'FAILFAST')).b)
FROM data -- value = '{"a": 1, "b": 2, "c": "bad"}'
```
With `spark.sql.optimizer.enableJsonExpressionOptimization=false` this raises `MALFORMED_RECORD_IN_PARSING`, because `c` is parsed and rejected. With the optimization on (the default) the schema is pruned to `a int, b int`, `c` is skipped, and the query returns `{a: 1, b: 2}`.
SPARK-32968 added this branch and SPARK-33907 added the `options.isEmpty` guard the following day, but only to the two `GetStructField`-style branches, so this one has been unguarded since 3.1.0.
### Does this PR introduce _any_ user-facing change?
Yes. `named_struct` over a `from_json` that carries parse options now honors those options again: a malformed record fails under `FAILFAST` instead of being silently accepted. Queries whose `from_json` has no options are unaffected. Note the rewrite also collapses several `from_json` evaluations into one; that consolidation is now skipped for the options-set case as well, matching what SPARK-33907 already accepted for the sibling branches.
### How was this patch tested?
Added an end-to-end `checkError` test in `JsonFunctionsSuite` that runs the query above with the optimization both on and off and asserts `MALFORMED_RECORD_IN_PARSING` either way; it fails on the unfixed tree because no exception is thrown. The bad field is a type mismatch rather than a structurally broken record: a structural malformation fails at tokenization no matter which schema is requested and would hide the pruning.
Added a plan-level test in `OptimizeJsonExprsSuite` asserting the rewrite is skipped for two different option maps (a parse mode and a formatting option, since the guard rejects any option), with an empty-options control asserting the same shape is still rewritten.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Closes apache#57605 from LuciferYang/SPARK-58373-json-prune-options.
Authored-by: YangJie <yangjie01@baidu.com>
Signed-off-by: yangjie01 <yangjie01@baidu.com>
…m the deduplicated subquery output ### What changes were proposed in this pull request? In `RewritePredicateSubquery.rewriteExistentialExprWithAttrs`, the `Not(InSubquery(...))` branch (the one handling a NOT IN nested inside a disjunction, e.g. `v > 0 OR x NOT IN (...)`) calls `dedupSubqueryOnSelfJoin` to alias the subquery's attributes when they conflict with the outer plan, but then builds the IN equality conditions from the pre-dedup `sub.output` instead of the deduplicated `newSub.output`. The join's right child uses `newSub`, so the condition can reference attributes that are no longer on the right side. This changes `sub.output` to `newSub.output`, matching the three sibling branches that already do this (the plain `InSubquery` branch in the same method, and both the top-level IN and NOT IN branches in `apply`). ### Why are the changes needed? When `dedupSubqueryOnSelfJoin` fires, it rebinds the conflicting subquery attributes to fresh exprIds. Building the condition from `sub.output` then uses the stale ids, which only exist on the outer side, so the null-aware anti-join condition collapses to trivially-true self-equalities like `id#2 = id#2` and no longer references the join's right child. That is exactly the SPARK-26078 defect the `dedupSubqueryOnSelfJoin` call is there to prevent, so today that call is dead weight on this branch. Analysis-time `DeduplicateRelations` currently renews subquery exprIds before the optimizer runs, so this is not reachable from user SQL on current `master` and produces no wrong results today. It is a latent correctness hole: any future change that lets an outer/subquery exprId conflict reach this rule would silently return wrong NOT IN results, and the branch is the odd one out among four otherwise-consistent sites. ### Does this PR introduce _any_ user-facing change? No. When dedup does not fire, `newSub` is the same object as `sub`, so the change is a no-op on every plan reachable from user SQL today. ### How was this patch tested? Added a `RewriteSubquerySuite` case that builds the colliding-attribute plan directly (bypassing the analyzer's `DeduplicateRelations`, which would otherwise renew the ids) and asserts the rewritten join condition references the deduplicated right-side output. It fails on the unfixed tree (the condition is `(a#0 = a#0) OR isnull((a#0 = a#0))`, referencing nothing on the right) and passes with the fix. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Closes apache#57558 from LuciferYang/SPARK-58365-notin-dedup. Authored-by: YangJie <yangjie01@baidu.com> Signed-off-by: yangjie01 <yangjie01@baidu.com>
…oercion in comparisons and predicates ### What changes were proposed in this pull request? Add non-ANSI type-coercion arms so that a string compared against a nanosecond-precision timestamp column (`TIMESTAMP_NTZ(p)`/`TIMESTAMP_LTZ(p)`, p in [7, 9]) is cast to that nanos type, mirroring the existing microsecond `TimestampType` handling (`StringPromotionTypeCoercion` equality arms + `TypeCoercion.findCommonTypeForBinaryComparison`). ### Why are the changes needed? Micros TimestampType has string-coercion arms that honor ...datetimeToString.enabled (legacy → promote to string; equality → cast to timestamp). Nanos had none, so it fell through to config-blind AtomicType promotion. This adds the arms so nanos matches TimestampType's legacy behavior. ### Does this PR introduce _any_ user-facing change? Only under legacy datetimeToString=true + ANSI off: range comparisons (<, BETWEEN, …) now promote to string (matching micros); equality unchanged. All other configs identical. ### How was this patch tested? Extended existing suites and added golden file tests. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code Closes apache#57223 from stevomitric/stevomitric/spark-57811-nanos. Authored-by: Stevo Mitric <stevomitric2000@gmail.com> Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com>
… from processPlan into a private helper ### What changes were proposed in this pull request? Extract the operator-ID-assignment phase of `ExplainUtils.processPlan` into a private `assignOperatorIds(plan, idMap)` helper. `processPlan` now initializes the idMap, delegates all ID assignment to `assignOperatorIds`, then performs the text-output pass over the returned subqueries and optimized-out exchanges. ### Why are the changes needed? `processPlan` conflated two independent sequential phases in one ~40-line body: 1. **ID assignment** — traversing the plan tree, subqueries, and adaptively-optimized-out exchanges (SPARK-42753) to populate an `IdentityHashMap` with monotonically-increasing operator IDs. 2. **Text output** — calling `processPlanSkippingSubqueries` on each discovered subtree to format the verbose explain string. These phases are sequential and independent: the text-output pass only begins after ID assignment is fully complete. Extracting phase 1 into `assignOperatorIds` shortens `processPlan` to its output logic and makes the boundary between the two phases explicit. No behavior change. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added `ExplainUtilsSuite` covering: - Operator IDs assigned to all visible plan nodes are unique - `processPlan` sets `CODEGEN_ID_TAG` on nodes inside `WholeStageCodegenExec` - Thread-local `localIdMap` is restored to its prior value after `processPlan` returns - Subquery section is emitted in the explain output when subqueries are present ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude (Anthropic) Closes apache#56216 from markj-db/mark-jarvin_data/explain-utils-generate-plan-ids. Authored-by: Mark Jarvin <mark.jarvin@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
### What changes were proposed in this pull request? This PR adds a new aggregate function `collect_union` that takes an array-typed column and returns the distinct union of the elements of the arrays across rows. `collect_union(col: array<T>) : array<T>` It is equivalent to `array_distinct(flatten(collect_list(col)))`, but the aggregation buffer holds only the distinct elements (a `HashSet`), so its size is bounded by the element universe rather than by the number of input rows. This avoids buffering every row's whole array, which for a hot grouping key can grow without bound. The function is implemented as a `Collect[mutable.HashSet[Any]]` (sibling of `collect_set`); the only material difference is that `update` iterates the input array and adds each non-null element, and the result element type is the input array's element type. NULL input arrays and NULL elements are skipped, following `collect_set` semantics. Added across the usual surfaces: Catalyst expression + registry, the Scala DataFrame API, and PySpark (classic + Spark Connect). Spark Connect needs no protocol change: the function travels as a generic `UnresolvedFunction` resolved against the registry. ### Why are the changes needed? There is no built-in aggregate that unions the elements of an array column across rows into a single distinct array. The workaround `array_distinct(flatten(collect_list(arr)))` buffers every row's whole array before de-duplicating, which can OOM on skewed keys. `collect_union` de-duplicates during aggregation, keeping the buffer bounded by the distinct-element universe. Note that `collect_set` cannot replace this. `collect_set(element)` over `explode(col)` does bound the buffer, but it stops being a plain aggregate: each array column must be exploded and grouped on its own and then joined back on the grouping keys. A query needing the distinct union of N array columns therefore pays N explodes + N joins purely to work around the missing array-input aggregate. `collect_union` keeps the bounded buffer while staying an ordinary aggregate, so multiple array columns aggregate together in one GROUP BY with no join. Industry precedent: BigQuery (GoogleSQL) already supports this style of array-input aggregate (`ARRAY_CONCAT_AGG`, which concatenates arrays across rows; distinct is then applied), whereas PostgreSQL has no dedicated function and users fall back to `array_agg(DISTINCT ...)` over `unnest(...)` (the analogue of the explode + `collect_set` workaround above). `collect_union` gives Spark a first-class, bounded-buffer form of this operation. ### Does this PR introduce _any_ user-facing change? Yes. It adds a new SQL function `collect_union` and the corresponding `functions.collect_union` in the Scala and Python DataFrame APIs. ### How was this patch tested? - New `collect_union function` case in `DataFrameAggregateSuite` (distinct union, NULL array, NULL element, per-group, empty result). Full suite: 170 tests, all pass. - New `test_collect_union` in `python/pyspark/sql/tests/test_functions.py` covering `array<int>`, `array<string>`, `array<double>`, NULL elements, and `array<struct>` (passes end-to-end through the PySpark runtime). - Spark Connect parity check in `test_connect_function.py`. - `ExpressionsSchemaSuite` regenerated `sql-expression-schema.md`. Closes apache#57592 from ChuckLin2025/collect_union-oss. Lead-authored-by: ChuckLin2025 <lzequn@gmail.com> Co-authored-by: Zequn Lin <chuck.lin@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
…pipelined-shuffle stage groups ### What changes were proposed in this pull request? This PR combines the two failure-handling layers of the pipelined-shuffle work into a single reviewable change, on top of the now-merged concurrent scheduling of a `PipelinedShuffleDependency` group (SPARK-58263 / apache#57341): the **group-atomic failure model** and the **fail-fast rejection of unsupported idioms** that together make co-scheduling safe under failure. (It builds on two merged predecessors -- the `PipelinedShuffleDependency` definition and type-based routing from SPARK-58185 / apache#57286, and the concurrent scheduling from SPARK-58263 / apache#57341 -- and targets `master` directly; its net-new content is the failure and fail-fast layers described below.) A pipelined shuffle is transient and incrementally read: its output is streamed to a co-scheduled consumer and is never materialized as durable, re-readable map output. That has two consequences the scheduler must enforce. **Group-atomic failure (a pipelined group succeeds or fails as a whole).** A transient shuffle cannot be recomputed in isolation and its members run concurrently, so the stock "resubmit one stage" recovery does not apply -- any member failure must fail the whole group, which the caller then reruns (a streaming query restarts the batch). - **Member task failure.** A pipelined group member's `TaskSet` is tagged `isPipelined` and pinned to `maxTaskFailures = 1`, so the first task failure aborts the set instead of retrying in place, which routes to a whole-group abort. - **Executor loss.** Losing an executor running a member task is force-counted as a member failure (except benign `TaskKilled` / `TaskCommitDenied`), aborting the group. - **FetchFailed on a member.** Handled by a dedicated branch that aborts the whole group rather than resubmitting the map stage in isolation (a lone-stage resubmit of a transient shuffle would deadlock the group). The failed executor's *regular* outputs are still unregistered from the `MapOutputTracker`, exactly as the base handler does, for the benefit of other jobs. - **No transient-producer resubmit.** A pipelined `ShuffleMapStage` records its completed partitions locally and monotonically (never in the `MapOutputTracker`) and never flips back to unavailable, so losing an executor that held an already-consumed pipelined output cannot make the scheduler resubmit the producer -- which would otherwise hang the producer's streaming writer waiting on termination acks from reducers that already finished. The `TaskSetManager` "Resubmitted" re-enqueue loop likewise excludes pipelined sets. - **Cross-job / cross-time reuse rejected.** A transient shuffle has no retained output for a second job to read, so binding a pipelined producer stage to a second job fails fast. **Fail-fast on unsupported idioms (spec S9).** A pipelined group is rejected up front, before any stage is created (so a rejection leaves no partial scheduler state), when it uses an idiom v1 cannot support: a producer feeding **more than one consumer** (1:N fan-out needs multicast, deferred); a **barrier**, **statically-indeterminate**, **checksum-mismatch-retry**, or **push-merge** producer; a **reliable RDD checkpoint** anywhere in a member's within-stage chain (it reintroduces cross-time reuse of a transient edge); or **members with differing resource profiles** (the gang slot check compares one demand against one profile's capacity, so v1 requires a single-profile group). These throw a typed `PipelinedShuffleUnsupportedException` (carrying the `PIPELINED_SHUFFLE_UNSUPPORTED` error class), which `handleJobSubmitted` matches by type. Main changes: - `DAGScheduler.scala` -- the FetchFailed group-abort branch, the no-resubmit handling of a pipelined `ShuffleMapStage`, cross-job reuse rejection, and `checkPipelinedGroupsSupportedInRDDGraph` / `checkPipelinedProducerSupported` fail-fast (typed exception). - `TaskSetManager.scala` -- `maxTaskFailures = 1` and force-counted executor loss for a pipelined set; exclusion from the "Resubmitted" re-enqueue loop. - `ShuffleMapStage.scala` -- monotonic local availability for a pipelined shuffle. - `PIPELINED_SHUFFLE_UNSUPPORTED` and `PIPELINED_SHUFFLE_CROSS_JOB_REUSE` error conditions. ### Why are the changes needed? Co-scheduling a pipelined group (apache#57341) is only safe if failure is handled at the granularity of the whole group: because the shuffle is transient and once-through, the stock per-stage resubmit recovery would either deadlock the group or hang a producer's streaming writer. This PR makes any member failure fail the group atomically (so the caller reruns it) and rejects up front the idioms whose recovery/semantics are incompatible with a transient, concurrently-read shuffle -- turning what would be a hang or a silently-wrong schedule into a clear, immediate failure. The two layers are combined into one PR because they are inseparable in review: the fail-fast rejections define exactly which group shapes the failure model must then handle, and both are gated on the same `PipelinedShuffleDependency` type. ### Does this PR introduce _any_ user-facing change? No. All new behavior is gated on a job using a `PipelinedShuffleDependency`, which nothing constructs yet, so every existing job is scheduled and recovers exactly as before. The new error conditions (`PIPELINED_SHUFFLE_UNSUPPORTED`, `PIPELINED_SHUFFLE_CROSS_JOB_REUSE`) can only surface for a job that uses a pipelined dependency. ### How was this patch tested? New unit tests in `DAGSchedulerSuite` and `TaskSetManagerSuite` cover: - group-atomic failure: `maxTaskFailures = 1` aborting a member set on the first failure; executor loss force-counted (and benign `TaskKilled` / `TaskCommitDenied` not force-counted); a member FetchFailed aborting the whole group rather than resubmitting a single stage; - no transient-producer resubmit: a post-executor-loss straggler success not resubmitting the producer; a completed pipelined producer's availability surviving executor loss; the "Resubmitted" loop excluding a pipelined set (including the partial-producer-on-decommission case); - cross-job reuse rejected; a group-atomic rerun resetting per-partition commit authorization; - fail-fast idioms: fan-out, barrier / indeterminate / checksum / push-merge producer, a reliable checkpoint in a producer's or a consumer's chain (including downstream in the consumer stage), and a mixed-resource-profile group -- each rejected up front; and that regular-shuffle idioms are NOT rejected (inertness of the fail-fast for a job with no pipelined dependency). The full `DAGSchedulerSuite` and `TaskSetManagerSuite` pass. ### Was this patch authored or co-authored using generative AI tooling? Co-authored with Claude Code (Opus 4.8) Closes apache#57361 from jerrypeng/stack/pipelined-shuffle-pr6-failfast. Authored-by: Boyang Jerry Peng <jerry.peng@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
…st listeners ### What changes were proposed in this pull request? Add a catch-all `case _ =>` as the final arm to a few SparkListener.onOtherEvent match blocks in test suites that enumerate only the specific event type(s) they care about and omit a catch-all: PluginContainerSuite, SQLExecutionSuite (jobTags / jobGroupId tests), and SparkConnectServiceInternalServerSuite. ### Why are the changes needed? SparkListenerBus.doPostEvent routes every non-built-in event to onOtherEvent, so a shared-queue listener receives all such events, not just the ones it enumerates. A match block with no `case _ =>` throws a scala.MatchError on every other event; ListenerBus.postToAll logs and swallows it, so tests pass but the logs are spammed with MatchError stack traces. The catch-all matches the convention the built-in listeners already follow and is always added last, so it never shadows an existing case. ### How was this patch tested? This change is test-only. ### Was this patch authored or co-authored using generative AI tooling? Co-authored w/ Claude Code. Closes apache#57618 from haoyangeng-db/minor-sparklistener-onotherevent-catchall. Authored-by: haoyangeng-db <haoyan.geng@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
…ry FileDescriptorSet ### What changes were proposed in this pull request? `from_protobuf` / `to_protobuf` used with a binary `FileDescriptorSet` hold their descriptor behind a `transient lazy val`, and building it parses the bytes into a large `FileDescriptor` object graph -- the tree that holds the (potentially millions of) `FieldDescriptor` instances. There is no caching, so each task instance in an executor JVM parses the set independently. Under high fan-out (many partitions -> many concurrent tasks in one JVM) with large descriptor sets (hundreds of KB, thousands of transitive fields), that produces many redundant, concurrently-retained copies of the same tree, driving sustained executor GC pressure. This PR adds a process-wide bounded cache at the `parseFileDescriptorSet` layer -- the layer that materializes the `FieldDescriptor` graph -- so concurrent task instances share one parsed graph per `FileDescriptorSet` instead of each rebuilding it. Because both `buildDescriptorFromFDS` and the `convertAnyFieldsToJson` `buildTypeRegistry(bytes)` path go through this layer, a given set is parsed at most once per JVM regardless of entry point. Building a `DescriptorWithExtensions` on top of the cached graph is cheap and happens once per task-instance init, so it is not itself cached; the extension-support flag is read per call, so a shared parse still yields the flag-appropriate result. The cache is keyed on a content hash of the descriptor bytes (which keeps the key small and avoids pinning the byte array) and does not cache failed parses. It uses `NonFateSharingCache` (SPARK-43300) so that a task cancelled while populating an entry does not cause spurious failures in other tasks blocked on the same key. Its size is bounded by a new internal config, `spark.sql.protobuf.descriptorCacheSize` (default 8; setting it to 0 disables the cache as a kill-switch). ### Why are the changes needed? To eliminate redundant, concurrently-retained copies of large parsed descriptor graphs and the resulting executor GC pressure for descriptor-heavy `from_protobuf` / `to_protobuf` workloads. ### Does this PR introduce _any_ user-facing change? No. A new internal config is added, but behavior is unchanged. ### How was this patch tested? New unit tests in `ProtobufFunctionsSuite` covering: repeated builds on the same bytes sharing one cached parse, distinct bytes cached separately, the `buildTypeRegistry(bytes)` path sharing the parse, the extension-support flag being honored per call over a shared parse, the disabled (size 0) kill-switch, an unknown message name surfacing the domain exception, and a failed parse not being cached (and surfacing the domain exception rather than a Guava wrapper). Existing `from_protobuf` / `to_protobuf` round-trip coverage exercises correctness with the cache enabled. ### Was this patch authored or co-authored using generative AI tooling? Yes, drafted with assistance from generative AI tooling. Closes apache#57321 from bhollis-dbx/protobuf-descriptor-cache. Authored-by: bhollis-dbx <ben.hollis@databricks.com> Signed-off-by: DB Tsai <dbtsai@dbtsai.com>
…TAMP_NTZ/LTZ(p) ### What changes were proposed in this pull request? This PR adds support for `TIMESTAMP_NTZ(p)` / `TIMESTAMP_LTZ(p)` (`p in [7, 9]`) in `+/- ANSI year-month interval` arithmetic. Concretely: - Extends `TimestampAddYMInterval` input typing to accept nanos timestamp types alongside the existing microsecond timestamp types. - Adds nanos-aware execution/codegen paths that apply the month shift on epoch micros while carrying the `nanosWithinMicro` remainder through unchanged. - Routes nanos timestamps into `TimestampAddYMInterval` from `BinaryArithmeticWithDatetimeResolver` for both `Add` and `Subtract` (the year-month branches previously matched only `TimestampType | TimestampNTZType`, so nanos timestamps fell through to an unresolved `Add` / `Subtract` and failed analysis). - Adds a `DateTimeUtils.timestampNanosAddMonths` helper. - Adds catalyst and SQL test coverage for NTZ/LTZ nanos year-month interval arithmetic parity, and regenerates the impacted nanos SQL golden files. ### Why are the changes needed? Spark already supports timestamp +/- ANSI year-month interval for the microsecond timestamp families, and SPARK-57501 added the day-time-interval parity for the nanos families, but year-month intervals were still unsupported for `TIMESTAMP_NTZ(p)` / `TIMESTAMP_LTZ(p)` (`p in [7, 9]`). This left valid datetime arithmetic unsupported for nanos types. These changes close that parity gap while preserving nanos precision semantics and existing LTZ/NTZ timezone behavior. ### Does this PR introduce _any_ user-facing change? Yes. `TIMESTAMP_NTZ(p)` / `TIMESTAMP_LTZ(p)` (`p in [7, 9]`) now support `+/- ANSI year-month interval` arithmetic. Examples: - `TIMESTAMP_NTZ '2020-01-02 03:04:05.123456789' + INTERVAL '1' YEAR` - `TIMESTAMP_LTZ '2020-01-02 03:04:05.123456789 UTC' - INTERVAL '1-2' YEAR TO MONTH` ### How was this patch tested? - `build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.expressions.DateExpressionsSuite org.apache.spark.sql.catalyst.util.DateTimeUtilsSuite'` - `SPARK_GENERATE_GOLDEN_FILES=1 build/sbt 'sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z "timestamp-ntz-nanos" -z "timestamp-ltz-nanos" -z "interval"'` - `build/sbt 'sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z "timestamp-ntz-nanos" -z "timestamp-ltz-nanos"'` - `build/sbt catalyst/scalastyle catalyst/Test/scalastyle` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) Closes apache#57636 from stevomitric/stevomitric/spark-57825-nanos. Authored-by: Stevo Mitric <stevomitric2000@gmail.com> Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com>
…for faster local startup
### What changes were proposed in this pull request?
Adds an opt-in fast path for local Spark Connect development. Today
`SparkSession.builder.remote("local[*]").getOrCreate()` starts a fresh in-process Connect server
that lives only as long as that Python process, so every run re-pays the cold start (JVM warmup,
`SparkContext` + server boot).
When `SPARK_LOCAL_CONNECT_REUSE=1` is set (or `spark.local.connect.reuse=true` on the builder), a
`local`-mode remote session instead reconnects to a persistent local Connect server. The first run
starts one through the standard `sbin/start-connect-server.sh` script (daemonized by
`sbin/spark-daemon.sh`, the same daemon a user would start by hand) and records host, port, auth
token, pid and Spark version in a discovery file. Later runs reuse that server if its Spark
version matches, its pid is alive, and its port accepts connections; otherwise they start a fresh
one. If a live but non-reusable server is still running (e.g. after a Spark upgrade), start-up
fails with an error pointing at the `--stop` command below, so old servers cannot silently
accumulate. User code is unchanged: the first run pays the cold start once, later runs reconnect
in a fraction of a second.
The implementation is `pyspark/sql/connect/local_server.py`, with `Discovery` handling the
discovery file (location, load/save, and the cross-process lock -- `Discovery` is a context
manager and all reads and writes happen under the lock), `LocalConnectServer` representing one
recorded server (its URL, reusability probe, stop), and `ServerLauncher` running the sbin script
and waiting for readiness. The first caller's startup confs are forwarded to the new server via a
`--properties-file`, mirroring what the in-process path does with its `SparkConf`.
The feature is off by default, Python-only, and POSIX-only (it relies on the `sbin/` shell
scripts). All state -- discovery file, daemon pid file, logs -- lives in a per-user `0700`
directory under the system temp dir rather than the home directory, so nothing accumulates
across reboots; `SPARK_LOCAL_CONNECT_DISCOVERY` overrides the location. The auth token is stored
with `0600` and the server binds localhost, so other users on the machine can neither read the
token nor reach the server. Each run is its own Connect session, so session-local state (temp
views, runtime SQL confs, artifacts) stays per-run; only shared `SparkContext` state (catalog,
global temp views, cached data) carries across runs.
The server runs until stopped with `python -m pyspark.sql.connect.local_server --stop`. It is an
ordinary `spark-daemon.sh` daemon, but it runs with a per-user pid directory and ident string so
it cannot collide with a manually started server -- which also means a plain
`sbin/stop-connect-server.sh` does not find it. The `--stop` command signals the recorded pid and
clears the discovery file; killing the pid directly also works, and the next run notices the dead
server and starts a fresh one. The earlier idle-timeout self-reaping was dropped with the switch
to the standard sbin daemon; a bounded lifetime would belong server-side (e.g. in
`SparkConnectService`), which also fits the follow-up discussed below of moving the
discovery-file write path into the server so any language client (and the proposed `spark
connect` CLI) can use the same mechanism.
### Why are the changes needed?
Creating a local Spark session for a quick edit/run loop takes a few seconds, and that cost is
one-time-per-process -- it does not amortize across separate runs. Keeping a warm server alive and
reconnecting to it is the only way to make a repeated local dev/test loop fast. This makes that
behavior available behind a single opt-in, without changing user code or default behavior.
### Does this PR introduce _any_ user-facing change?
Only when the opt-in is enabled. With `SPARK_LOCAL_CONNECT_REUSE=1` (or
`spark.local.connect.reuse=true` on the builder),
`SparkSession.builder.remote("local[*]").getOrCreate()` starts a persistent local Connect server on
the first run and reconnects on later runs, instead of booting a fresh in-process server each time.
With the opt-in unset (the default), behavior is unchanged. A new documentation section describes
the manual persistent-server workflow.
### How was this patch tested?
New `python/pyspark/sql/tests/connect/test_connect_local_server.py`. Unit tests cover the
`Discovery` save/load round-trip, file permissions and malformed-input rejection, the
`is_reusable` decision (version mismatch, dead pid, listening, closed port, and the Windows
pid-probe guard), stop semantics, the `--stop` CLI, and the POSIX guard. Four end-to-end tests
start a real server via the sbin scripts: builder integration with the opt-in, three concurrent
first-time startups converging on one server, reuse across calls with session isolation between
two connections, and the first caller's startup confs reaching the server's `SparkConf`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8, Fable 5)
Closes apache#56907 from ericm-db/local-connect-reuse.
Authored-by: Eric Marnadi <eric.marnadi@databricks.com>
Signed-off-by: Tian Gao <gaogaotiantian@hotmail.com>
ericm-db
force-pushed
the
local-connect-pool
branch
from
July 30, 2026 23:28
e350657 to
462580e
Compare
Expose the per-user runtime directory and startup seed configuration, and let LocalConnectServer start isolated daemons with an ephemeral port and precomputed configuration. Keep persistent-server reuse on the same launch path and add focused unit coverage.
ericm-db
force-pushed
the
local-connect-pool
branch
from
July 31, 2026 17:08
17787fa to
f43947b
Compare
Co-authored-by: Isaac Claude-Session: https://claude.ai/code/session_011tpdLYqY4VRw4s1KjDb2fi
Co-authored-by: Isaac Claude-Session: https://claude.ai/code/session_011tpdLYqY4VRw4s1KjDb2fi
Co-authored-by: Isaac Claude-Session: https://claude.ai/code/session_011tpdLYqY4VRw4s1KjDb2fi
Co-authored-by: Isaac Claude-Session: https://claude.ai/code/session_011tpdLYqY4VRw4s1KjDb2fi
ericm-db
force-pushed
the
local-connect-pool
branch
from
July 31, 2026 18:20
f43947b to
320bffc
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
8 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?
Why are the changes needed?
Does this PR introduce any user-facing change?
How was this patch tested?
Was this patch authored or co-authored using generative AI tooling?