Support reverse command with Calcite

[selsong]

Description

Implement the reverse command in PPL to flip the result order of records. This PR only includes the grammar implementation, pushdown optimization will be raised in a 2nd followup PR.

Syntax

<source> | [commands] | reverse | [commands]

• reverse takes no arguments.
• It can appear anywhere in the pipeline including the end, and will reverse the order of rows produced up to that point.
• Unlike sort, which allows field-based sorting, reverse simply inverts the row order as is.

Semantics

Expected Behavior:

• Action: Reverses the display order of all rows in the current result set
• Scope: Operates on the entire result set at the point where reverse appears in the pipeline
• Data Preservation: Does not modify field values or schema

Implementation Approach:
Assigns sequential numbers to each row, then sorts by those numbers in descending order to achieve proper reversal regardless of existing sort operations.

Example Queries

-- Reverse original ingestion order
source=accounts | reverse

-- Reverse after sorting by age (youngest to oldest → oldest to youngest)
source=accounts | sort age | reverse

-- Get the last 5 records in the original order
source=accounts | reverse | head 5

-- Reverse after filtering and projecting fields
source=accounts | fields name, age | reverse

Restrictions / Limitations

• Calcite Engine Only: Requires plugins.query.calcite.enabled=true
• Legacy Engine: Throws UnsupportedOperationException when Calcite is disabled
• Memory Overhead: Uses row numbering and sort which may impact performance on very large result sets

Performance Test

• Dataset: big5 (1.16B documents, 257GB)
• Test Date: Fri Jul 25 2025

Baseline Performance

• head 1: Avg = 25ms, P90 = 26ms (39 QPS)
• head 10: Avg = 26ms, P90 = 28ms
• head 100: Avg = 32ms, P90 = 33ms

Reverse Command

• reverse of head 1: Avg ≈ 38ms, P90 = 40ms
• reverse of head 10: Avg ≈ 38ms, P90 = 40ms
• reverse of head 100: Avg ≈ 44ms, P90 = 45ms
• reverse of head 9000 - 10000: ~800ms

Note:

• reverse should only be used on smaller datasets.
• Pushdown of reverse is not yet implemented, so passing in a large, unfiltered dataset directly into reverse like source=big5 | reverse | head 10 processes the entire dataset. If the dataset size passed into reverse is over 10,000 it can cause timeout by hitting the circuitbreaker. [BUG] Circuit Breaker Triggered by Reverse Operation on Large Datasets #3925

Related Issues

#3873

Check List

• New functionality includes testing.
• New functionality has been documented.
• New functionality has javadoc added.
• New functionality has a user manual doc added.
• API changes companion pull request created.
• Commits are signed per the DCO using --signoff.
• Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[LantaoJin]

@selsong Thanks for starting contribution. But I doubt that we need the reverse command for right now since
Update:

1. For new PPL commands, a Feature or RFC issue is required for syntax review before pull request.
2. This restriction can be related. PPL as an unified query language, new commands and functions are required to be implemented both in PPL-on-OpenSearch and PPL-on-Spark. But this won't be a problem if [FEATURE] Export PPL-Calcite engine as reusable library #3734 is done. Before that, new commands should be planned in release to avoid fragmented user experience. The current workaround is implementing it in two repositories.
3. reverse could cause performance crisis when handle large dataset. A pushdown or restriction solution should be considered. To unblock new command, performance improvement could be separated to a follow-up PR.

[yuancu]

Thank you for the contribution! The implementation is well structured and thoroughly tested. I left a few comments for the discussion of my concerns.

[yuancu]

Will the the digest of a logical sort ever contain row_number? You projected it as another column, it will refer it with its position instead of name.

I tested with the index bank with the ppl: source=bank | reverse | fields address. The digest I get of this logical sort is LogicalSort.NONE.[20 DESC](input=RelSubset#2967,sort0=$20,dir0=DESC). The 20 corresponds to the added column in the following logical plan:

LogicalProject(address=[$2])
  LogicalSort(sort0=[$20], dir0=[DESC])
    LogicalProject(account_number=[$0], firstname=[$1], address=[$2], birthdate=[$3], date_value=[$4], gender=[$5], city=[$6], lastname=[$7], balance=[$8], employer=[$9], state=[$10], age=[$11], email=[$12], male=[$13], _id=[$14], _index=[$15], _score=[$16], _maxscore=[$17], _sort=[$18], _routing=[$19], __reverse_row_num__=[ROW_NUMBER() OVER ()])
      CalciteLogicalIndexScan(table=[[OpenSearch, bank]])

The digest doesn't contain the name of the column. You need another way to examine whether it's a reverse.

[yuancu]

In the RFC #3873 , you mentioned the following case can be pushed down:

The optimization is triggered when the logical plan has this exact shape:

LogicalSort (reverse sort)
   CalciteLogicalIndexScan

...
If reverse immediately follows the source, pushdown to a descending index scan results in efficient queries.

However, in practice, they are not pushed down. For example, the physical plan of source=bank | reverse is as follows:

EnumerableCalc(expr#0..14=[{inputs}], proj#0..13=[{exprs}])
  EnumerableSort(sort0=[$14], dir0=[DESC])
    EnumerableWindow(window#0=[window(rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
      CalciteEnumerableIndexScan(table=[[OpenSearch, bank]], PushDownContext=[[PROJECT->[account_number, firstname, address, birthdate, date_value, gender, city, lastname, balance, employer, state, age, email, male]], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","_source":{"includes":["account_number","firstname","address","birthdate","date_value","gender","city","lastname","balance","employer","state","age","email","male"],"excludes":[]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])

Moreover, I doubt whether can it be pushed down. You added a new column to implement reverse. However, the new column isn't present in the datasource. Unless we support pushing down window function of row_number(), in which case we can push down logical sort after row number altogether, there is no way for the OpenSearch engine to sort by this non-existing column.

Nevertheless, I think it can be pushed down when there are preceding sort -- in this way, you can not rely on the added __row_number__ column, but on the field to reverse the order of the results.

After fixing this, please kindly add tests in CalciteExplainIT to examine whether the generated physical plan is as expected.

[penghuo]

In PPL module, we usually test the parser, and verifyPPLToSparkSQL.
correctness teast should move to IT.

[penghuo]

Could u add a test in ExplainIT for reverse command?

[penghuo]

@yuancu @LantaoJin please take another look.

[yuancu]

Looks good to me :)

[LantaoJin]

LGTM except the explain IT.

[LantaoJin]

Missing test in PPLQueryDataAnonymizerTest

[LantaoJin]

IMO, reverse by row_number should only apply to the plan (RelNode) which doesn't contain collation. For example:
source=t | ... | sort - age | reverse should apply sort by age asc, rather than row_number.
You can try:

RelCollation collation = context.relBuilder.peek().getTraitSet().getCollation();
if (collation == null || collation == RelCollations.EMPTY) {
  // Add ROW_NUMBER() column
    RexNode rowNumber =
   ...
} else {
  RelCollation reversedCollation = reverseCollation(collation);
  context.relBuilder.sort(reversedCollation)
}

Add reverseCollation() to PlanUtils.class

public static RelCollation reverseCollation(RelCollation original) {
  if (original == null || original.getFieldCollations().isEmpty()) {
        return original;
    }

    List<RelFieldCollation> reversedFields = new ArrayList<>();
    for (RelFieldCollation field : original.getFieldCollations()) {
        RelFieldCollation.Direction reversedDirection = 
            field.direction.reverse();
        
        RelFieldCollation reversedField = new RelFieldCollation(
            field.getFieldIndex(),
            reversedDirection,
            field.nullDirection
        );
        reversedFields.add(reversedField);
    }

    return RelCollations.of(reversedFields);
}

[LantaoJin]

Then add some tests:

source=t | ... | sort - age | reverse
source=t | ... | sort - age, + gender | reverse
source=t | ... | sort - age | reverse | reverse
source=t | ... | sort - age, + gender | reverse  | reverse

[penghuo]

@LantaoJin @selsong is it correct issue? or performance imporvement?
If it is performance improvement, we can track by issue #3924. And raise 2nd PR.

[selsong]

Thanks for the suggestion! This is a performance improvement, and it's already tracked under issue #3924. The current implementation still produces correct results, but applying reverseCollation where applicable would avoid unnecessary computation. I'll address this enhancement in a follow-up PR.

[LantaoJin]

I don't think my suggestion is a pushdown enhancement, but is okey to refactor in separate PR.