[SPARK-58265][SQL] Reuse projected broadcast values for dynamic partition pruning

[sunchao]

What changes were proposed in this pull request?

Tracks SPARK-58265.

Extend dynamic partition pruning so Spark can derive partition values from the complete rows of a broadcast hash relation it has already built, rather than only from the hash keys of that relation.

Spark continues to prefer its existing, exact broadcast-key reuse. When that cannot satisfy a broadcast-only pruning filter, it looks for a broadcast already required by an earlier inner equi-join. If the broadcast comes from the expected source and uses the complete ordered join keys and the correct hash mode, Spark evaluates a supported deterministic partition expression against its stored value rows and feeds the resulting values into the existing partition-pruning path.

The key difference is that Spark reuses work the query already has to perform. It does not create a second broadcast or rerun the filtering-side join just to discover the partitions to read. The same approach works with adaptive and nonadaptive planning and with V1 and V2 partition filtering.

Correctness takes priority over pruning. The projected values must include every partition the original query could need; they may also include values from broadcast rows that an earlier join later rejects. If Spark cannot verify the broadcast, cannot safely evaluate the expression, cannot obtain runtime statistics, or exceeds a configured row or size limit, it discards the entire projected domain and continues without this optimization. An unavailable domain is never interpreted as an empty domain.

Why are the changes needed?

Broadcast hash relations contain full build-side rows, not just hash keys. Existing dynamic partition pruning can reuse the keys, but it cannot use another column from the same broadcast even when that column identifies the partitions of a later scan.

For example, suppose sales is partitioned by category:

SELECT s.*
FROM sales s
JOIN (
  SELECT /*+ BROADCAST(p) */
    p.category
  FROM products p
  JOIN promotions promo
    ON p.product_id = promo.product_id
  WHERE promo.active
) eligible
  ON s.category = eligible.category

The first join broadcasts products using product_id as its hash key. Each broadcast row also contains category, but category is not part of that hash key:

product_id   category
----------   --------
101          books
202          toys
303          clothing

Suppose only products 101 and 202 have active promotions, and sales has four category partitions: books, toys, clothing, and electronics.

Without this change, Spark cannot derive the sales partition values from the existing product_id broadcast. With spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=true, it drops the pruning filter and scans every sales partition:

Before:                 books, toys, clothing, electronics
Exact active products:  books, toys
Reused broadcast rows:  books, toys, clothing

With this change, Spark reads category directly from the existing broadcast and avoids the electronics partition. It may still read clothing: the broadcast contains product 303 even though the promotions join will later eliminate it. That extra partition is a safe superset, not a regression: Spark previously scanned clothing as part of the full scan, and the original join still excludes inactive products from the final result.

If Spark can already construct an exact reusable filter, it keeps using that filter first. The projected-broadcast path is for queries that would otherwise receive no broadcast-only pruning; it does not replace a more selective existing pruning strategy.

Does this PR introduce any user-facing change?

No. The optimization is disabled by default and controlled only by internal SQL configurations. It does not introduce public configuration, SQL syntax, APIs, or changes to query results.

How was this patch tested?

Regression tests cover derived partition expressions; composite and single-column broadcast keys; dense and sparse long-key hash relations; duplicate, null, and unmatched broadcast rows; safe-superset pruning; feature-disabled behavior; row and byte limit fallback; preservation of Catalyst case-class identity and metadata across rewrites; and both passes of iterative V2 partition filtering.

The shared dynamic partition pruning tests run against V1, V2, and runtime-filtering V2 sources with adaptive execution both enabled and disabled. Enhanced V2 connector tests separately verify that a valid projected domain prunes partitions and that an unavailable domain leaves all original partitions intact.

The focused pruning, ordinary subquery, runtime-filter injection, and existing Bloom filter suites passed locally: 390 executed tests passed, with five skipped.

SERIAL_SBT_TESTS=1 ./build/sbt \
  'catalyst/testOnly org.apache.spark.sql.catalyst.expressions.DynamicPruningSubquerySuite' \
  'sql/testOnly org.apache.spark.sql.DynamicPartitionPruningV1SuiteAEOff org.apache.spark.sql.DynamicPartitionPruningV1SuiteAEOn org.apache.spark.sql.DynamicPartitionPruningV2SuiteAEOff org.apache.spark.sql.DynamicPartitionPruningV2SuiteAEOn org.apache.spark.sql.DynamicPartitionPruningV2FilterSuiteAEOff org.apache.spark.sql.DynamicPartitionPruningV2FilterSuiteAEOn org.apache.spark.sql.connector.DataSourceV2EnhancedRuntimePartitionFilterSuite'

SERIAL_SBT_TESTS=1 ./build/sbt \
  'sql/testOnly org.apache.spark.sql.SubquerySuite org.apache.spark.sql.InjectRuntimeFilterSuite org.apache.spark.sql.BloomFilterAggregateQuerySuite'

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex (GPT-5)

[sunchao]

cc @cloud-fan @viirya @peter-toth @dongjoon-hyun please take a look and share your opinions on this PR. Thanks 🙏

[viirya]

This is a complex but carefully-designed optimization, and I verified the core correctness claims hold (I traced the hot paths and had two focused passes over the hardest invariants):

• Safe-superset invariant holds. find descends to the first inner equi-join and keys the projection on the source side's join keys; since the pre-join source is a superset of post-join rows, the projected values are a superset. Null handling matches existing DPP (null hash keys and null projected values are correctly excludable — they never match an equi-join key), and isSafeValueExpression is restrictive enough that evaluating the expression on build rows equals evaluating it on the join output.
• Broadcast reuse binds the right relation. The non-AQE path matches on both sameResult and broadcastMode equality; HashedRelationBroadcastMode stores full rows, so non-key value columns are projectable, and bindReference resolves by ExprId (not positional ordinal), so a reordering Project between the join and the scan is handled correctly. isNullAware handling is consistent across AQE/non-AQE.
• "Unavailable is never empty" is enforced end to end. On any failure (unverifiable broadcast, missing runtime stats, row/byte/source limits, evaluation error) the projection fails open: V1 via InSubqueryExec.eval == true, V2 via explicit isResultUnavailable/TrueLiteral guards in translateRuntimeFilterV2 and PushDownUtils. Default-off means no plan changes (confirmed no golden-plan diffs).

A few non-blocking points:

1. Config visibility vs. the "user-facing" claim. The description says this is a user-facing, opt-in feature, but all four configs are .internal(), so they won't appear in the generated SQL config docs. Please make the contract explicit either way: either promote them to public experimental configs and document them (stating the safe-superset guarantee, that the optimization may silently no-op, and the three limits), or keep them internal and drop the "user-facing" framing. An internal, default-off flag is a perfectly normal way to land an opt-in optimization — the issue is just that the current wording and the config visibility disagree, which is unclear for both users and future maintainers.

2. BroadcastValueProjection is carried as a TreeNodeTag side-channel rather than a case-class field, so it's copied manually in the overridden withNewPlan/withNewOuterAttrs/withNewHint/withNewChildInternal. Any transform path that doesn't preserve the tag silently drops the projection — which is fail-open (just disables the optimization), so it's safe, but it's a non-standard Catalyst pattern. Worth calling out the tradeoff in the PR and getting @cloud-fan / @peter-toth to weigh in on side-channel tag vs. a real field.

3. Add a targeted test for value-column ordinal alignment when a Project reorders columns between the join and the partition scan. ExprId-based binding makes this correct today, but it's the one dependency this path adds beyond key-only reuse, so a test would lock it in against future refactors.

4. Minor consistency nit: the AQE reusableBroadcast matches on sameResult(exchange) alone while the non-AQE path also checks broadcastMode == requiredMode. Equivalent in practice, but making them identical would read more clearly.

Given the complexity and correctness sensitivity, I'd like @cloud-fan / @peter-toth to also take a look, especially on the metadata-tag design and the config contract.

[dongjoon-hyun]

I agree with the above @viirya 's comments. I added three other comments.

[dongjoon-hyun]

The isSafeValueExpression whitelist seems inconsistent with the motivating example in the PR description. The Attribute case requires UnsafeRow.isFixedLength(attribute.dataType), which rejects variable-length types like StringType — yet the example in the description projects a string column (category) from the broadcast. As written, that example query would not benefit from this optimization.

Also, the DateFormatClass case hardcodes the 'yyyy-MM-dd' format string, which looks tailored to the test queries rather than a principled rule.

Could you clarify:

1. Why is fixed-length required for plain attribute projections? Output size is already bounded by maxOutputBytes, so it doesn't seem to be a memory concern, and correctness shouldn't depend on the value width.
2. If the restriction is intentional, please update the PR description so the example matches what the whitelist actually supports; if not, consider generalizing the whitelist (at least plain string attributes) and documenting the criteria for extending it.

[dongjoon-hyun]

To make it sure, please add a test case including String type at least.

[sunchao]

Fixed in 5c57963.

Plain default-collation StringType attributes are now supported; the existing distinct-value byte limit still bounds variable-width output, and non-binary collations remain conservatively excluded. DateFormatClass now accepts any non-null, resolved string-literal pattern instead of hardcoding yyyy-MM-dd.

The new reuse ancestor broadcast string values with a residual join predicate regression uses the products.category example, verifies both the exact query result and the projected books, toys, clothing superset, and runs for V1/V2 scans with adaptive execution both enabled and disabled. The reordered-column regression separately exercises the yyyyMMdd format.

[dongjoon-hyun]

This tightens the existing direct reuse path, independently of the new feature: the old code only checked left.sameResult(sparkPlan), while the new code additionally requires broadcast mode equality and excludes isNullAwareAntiJoin joins.

With the default spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=true, an edge case that previously got DPP (at the cost of planning an extra broadcast that ReuseExchange couldn't dedup, e.g. when key order differs or the only matching join is a null-aware anti join) now falls back to Literal.TrueLiteral, i.e. loses DPP entirely.

Making the non-AQE path consistent with the AQE path (which already compares the whole exchange including its mode) is arguably an improvement, but it is a semantic change to released behavior that ships silently inside this feature PR. Could you either call this out explicitly in the PR description (ideally with a test pinning the new behavior), or split it into a separate commit/PR so it can be evaluated on its own?

[sunchao]

Fixed in 5c57963. The non-adaptive direct-reuse path now preserves the existing candidatePlan.sameResult(sparkPlan) behavior, including mismatched hash modes and null-aware joins. Exact hash-mode matching and the null-aware restriction apply only to the new projected-broadcast path.

I also added preserve direct broadcast pruning for existing hash and null-aware modes, which explicitly verifies both pre-existing cases with the new optimization disabled. This keeps the feature from changing released direct-reuse behavior.

[dongjoon-hyun]

executeCollect() maps Unavailable to an empty array. The whole design rests on "an unavailable domain is never interpreted as an empty domain", but that invariant is only upheld because InSubqueryExec.updateResult goes through ProjectedBroadcastValueSubqueryExec.resultOf. Any future caller that invokes executeCollect() directly would read "empty domain" and prune every partition — a silent correctness bug. Consider throwing from executeCollect() instead (like doExecute() does), or at minimum adding a prominent comment stating that callers must use resultOf/executeCollectResult()?

[sunchao]

Fixed in 5c57963. executeCollect() now throws SparkException.internalError when the projected domain is unavailable rather than returning an empty array. Dynamic partition pruning explicitly consumes the typed executeCollectResult()/resultOf path and still fails open.

The regression exercises the row, projected-byte, and source-byte limits and asserts that direct executeCollect() raises INTERNAL_ERROR in each case. An unavailable domain therefore cannot silently become an empty pruning domain.

[dongjoon-hyun]

Since InSubqueryExec is also used a general IN subquery, shall we assert(isDynamicPruning) here?

Suggested change

	result = if (unavailable) {
	result = if (unavailable) {
	assert(isDynamicPruning,
	"An unavailable broadcast value projection result is only allowed for dynamic " +
	"pruning, where the filter can safely fail open to true.")

[sunchao]

Fixed in 5c57963; updateResult() now asserts isDynamicPruning before installing the unavailable-result marker.

The limit regressions also copy the same expression with isDynamicPruning = false and verify that updateResult() raises AssertionError. Ordinary SQL IN therefore cannot accidentally use dynamic pruning's fail-open semantics.

[peter-toth]

Thanks for the PR, @sunchao! I reviewed this independently before reading the other comments, with a couple of focused passes on the correctness-sensitive parts.

I land where @viirya did: the safe-superset holds, null handling matches existing DPP (a source row with a null in any hash-key component is dropped from the broadcast -- which is exactly a row an inner equi-join could never keep), and "unavailable is never empty" is enforced end to end. I also checked the semi/anti reuse worry: HashedRelationBroadcastMode.transform always builds with ignoresDuplicatedKey = false (that optimization is ShuffledHashJoinExec-only), so a reused broadcast never drops value rows.

I agree with @viirya's points (internal-config vs. "user-facing" wording, the TreeNodeTag side-channel, the reorder-Project ordinal test, AQE/non-AQE mode-check consistency) and @dongjoon-hyun's (the direct-reuse tightening in PlanDynamicPruningFilters, the executeCollect() -> empty footgun, assert(isDynamicPruning)) -- these are all still open, so treat this as adding to them, not duplicating. One add to @dongjoon-hyun's isSafeValueExpression thread: he's right that correctness doesn't depend on the value width -- the safe-superset holds for any deterministic expression over the source row -- so the UnsafeRow.isFixedLength gate is purely conservative, and it happens to exclude string partition columns, which is both the common DPP case and the description's own category example.

My one net-new point is on the metadata design (inline).

Alternatives

• 1. Re-derive the projection in the planner rather than carry it across phases: nearly everything find returns is derivable from the DPP's buildQuery (which the direct-reuse path already re-plans at physical time), and re-deriving there is also lazy -- computed only in the narrow fallback where it's used, instead of eagerly for every DPP. The only find input missing at planning is excludedPlan, which is always the pruned leaf scan. If carrying is deliberate, a @transient field in a second parameter list (as LogicalRDD does) fits better than the tag. [inline: PartitionPruning.scala:114]

[peter-toth]

Finding 1. Nearly everything find produces here is re-derivable from the DPP's own buildQuery, which the planner already has -- so this may not need to cross phases at all, and could be computed lazily only when it's used:

• the direct-reuse path in PlanDynamicPruningFilters already does createSparkPlan(planner, buildPlan) and matches it via sameResult, and sourcePlan is just a subtree of that same buildQuery;
• the filtering key is buildKeys(broadcastKeyIndices.head).

Today find runs eagerly here for every candidate DPP whenever the config is on, but its result is consumed only when direct reuse fails and onlyInBroadcast -- a narrow case. Running find(buildKeys(broadcastKeyIndices.head), buildQuery, ...) in the planner, gated behind that fallback, drops the tag machinery entirely and does strictly less optimizer work.

The one input not available at planning is find's excludedPlan -- but it's always the pruned leaf scan (getFilterableTableScan returns only LogicalRelation/HiveTableRelation/DataSourceV2ScanRelation), used only for the self-reference guard !source.exists(_.sameResult(excludedPlan)), which doesn't affect the safe-superset and targets a scan the DPP already identifies via pruningKey.

If carrying is deliberate instead, picking up @viirya's tag-vs-field question: a @transient field in a second parameter list -- exactly how LogicalRDD carries session/originStats/originConstraints and @transient stream -- would fit better than the tag. It keeps productArity at 7 (the existing 7-arg extractors and the productArity === 7 test are untouched, so "without changing the case-class shape" still holds), stays out of equals/canonicalization (dedup unaffected, like today), stays transient (meets the "don't serialize the source plan" requirement), and is carried through transforms via otherCopyArgs -- removing the four manual copyBroadcastValueMetadataTo overrides and the silent-drop-on-.copy() risk. (SubqueryAdaptiveBroadcastExec would get the same treatment for the AQE path.)

[sunchao]

Thanks, this is a helpful distinction. You are right that find is eager once the internal flag is enabled, even though projected reuse is only needed after direct reuse fails.

For this revision I kept the transient tag so that the exact pruned-leaf exclusion and selected source are carried consistently into both the adaptive and non-adaptive planners, without changing the existing case-class constructor or canonicalization. The tag is transient, all four rewrite paths are covered by broadcast value metadata survives logical rewrites without changing DPP identity, and a missing tag disables the optimization rather than changing the query result.

I agree that either lazy re-derivation or a transient second parameter list could simplify the implementation. The second-parameter-list approach in particular addresses the productArity concern. I would be happy to switch if you or @viirya prefer that direction; I did not want to silently broaden the metadata and AQE changes while addressing the correctness feedback.

The same update also switches to ExtractEquiJoinKeys, so residual ancestor-join conditions no longer incorrectly prevent reuse.

[peter-toth]

Thanks @sunchao -- one clarification first: I'd call it "lazy derivation" rather than "re-derivation." The idea isn't to run find twice; it's to not run it in PartitionPruning at all, and instead run it once in PlanDynamicPruningFilters / PlanAdaptiveDynamicPruningFilters, gated behind the directBroadcast.isEmpty && onlyInBroadcast fallback. Both rules already seem to have what find needs: the SparkPlan is built (so the reusable-BroadcastHashJoinExec search the direct path already does is right there), and the DynamicPruningSubquery still carries buildQuery (the AQE node carries it as buildPlan), so find(buildKeys(broadcastKeyIndices.head), buildQuery, ...) runs with no new inputs -- with one exception.

That exception is the pruned-leaf exclusion (excludedPlan), which I flagged above as the one input not in the planner -- I think it's also what you mean by carrying "the exact pruned-leaf exclusion" consistently. It's always the pruned leaf scan and the DPP already identifies its pruning target through pruningKey, so it seems re-expressible in the planner -- but if it turns out it can't be done cleanly there, I'd take that as a fair argument for keeping the projection carried as it is now.

The reason I'd single lazy derivation out from the transient field: they're not the same weight. The transient field still carries the derived projection across phases (just via Catalyst's copy machinery instead of a tag). Lazy derivation carries nothing new -- no tag, no field, no PlanAdaptiveSubqueries hand-off -- and does strictly less optimizer work, since find currently runs eagerly for every candidate DPP even though projected reuse only matters after direct reuse fails.

On the transient point itself: I don't think it really applies on the logical side. sourcePlan is a subtree of buildQuery, and buildQuery is a plain, non-transient field on the same DynamicPruningSubquery -- it can be non-transient because the node is Unevaluable and is rewritten into InSubqueryExec before the plan is ever serialized for execution. So a serialized plan would already retain sourcePlan through buildQuery; the @transient on the projection doesn't add protection there. (It does genuinely matter on SubqueryAdaptiveBroadcastExec.buildPlan, since that's a physical node.)

All that said, transient second-parameter-list field is a perfectly good outcome and clearly unblocks merge -- it keeps productArity at 7 and lets the normal copy machinery carry the metadata, which removes the silent-drop risk. I just wanted to make the lazy option explicit, since it's the lighter of the two. Happy either way -- whatever you and @cloud-fan prefer.

[sunchao]

Thanks @peter-toth. This is now addressed with the @transient second-parameter-list approach on both DynamicPruningSubquery and SubqueryAdaptiveBroadcastExec. otherCopyArgs carries the metadata through Catalyst rewrites without changing product arity, equality, or canonicalization. I kept the explicit handoff so the exact pruned-leaf exclusion remains available consistently to both adaptive and nonadaptive planning.

[sunchao]

Thanks @viirya, @dongjoon-hyun, and @peter-toth for the careful reviews. I have pushed 5c57963 with the following updates:

• Support default-collation string projection and arbitrary non-null literal date formats, with an end-to-end category regression.
• Preserve the original non-adaptive direct-broadcast reuse behavior; apply exact broadcast-mode and null-aware checks only to projected reuse.
• Throw from executeCollect() for unavailable projected domains and assert that fail-open semantics are used only for dynamic partition pruning.
• Add an explicit reordered-column regression; the value is bound against the broadcast output by expression ID, not assumed ordinal.
• Use Spark's ExtractEquiJoinKeys, so ancestor broadcast joins with residual conditions can still reuse their actual ordered hash keys.

The four settings remain intentionally internal and default-off; this PR does not introduce a public configuration or API. I replied separately to the metadata-design thread because lazy re-derivation versus a transient second parameter list is the remaining design decision.

Validation for the updated commit: 279 tests passed across the Catalyst suite, all six adaptive/non-adaptive V1/V2 dynamic-pruning variants, and the iterative V2 connector suite. All four Catalyst/SQL Scala style checks also passed.

[cloud-fan]

1 blocking, 0 non-blocking, 0 nits.
The execution semantics look sound, but the hidden TreeNodeTag transport should be replaced with an explicit transient field before merge.

Already raised in existing discussion (1)

• Replace the TreeNodeTag side channel with an explicit transient field in a second parameter list on DynamicPruningSubquery (and the adaptive carrier). The tag requires manual propagation through selected rewrite hooks, so a future .copy(...) or Catalyst rewrite can silently lose the candidate. A transient second parameter list preserves the existing seven-field extractor, product arity, equality, and canonicalization while allowing Catalyst's normal copy machinery to carry the metadata. -- existing discussion

Verification

Reviewed the complete 18-file patch and traced the logical DPP subtree, candidate discovery, AQE/non-AQE physical matching, hashed-relation row recovery, bounded projection, unavailable-state handling, V1/V2 consumers, configuration boundaries, and focused tests. The text-quality scanner covered 29 candidates with no findings; the contract scanner verified that only projected reuse receives the stricter hash-mode gate.

[sunchao]

Thanks, @cloud-fan. I replaced the TreeNodeTag side channel with an explicit @transient field in a second parameter list on both DynamicPruningSubquery and SubqueryAdaptiveBroadcastExec.

Both nodes now use otherCopyArgs, so Catalyst's normal makeCopy/rewrite machinery preserves the projection metadata automatically. The existing extractor and product arities remain unchanged, the metadata does not affect equality, and canonicalization explicitly clears it. I also removed the old tag helper and added regression coverage for direct copies, reflective makeCopy, child/plan rewrites, AQE transport, transient fields, and canonicalization.

Validation passed across the Catalyst, V1/V2 DPP with AQE on and off, regular subquery, runtime-filter, and Bloom-filter suites, as well as the relevant Scala style checks.

Would you mind taking another look?

[peter-toth]

Re-checked through f4c8dce — finding 1 resolved: the projection now rides a @transient field in a second parameter list on both DynamicPruningSubquery and SubqueryAdaptiveBroadcastExec, carried by otherCopyArgs, with productArity still 7 and canonicalization clearing it. That's @cloud-fan's pick between the two options I laid out, so I'm not re-opening the lazy-derivation half.

I re-verified what the last two commits changed: projection.isEmpty || makes the non-AQE direct-reuse path equivalent to base; the switch to ExtractEquiJoinKeys can't admit a null-matching join, because EqualNullSafe yields Coalesce/IsNull keys that isSafeSourceHashKey rejects; PlanAdaptiveDynamicPruningFilters runs before ReuseAdaptiveSubquery (AdaptiveSparkPlanExec.scala:155-157), so canonicalization dropping the projection can't cross-wire two DPP filters; and resultOf unwraps nested ReusedSubqueryExec, so the new executeCollect() throw is unreachable from the DPP path.

Two still-open items from others that I'm not re-raising, just flagging as unaddressed: @viirya's config-visibility point — the description still answers "Yes" to user-facing change and names one of the four new configs, while your reply says the PR "does not introduce a public configuration or API"; those should agree, since the description becomes the commit message. And the second half of @dongjoon-hyun's isSafeValueExpression ask, "document the criteria for extending it" — worth noting there that determinism, the no-subquery/outer-ref check, the row/byte limits and the NonFatal fallback already carry the safety, so the structural allowlist is a policy choice rather than a correctness one.

Non-blocking

• 2. withNewPlan now also clears resultBroadcast (late catch): clearing result is all this PR needs, and clearing resultBroadcast changes the shared non-DPP IN-subquery path away from base with no test or comment. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:131]
• 3. A malformed projection un-resolves the DPP node instead of being dropped (new): every clause added to resolved is already guaranteed by ReusableBroadcastValueProjection.find, so it can only fire for a projection built some other way — and then it inverts the PR's own fail-open contract. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/DynamicPruning.scala:90]

[peter-toth]

Finding 2. Clearing result is what this PR needs; clearing resultBroadcast reaches a path the PR isn't targeting.

resultBroadcast is only ever assigned for non-DPP subqueries — updateResult guards it with !isDynamicPruning && !isResultUnavailable — so the unavailable marker can only ever live in result, and for DPP resultBroadcast is always null anyway. The one case where the extra reset does something is an ordinary IN subquery: ReuseAdaptiveSubquery calls withNewPlan(ReusedSubqueryExec(...)), and if that lands after updateResult() has run, the new instance now drops a broadcast that base kept and has to re-run updateResult() and re-broadcast. It recovers (a fresh instance re-prepares), so nothing breaks — but it's a silent behavior change to a shared method, with no test and nothing in the description.

Suggested change

	copy(plan = plan, resultBroadcast = null, result = null)
	copy(plan = plan, result = null)

If clearing the broadcast is deliberate — say you want a plan swap to always recompute — a one-line comment saying so would stop the next reader from "simplifying" it back.

[sunchao]

Fixed in 549b6db, thanks @peter-toth. InSubqueryExec.withNewPlan now clears only result and preserves resultBroadcast, restoring the existing behavior for ordinary IN subqueries. The new IN predicate subquery preserves its broadcast when replacing its plan regression clears the transient result and verifies that evaluation still succeeds through the retained broadcast.

[peter-toth]

Finding 3. This is the one place in the PR where a bad projection fails the query rather than being discarded.

Every clause here is already guaranteed by ReusableBroadcastValueProjection.find:

• sourcePlan.deterministic, sourceHashKeys.nonEmpty, and the hash-key determinism come from the candidate.filter block plus isSafeSourceHashKey;
• sourceHashKeys / valueExpression references.subsetOf(sourcePlan.outputSet) come from ExtractEquiJoinKeys' canEvaluate and from find's own value.references.subsetOf(left.outputSet);
• valueExpression.deterministic comes from isSafeValueExpression;
• valueExpression.dataType == pruningKey.dataType follows because the value starts as filteringKeys.head, PartitionPruning sets indices = Seq(joinKeys.indexOf(filteringKeys.head)), and replaceAlias is type-preserving — so it is the same expression the line above already type-checks against pruningKey.

So these clauses can only fire for a projection that did not come from find. When they do, the effect is resolved == false on a node inside an already-optimized plan: the containing Filter reports unresolved, LogicalPlanIntegrity.validateOptimizedPlan flags it under plan-change validation, and outside validation an unresolved node just keeps flowing. That inverts the contract the description states — "If Spark cannot verify the broadcast, cannot safely evaluate the expression ... it discards the entire projected domain and continues without this optimization." Your own test pins the fail-closed shape:

assert(!pruning.copy()(Some(projection.copy(sourceHashKeys = Seq(missing)))).resolved)

Keeping the checks but not the hard failure, either:

// drop rather than un-resolve; both planner rules read this instead of the raw field
private[sql] lazy val usableBroadcastValueProjection: Option[BroadcastValueProjection] =
  broadcastValueProjection.filter(isSafeProjection)

or move them to a require in BroadcastValueProjection's producer, so a malformed projection is a bug at the producer instead of a plan-validation failure later. Either way DynamicPruningSubquery.resolved stays about the DPP itself, and an unusable projection behaves like every other failure in this feature.

[sunchao]

Fixed in 549b6db, thanks @peter-toth. Projection validation now lives in usableBroadcastValueProjection instead of DynamicPruningSubquery.resolved. Both nonadaptive planning and the adaptive handoff consume only the filtered projection, so malformed optional metadata leaves the DPP expression resolved and simply disables projected reuse. The Catalyst regression covers missing source/value attributes, empty hash keys, and mismatched value types.

[cloud-fan]

1 addressed, 0 remaining, 0 new.
0 blocking, 0 non-blocking, 0 nits.
The previous metadata-transport concern is addressed, and the current implementation preserves the safe-superset and fail-open invariants across the reviewed execution paths.

Verification

I traced candidate discovery from PartitionPruning through both physical planning rules, checked that projected reuse requires the source plan and complete ordered hash mode to match an existing broadcast, and followed Available and Unavailable results through InSubqueryExec and both V2 runtime-filter consumers. I also verified that direct reuse remains first, malformed internal carrier state is rejected by Catalyst resolution, and recoverable projection failures cannot become an empty pruning domain.

[peter-toth]

Re-checked through 549b6db — findings 2, 3 resolved (withNewPlan keeps resultBroadcast; projection validation moved out of resolved into usableBroadcastValueProjection, so a malformed projection is now dropped instead of un-resolving the node), nothing regressed. I confirmed usableBroadcastValueProjection is the only consumer on both planning paths — non-AQE reads it directly, AQE reads the already-filtered field off SubqueryAdaptiveBroadcastExec via PlanAdaptiveSubqueries — and the new allowlist comment covers the second half of @dongjoon-hyun's ask.

Two late catches below, both on code that has been in the PR since the first commit.

Still unaddressed from others, not re-opening: @viirya's config-visibility point — the description still answers "Yes" to user-facing change and names one of the four new configs, while all four are .internal() and your reply says the PR "does not introduce a public configuration or API".

Blocking

• 4. New BaseSubqueryExec subtype not registered in SQLLastAttemptAccumulator (late catch): extractStageRDDScopes matches the BaseSubqueryExec subtypes exhaustively and bails out on anything unknown, so every query that plans a ProjectedBroadcastValueSubqueryExec makes lastAttemptValueForQueryExecution/lastAttemptValueForDataset return None. One line, next to the existing SubqueryBroadcastExec case. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/ProjectedBroadcastValueSubqueryExec.scala:36]

Non-blocking

• 5. Dropping DynamicPruningExpression(Literal.TrueLiteral) changes the existing V2 iterative pushdown path (late catch): on base that filter reaches PartitionPredicateImpl and pushes a no-op true partition predicate, which also forces a second planInputPartitions(). Stopping that is an improvement, but it is default-on, untested, and unrelated to this feature — the same shape as the direct-reuse tightening you split out earlier. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:439]

[peter-toth]

Finding 4. This new BaseSubqueryExec subtype needs to be registered in SQLLastAttemptAccumulator, which matches the subtype set exhaustively and deliberately bails out on anything it doesn't know (SQLLastAttemptAccumulator.scala:365-384, untouched by this PR):

case sl: BaseSubqueryExec => sl match {
  case s: SubqueryExec => scopeIds(s.child)
  case _: SubqueryBroadcastExec =>
    // Used by DPP filter only, not part of main flow of query execution.
    Nil
  case _: SubqueryAdaptiveBroadcastExec =>
    // Used by DPP filter only.
    Nil
  case r: ReusedSubqueryExec => recurse(r.child)
  case p =>
    // Bail out if future unknown implementation is encountered.
    bailOutReason = Some(s"Unsupported BaseSubqueryExec: ${p.getClass.getName}")
    Nil
}

extractStageRDDScopes ends with p.subqueries.flatMap(recurse), and AdaptiveSparkPlanHelper.flatMap applies the function to the root node first (foreach(p) calls f(p) before descending), so the ProjectedBroadcastValueSubqueryExec itself falls into case p and sets bailOutReason. Once that is set, extractStageRDDScopes returns Left(...) and lastAttemptValueForQueryExecution — and lastAttemptValueForDataset through it — returns None for the whole query, with an "Unable to extract RDD scopes from query execution plan" warning. unexpectedLastAttemptMetricOperation is called with invalidate = false and no exception, so it only logs: it fails quietly and no existing test notices — SQLLastAttemptMetricPlanShapesSuite covers lastAttemptValueForDataset per plan shape but has no DPP shape.

Both DPP siblings are already listed there, so the existing SubqueryBroadcastExec/SubqueryAdaptiveBroadcastExec paths keep working and only the new node breaks it. Fix is the same shape as its neighbours:

        case _: ProjectedBroadcastValueSubqueryExec =>
          // Used by DPP filter only.
          Nil

[sunchao]

Fixed in 4db1868f, thanks @peter-toth. SQLLastAttemptAccumulator.extractStageRDDScopes now handles ProjectedBroadcastValueSubqueryExec alongside SubqueryBroadcastExec, returning Nil for both DPP-only subqueries instead of taking the unsupported-subquery bailout. The existing projected-broadcast regression now also asserts lastAttemptValueForDataset(df) == Some(0L) after confirming the projected subquery is present; it runs against V1, V2, and runtime-filtering V2 scans with AQE both enabled and disabled.

[peter-toth]

Finding 5. This line changes behaviour on the pre-existing, default-on iterative V2 pushdown path (SPARK-55596), independently of the new feature. (The isResultUnavailable case above it is essential — without it translateRuntimeFilterV2 would hit values().getOrElse { throw internalError }. And the case TrueLiteral => None you added in translateRuntimeFilterV2 is behaviour-neutral, since base's case other already returned None after a warning. This one is different: it actually changes what gets pushed.)

DynamicPruningExpression(Literal.TrueLiteral) is what PlanDynamicPruningFilters/PlanAdaptiveDynamicPruningFilters emit whenever onlyInBroadcast holds and no broadcast can be reused — the default with spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=true. On base it survives all the way into a pushed predicate:

• pushRuntimeFilters keeps it as a candidate: translateRuntimeFilterV2 returns None for it, and TrueLiteral.references is empty so f.references.subsetOf(filterAttrs) is trivially true;
• createRuntimePartitionPredicates maps it to Some(TrueLiteral) through case DynamicPruningExpression(e);
• in createPartitionPredicates, getPartitionFiltersAndDataFilters first puts it in dataFilters (it requires references.nonEmpty), but then extractPredicatesWithinOutputSet pulls it straight back as an extraPartitionFilter — predicates.scala:266-271 returns Some(other) when other.references.subsetOf(outputSet), which an empty reference set always satisfies;
• isPushablePartitionFilter accepts it (deterministic, no subquery, no PythonUDF), and PartitionPredicateImpl.apply returns Some because it only rejects unmatched references (PartitionPredicateImpl.scala:105-117).

So today a degenerate DPP filter pushes a PartitionPredicate wrapping true down to the connector and sets partPredicatesPushed = true, which makes replanWithRuntimeFilters re-run scan.toBatch.planInputPartitions() for no gain. Dropping it is a real improvement — but it is a silent change to released, default-on behaviour, with no test and nothing in the description: the same shape as the direct-reuse tightening @dongjoon-hyun asked you to separate out.

DataSourceV2EnhancedRuntimePartitionFilterSuite already has everything needed to pin it — a DPP query with reuseBroadcastOnly=true and no reusable broadcast, asserting assertPushedPartitionPredicates(df, expectedCount = 0). Otherwise, please call it out in the description, or split it out (arguably under its own ticket, since it is a SPARK-55596 bug rather than part of SPARK-58265).

[sunchao]

Fixed in 4db1868f, thanks @peter-toth. I removed the DynamicPruningExpression(Literal.TrueLiteral) => None special case, preserving pre-existing iterative V2 behavior even when projected broadcast pruning is disabled. The isResultUnavailable guard remains, so unavailable projected domains still cannot reach connector filtering. The new disabled broadcast projection preserves existing no-op iterative V2 filtering regression checks the surviving true DPP filter, exactly one zero-reference partition predicate, all five original partitions, and unchanged results.

I also updated the PR description to clarify that all four settings are internal, the optimization is disabled by default, and there is no user-facing configuration or API, addressing @viirya's earlier concern.

[peter-toth]

Re-checked through 4db1868 — findings 4, 5 resolved, nothing regressed.

I traced both new tests against 549b6db to confirm they actually fail on the unfixed code rather than green-lighting there too: lastAttemptValueForDataset(df) returns None on 549b6db (the Unsupported BaseSubqueryExec bail-out) instead of Some(0L), and disabled broadcast projection preserves existing no-op iterative V2 filtering sees expectedCount = 0 there instead of the reference-free PartitionPredicate base pushes. All three V2 branches — available domain, unavailable domain, projection off — now have a test. The description's user-facing answer is corrected too, which closes @viirya's config-visibility point.

One new item, present since the first round.

Non-blocking

• 6. Dead canonicalized on BroadcastValueProjection (late catch): nothing calls it, and both holders deliberately clear the projection to None when canonicalizing, so it is unreachable and reads as if the projection took part in canonicalization. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BroadcastValueProjection.scala:34]

[peter-toth]

Finding 6. This canonicalized is unreachable — nothing in the tree calls it.

Both holders of a BroadcastValueProjection deliberately drop it when canonicalizing:

• DynamicPruningSubquery.canonicalized → copy(...)(None) (sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/DynamicPruning.scala:117)
• SubqueryAdaptiveBroadcastExec.doCanonicalize → copy(...)(None) (sql/core/src/main/scala/org/apache/spark/sql/execution/SubqueryAdaptiveBroadcastExec.scala:53)

and both are pinned by tests asserting broadcastValueProjection.isEmpty after canonicalized. Grepping BroadcastValueProjection across the tree turns up no .canonicalized call, so this member — and the QueryPlan import on line 20 that only it needs — is dead. Worth deleting rather than leaving: as written it suggests the projection participates in canonicalization, which is the opposite of the design the tests lock in.

[sunchao]

Fixed in afebab43d82, thanks @peter-toth. I removed the unused BroadcastValueProjection.canonicalized method and its now-unneeded QueryPlan import. Canonicalization continues to drop the transient projection from both existing holders. DynamicPruningSubquerySuite passed all 11 tests, and Catalyst scalastyle passed.

[sunchao]

Thanks for the review! Merged to master / branch-4.x