[BUG] PPL FIRST/LAST/TAKE appear to ignore a preceding sort on multi-shard indices

[mengweieric]

What is the bug?

On a multi-shard index, FIRST(), LAST(), and TAKE() appear not to preserve a preceding PPL sort.

The same query returns the expected values on a one-shard index. earliest() and latest() remain correct over the same documents.

Reproduction

Create identical one-shard and five-shard indices:

for SH in 1 5; do
  curl -XDELETE "localhost:9200/rp_$SH"
  curl -XPUT "localhost:9200/rp_$SH" -H 'Content-Type: application/json' -d "{
    \"settings\":{\"number_of_shards\":$SH,\"number_of_replicas\":0},
    \"mappings\":{\"properties\":{\"@timestamp\":{\"type\":\"date\"},\"txt\":{\"type\":\"keyword\"}}}}
  }"
  curl -XPOST "localhost:9200/rp_$SH/_bulk?refresh=true" \
    -H 'Content-Type: application/x-ndjson' --data-binary '
{"index":{"_id":"1"}}
{"@timestamp":"2024-01-01T10:00:00Z","txt":"a"}
{"index":{"_id":"2"}}
{"@timestamp":"2024-01-02T10:00:00Z","txt":"b"}
{"index":{"_id":"3"}}
{"@timestamp":"2024-01-03T10:00:00Z","txt":"c"}
'
done

Run against each index:

source=rp_N
| sort @timestamp
| stats FIRST(@timestamp), LAST(@timestamp), TAKE(@timestamp, 2),
        earliest(@timestamp), latest(@timestamp)

Observed results:

1 shard: FIRST=01-01  LAST=01-03  TAKE=[01-01,01-02]  earliest=01-01  latest=01-03
5 shards: FIRST=01-03  LAST=01-03  TAKE=[01-03,01-02]  earliest=01-01  latest=01-03

The five-shard result is repeatable, but does not follow the explicit timestamp ordering.

Existing integration-test failures

The existing alias-field tests show the same issue with different shard placement:

CalciteAliasFieldAggregationIT::testFirstWithAliasField
query:    source=%s | sort @timestamp | stats FIRST(@timestamp)
expected: 2024-01-01 10:00:00
actual:   2024-01-02 10:00:00

CalciteAliasFieldAggregationIT::testLastWithAliasField
query:    source=%s | sort @timestamp | stats LAST(@timestamp)
expected: 2024-01-03 10:00:00
actual:   2024-01-02 10:00:00

CalciteAliasFieldAggregationIT::testTakeWithAliasField
query:    source=%s | sort @timestamp | stats TAKE(@timestamp, 2)
expected: [2024-01-01, 2024-01-02]
actual:   [2024-01-01, 2024-01-03]

Expected behavior

After sort @timestamp, FIRST() and LAST() should return the first and last values in that order, independent of shard count.

For TAKE(), the documentation currently says result order is not guaranteed. However, it would be useful to clarify whether the preceding sort should at least determine which values are selected.

Possible implementation clue

This may be related to how these aggregations are pushed down as top_hits:

FIRST -> createTopHitsBuilder(..., 1,    true, false, null,   null)
TAKE  -> createTopHitsBuilder(..., size, true, false, null,   null)
LAST  -> createTopHitsBuilder(..., 1,    true, true,  "_doc", SortOrder.DESC)

FIRST and TAKE do not pass a sort key, while LAST uses _doc, which is not a globally comparable document-order key across shards. The preceding PPL collation may therefore not reach the generated top_hits aggregation. This is only a possible explanation and needs maintainer confirmation.

This also appears consistent with the limitation noted in #5537: adding an explicit sort does not stabilize stats first()/last()/take() on multi-shard execution.

Environment

• OpenSearch 3.7.0
• opensearch-sql plugin
• Single node
• One versus five primary shards
• No replicas

[mengweieric]

Root cause

The preceding sort is removed during optimization before aggregation pushdown. AggregateAnalyzer therefore receives no input collation.

The generated top_hits request cannot recover that ordering:

• FIRST and TAKE use no field sort, so distributed reduction falls back to score, shard ordinal, and Lucene document ID.
• LAST uses _doc DESC, which is shard-local rather than a global document order.

This makes results dependent on shard layout.

Possible fix

Use the existing dedup sort-hint approach as a precedent:

1. Capture the input collation when constructing the LogicalAggregate, before optimization removes the Sort.
2. Store field priority, direction, and null ordering in a generic aggregate-input sort hint.
3. Apply the order unchanged for FIRST, and reverse each key’s direction and null ordering for LAST.
4. Apply it to TAKE only if maintainers confirm that preceding sort determines selected membership.
5. Disable pushdown when the sort cannot be represented safely as an OpenSearch field sort.
6. Fix the non-pushdown path separately and validate FIRST/LAST non-null semantics.

OpenSearch already performs a global sort-aware top_hits merge when given comparable field-sort values, so no distributed-reduction redesign should be necessary for a total ordering. Behavior for tied keys and queries without an explicit sort still needs to be defined.

---

Deterministic before/after request evidence

The comparison used the same cluster, the same five-shard index, and the same three documents. Placement was intentionally adversarial: the newest document was forced onto shard 0 and the oldest onto shard 4, so shard-ordinal tie-breaking produces the wrong answer reliably rather than by chance. Only the plugin JAR changed between captures.

The user's intent survives into the logical plan

The following line is identical before and after:

LogicalSort(sort0=[$8], dir0=[ASC-nulls-first])

The requested ordering is therefore present in the logical plan. The pre-fix error occurs downstream, where that ordering is not encoded in the generated aggregation request.

Before: pre-fix baseline

Baseline commit: 5f5876dd0 (the parent of PR #5719).

Function	Logical aggregate	Generated top_hits sort	Result	Expected	Verdict
FIRST	FIRST($0)	Absent	2024-01-03	2024-01-01	Wrong
LAST	LAST($0)	[{"_doc":{"order":"desc"}}]	2024-01-03	2024-01-03	Correct only because of this placement
TAKE	TAKE($0, $1)	Absent	[01-03, 01-02]	[01-01, 01-02]	Wrong relative to the test expectation

FIRST emits top_hits without a sort key. LAST sorts on _doc, which is a shard-local Lucene document ID and has no global meaning across shards. Neither request carries the user's created_at ordering.

LAST happens to return the expected value because the newest document is on shard 0, so the fallback shard-ordinal tie-break selects it. Changing document placement can change the result.

After: PR #5719

A representative rewritten aggregate is:

LogicalAggregate(FIRST(`@timestamp`)=[FIRST_BY_SORT($0, $0, $1, $2, $1) FILTER $3])
  LogicalProject(@timestamp=[$8], $f1=[false], $f2=[true], $f3=[IS NOT NULL($8)])

The aggregate now carries the sort key, direction (false for ascending), null placement (true for nulls first), and IP-type flag. It also carries a non-null filter for the measure.

Function	Generated top_hits sort	Result	Expected	Verdict
FIRST	created_at ASC, missing: _first	2024-01-01	2024-01-01	Correct
LAST	created_at DESC, missing: _last	2024-01-03	2024-01-03	Correct
TAKE	Still absent	[01-03, 01-02]	[01-01, 01-02]	Unchanged

FIRST and LAST are also wrapped in an exists(created_at) filter aggregation, preserving their first/last-non-null contract during pushdown. The @timestamp alias is resolved to its concrete mapped field, created_at, in the generated DSL.

What this proves—and what it does not

To isolate the request behavior from plugin execution, the exact top_hits request generated for TAKE was replayed directly against the same index:

Direct OpenSearch request	Result
Exactly as generated, with no sort	[2024-01-03, 2024-01-02]
Identical request plus created_at ASC	[2024-01-01, 2024-01-02]

OpenSearch can merge field-sorted top_hits correctly across shards. For FIRST and LAST, the explicit-sort contract makes dropping the available collation a confirmed product defect; PR #5719 corrects that request generation.

For TAKE, the experiment isolates the technical cause of the shard-dependent membership: its request still omits the available collation. Whether that omission violates the product contract remains a maintainer decision, because the documentation says returned order is not guaranteed but does not define whether a preceding sort controls selected membership.

Execution-path cross-check

The capture above validates pushdown. The same adversarial fixture was also executed with pushdown disabled:

Execution path	FIRST	LAST	TAKE
Pushdown enabled	2024-01-01	2024-01-03	[01-03, 01-02]
Pushdown disabled	2024-01-01	2024-01-03	[01-01, 01-02]

FIRST and LAST agree and are correct on both paths. TAKE differs between paths on identical data. This path difference is concrete; whether it is contractually incorrect depends on the unresolved TAKE membership semantics.

Why the existing TAKE test can flap

testTakeWithAliasField indexes documents with auto-generated IDs ({"index":{}}), so routing and shard placement vary between runs. A favorable placement can make shard-ordinal tie-breaking accidentally return the expected values. The implementation behavior is deterministic for a fixed layout, but the fixture layout is not. FIRST and LAST are now independent of that placement because they carry a globally comparable field sort; TAKE is not.

Tie behavior is not part of this comparison

The fixture uses three distinct timestamps. Rows with equal complete sort tuples remain unordered by design and are not involved in this before/after result.