[SPARK-58578][SQL] Mark ApplyFunctionExpression stateful to prevent reusedRow data races under concurrent evaluation

[MrHappyEnding]

What changes were proposed in this pull request?

Mark ApplyFunctionExpression as stateful and add a regression test verifying
that it produces a fresh expression instance.

Why are the changes needed?

Sibling QueryExecutions derived from the same base DataFrame can share the
same expression instance. When they execute concurrently, the mutable
reusedRow can therefore be overwritten by another evaluator.

Does this PR introduce any user-facing change?

Yes. It prevents incorrect results during concurrent evaluation of
ApplyFunctionExpression. There is no public API change.

How was this patch tested?

Added ApplyFunctionExpressionSuite and ran:

build/sbt 'catalyst/testOnly *ApplyFunctionExpressionSuite'
build/sbt 'catalyst/testOnly *DataSourceV2FunctionSuite'
dev/lint-scala

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

Generated-by: Codex (GPT-5)

[uros-b]

Thank you @MrHappyEnding! Adding @peter-toth for further review.

[peter-toth]

Thanks for the PR, @MrHappyEnding!

ApplyFunctionExpression keeps a reusable SpecificInternalRow on the instance (reusedRow, there since SPARK-34981), so it needs stateful = true for freshCopyIfContainsStatefulExpression() to give each evaluator its own copy. I traced the four callers of that method — ExpressionsEvaluator.prepareExpressions (so every interpreted projection, including the one ConvertToLocalRelation builds), CodeGenerator:1379, InterpretedOrdering's right-side evaluators, and ConstantFolding.tryFold — and the copy does get a fresh row, since withNewChildrenInternal hands the new instance an uninitialized lazy val. foldable stays safe because tryFold fresh-copies before evaluating, and nothing in the tree branches on stateful other than that mechanism, so there's no subexpression-elimination or planning side effect. Both new tests fail when I revert just the one-line change, and the style matches the two sibling sub-tasks that already landed (SPARK-58204, SPARK-58205). The code looks right to me; two documentation-level items below.

For whoever merges: both merged siblings went to master, branch-4.x and branch-4.3 (#57354 as c2685ecf98e / bfb95152ba9 / 9251fab63e8), and this is the same class of silent-wrong-results fix, so it probably wants the same three branches.

Non-blocking

• 1. the "Why" section never says when two evaluators come to share one instance: reachability is the only thing that really needs judging here, and SPARK-58578 already has it — "when sibling QueryExecutions derived from the same base DataFrame run concurrently, they share the same expression instance". Worth lifting that into the PR body, with a pointer to the SPARK-58154 umbrella and the two sub-tasks already fixed this way, so the next reader doesn't have to reconstruct it.

Minor

• 2. the second test's name claims more than it checks: it pins that two freshCopyIfContainsStatefulExpression() results don't share reusedRow, not that concurrent evaluation is safe in general.

[peter-toth]

Finding 2. The name reads as "concurrent evaluation of this expression is safe", but the test constructs the two evaluators itself with freshCopyIfContainsStatefulExpression(), so what it actually pins is that two fresh copies don't share reusedRow. The query-level property needs one more step that isn't exercised here — each concurrent execution building its own projection — so a reader who trusts the name could conclude more than the test shows.

Suggest matching the name to the assertion:

Suggested change

	test("SPARK-58578: concurrent evaluation uses independent input rows") {
	test("SPARK-58578: fresh copies do not share the reused input row") {

Worth keeping the test as it is otherwise — I reverted just the stateful override and it fails exactly as intended, with firstResult coming back as 2 because both evaluators are then the same instance.

[MrHappyEnding]

Thanks for the review! I clarified how the expression instance can be shared in the PR description and renamed the test to match what it actually verifies.