[SPARK-57424][SQL] Add First/Last to segment-tree window aggregate allowlist

[yadavay-amzn]

What changes were proposed in this pull request?

Add classOf[First] and classOf[Last] to WindowSegmentTree.EligibleAggregates,
routing First/Last window aggregates through the segment-tree path established
by SPARK-56546 (sliding) and SPARK-57220 (shrinking) instead of the legacy
O(N x W) sliding / O(N^2) shrinking frame implementations. No new frame class,
no new SQLConf, no dispatcher changes -- the existing dispatcher branches in
WindowEvaluatorFactoryBase already gate on eligibleForSegTree, which calls
WindowSegmentTree.isEligible.

Why are the changes needed?

First and Last were previously denylisted as "order-dependent". This was
over-conservative: order-dependence in row-traversal order is exactly what
WindowSegmentTree.query provides. The query walks left-to-right (left
partial -> full blocks ascending -> right partial; within a block,
queryDescend walks children in ascending index order). First.mergeExpressions
and Last.mergeExpressions are correct under that traversal -- they pick the
row-order extreme across any contiguous range. For IGNORE NULLS the same merge
is mode-agnostic: per-row updateExpressions only set valueSet=true on
non-null values, so a per-block partial of (null, false) for an all-NULL
block is correctly skipped when merged with a later non-null block.

JIRA: https://issues.apache.org/jira/browse/SPARK-57424

Does this PR introduce any user-facing change?

Yes -- when spark.sql.window.segmentTree.enabled=true, FIRST/LAST window
aggregates over sliding or shrinking ROWS/RANGE frames execute through the
segment-tree path instead of the legacy frame implementations. Same opt-in
conf (default off), same eligibility allowlist mechanism, same fallback below
minPartitionRows, same SQLMetrics. No public API changes.

How was this patch tested?

New tests, all differential against the legacy frame (segment-tree on vs off):

• WindowSegmentTreeAllowlistSuite: routing tests for
  first / last / first_ignore_nulls / last_ignore_nulls; the previous
  "first/last falls through" negative tests are flipped; the mixed-allowlist
  test now uses collect_list (still on the denylist).
• SegmentTreeWindowFunctionSuite: sliding First/Last respect-nulls and
  ignore-nulls, all-NULL columns in both modes, a stretches-of-consecutive-NULLs
  case for the IGNORE NULLS merge path, and a multi-block case (block size 16, a
  40-row partition with an all-NULL middle block, wide frame) that forces the
  cross-block left-to-right combine spine and the all-NULL block partial the
  correctness argument depends on.
• UnboundedFollowingSegmentTreeSuite: shrinking First/Last respect-nulls and
  ignore-nulls plus an all-NULL column boundary case.

Benchmark

FirstLastSegmentTreeWindowBenchmark (results checked in at
sql/core/benchmarks/FirstLastSegmentTreeWindowBenchmark-results.txt; Linux
x86_64, Intel Xeon Platinum 8259CL @ 2.50GHz, OpenJDK 17):

Sliding frame [-1000, +1000] at N=10K:

Aggregate	Naive	Segtree	Speedup
FIRST respect-nulls	439 ms	97 ms	4.5x
LAST respect-nulls	540 ms	86 ms	6.3x
FIRST ignore-nulls	535 ms	88 ms	6.1x
LAST ignore-nulls	729 ms	83 ms	8.8x

Shrinking frame [CURRENT ROW, UNBOUNDED FOLLOWING] at N=10K:

Aggregate	Naive	Segtree	Speedup
FIRST respect-nulls	2,190 ms	78 ms	28.0x
LAST respect-nulls	2,175 ms	86 ms	25.4x
FIRST ignore-nulls	2,433 ms	71 ms	34.5x
LAST ignore-nulls	2,887 ms	72 ms	39.9x

N-sweep on FIRST shrinking:

N	Naive	Segtree	Speedup
5K	584 ms	66 ms	8.8x
25K	13,473 ms	96 ms	140.3x
50K	53,593 ms	154 ms	347.5x
100K	--	224 ms	--

Naive at N=100K is omitted (extrapolated cost ~3-4 min/iter); segtree path
stays sub-second.

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

Yes.

[HyukjinKwon]

0 blocking, 1 non-blocking, 2 nits.
Clean, minimal, well-tested change — the correctness argument holds under a full trace of the combine and the differential oracle tests are meaningful. Only minor/optional items below.

Suggestions (1)

• FirstLastSegmentTreeWindowBenchmark.scala:43: benchmark added without a checked-in results file — see inline

Nits: 2 minor items — one inline (benchmark comment grammar), plus (non-inline) the PR body says "99 tests" while the commit message says "97 tests / 6 oracle tests"; actual added test() blocks are 7 in SegmentTreeWindowFunctionSuite and 5 in UnboundedFollowingSegmentTreeSuite. Purely cosmetic.

Verification

Traced the segment-tree combine end-to-end for First/Last. WindowSegmentTree.query merges strictly ascending — left partial (lowest rows) -> full blocks blo+1..bhi-1 ascending -> right partial, and within a block queryDescend walks children c=0..fanout ascending — so the left operand is always the lower-row-index accumulator and right the higher-row-index chunk. First.mergeExpressions = If(valueSet.left, first.left, first.right) therefore keeps the lowest-row value and Last.mergeExpressions = If(valueSet.right, last.right, last.left) the highest — result-equivalent to the legacy frame. Respect-nulls carries (null, true) for a null extreme; IGNORE NULLS sets valueSet only on non-null, so an all-NULL block partial (null, false) is transparently skipped by the merge. No plan operator is synthesized, so output nullability/dataType is unchanged (First/Last stay nullable=true). The oracle differential tests (segtree on vs off, MIN_PARTITION_ROWS=1) and the blockSize=16 / 40-row / all-NULL-middle-block test exercise the cross-block combine spine and all-NULL partial the rationale depends on.

[yadavay-amzn]

Thanks for reviewing @HyukjinKwon !

Added the benchmark file and fixed the other nits as well. Please take a look when you get a chance.

[yadavay-amzn]

Also requesting @cloud-fan@uros-b to take a look

[cloud-fan]

0 blocking, 0 non-blocking, 0 nits. Traced the segment-tree combine independently for First/Last and confirm the change is correct.

WindowSegmentTree.query merges strictly left-to-right (left partial over the lowest rows -> full blocks ascending -> right partial; within a block queryDescend walks children ascending), so First's If(valueSet.left, first.left, first.right) keeps the lowest-row value and Last's If(valueSet.right, last.right, last.left) the highest -- result-equivalent to the legacy frame. IGNORE NULLS sets valueSet only on non-null values, so an all-NULL block's (null, false) partial is skipped on merge; (null, false) is a two-sided identity for both merges. No plan operator is synthesized -- the segtree path reuses the same AggregateProcessor buffer schema as the legacy frame, so output nullability/dataType are unchanged (First/Last stay nullable=true).

The differential checkEquivalence oracle tests (segtree-on vs -off, MIN_PARTITION_ROWS=1) plus the blockSize=16 / 40-row / all-NULL-middle-block test genuinely exercise the cross-block combine spine and the all-NULL partial the rationale depends on, across both sliding and shrinking frames. Prior review items (benchmark results file, comment grammar, PR-body test count) all verified addressed in the current diff.

[cloud-fan]

Merge Summary:

• merged into master ed150f9
• merged into branch-4.x ad02486

Posted by merge_spark_pr.py

[yadavay-amzn]

Thanks for reviewing @cloud-fan and @HyukjinKwon