[AURON #2351] Reduce native broadcast task serialization size

[Sigma-Ma]

Which issue does this PR close?

Closes #2351

Rationale for this change

NativeBroadcastExchangeBase.doExecuteBroadcast returned an anonymous Broadcast
that captured the transformed relation. Spark serialized that wrapper with tasks,
causing the task binary size to grow with the build-side data.

What changes are included in this PR?

• Return a real Spark broadcast for the transformed relation instead of a value-capturing wrapper.
• Keep the existing native IPC broadcast path unchanged.
• Add regression coverage for mixed native/Spark execution and serialized broadcast size.

Are there any user-facing changes?

No. This only changes the internal broadcast handling.

How was this patch tested?

./build/mvn -Pspark-3.5 -Pscala-2.12 -pl spark-extension-shims-spark test -Dsuites=org.apache.auron.AuronCheckConvertBroadcastExchangeSuite -Dsuffixes=NoDiscoveredSuites

./build/mvn -Pspark-3.5 -Pscala-2.12 -pl spark-extension-shims-spark -am install -DskipTests -DskipBuildNative

./dev/reformat --check

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

• Yes
• No

Generated-by: OpenAI Codex (GPT-5)

ASF guidance: https://www.apache.org/legal/generative-tooling.html

[wForget]

This isn’t an optimal solution, as it performs two broadcasts. But it’s still better than the current approach because it avoids serializing the broadcast value in every task.

Could you add a comment explaining this?

[Sigma-Ma]

Agreed. Mixed native/Spark execution still needs a separate JVM broadcast because the two paths consume different representations. I’ve added a comment to make that trade-off clear.

[slfan1989]

Thanks for fixing this.

Replacing the value-capturing wrapper with a real Spark broadcast addresses the oversized task serialization issue, and the regression test covers both correctness and serialized size. I agree that a short comment should be added to explain why a second broadcast is intentionally required for mixed native/Spark execution. No blocking concerns otherwise.

LGTM.

[lyne7-sc]

should we cache the Spark-side broadcast here, similar to BroadcastExchangeExec.relationFuture.

It looks like executeBroadcast() may be called more than once. For example, BroadcastHashJoinExec can call it through both prepareBroadcast() and needCopyResult.

BroadcastExchangeExec avoids rebuilding the broadcast by memoizing the relation. Here, each call to doExecuteBroadcast() materializes and transforms the native data again, then creates a new Spark broadcast with a new broadcast ID.

so should we keep a single cached future/broadcast per NativeBroadcastExchangeExec?

[Sigma-Ma]

executeBroadcast() can be called more than once, so I’ve cached the Spark-side broadcast as a transient lazy value. Each exchange now materializes it only once.

[lyne7-sc]

Since a.payload is not referenced, Catalyst may prune it before the broadcast exchange, leaving only a.key in the broadcast relation. As a result, the serialized relation may remain small enough for this assertion to pass even with the previous value-capturing implementation, so the test may not reliably cover the intended regression.

Should we keep payload in the query output here?

[Sigma-Ma]

Catalyst can prune payload in the original query. I’ve kept it in the output and updated the expected rows, so the test now covers the full build-side payload.

[Sigma-Ma]

Thanks for fixing this.

Replacing the value-capturing wrapper with a real Spark broadcast addresses the oversized task serialization issue, and the regression test covers both correctness and serialized size. I agree that a short comment should be added to explain why a second broadcast is intentionally required for mixed native/Spark execution. No blocking concerns otherwise.

LGTM.

Thanks for the review. I’ll add the comment and also cached the Spark-side broadcast per exchange.