ESQL: Pull OrderBy followed by InlineJoin on top of it (elastic#137648)

Add optimisation rule to pull `OrderBy` above `InlineJoin`. This is required since otherwise the `OrderBy` won't be moved out of the left hand-side of the join, ending up as a stand-alone node that can't be turned into an executable -- it'll just be rejected by the verification as an "unbounded sort". Since the `InlineJoin` is sort agnostic, the `OrderBy` can be moved upwards past it, ending up as a top `TopN`.

This doesn't entirely solve the issue of unbounded sorts, however: some nodes, such as `MV_EXPAND` or `LOOKUP JOIN` break the inbound sort, so pulling a `SORT` on top of them isn't possible. In this cases the query will still fail. This is not `INLINE JOIN` specific, though.

This is a revival of elastic#132417, but with a new approach on how to deal with the attributes used by `SORT`, but dropped or overshadowed by the time `INLINE JOIN` is reached. In this case, we'll add temporary internal attributes kept until the INLINE JOIN and dropped afterwards.

Related elastic#124715 (elastic#113727)
Related elastic#124721 (elastic#133120)

Author: bpintea

docs/changelog/137648.yaml

@@ -0,0 +1,5 @@
+pr: 137648
+summary: Pull `OrderBy` followed by `InlineJoin` on top of it
+area: ES|QL
+type: enhancement
+issues: []

x-pack/plugin/esql/qa/testFixtures/src/main/resources/inlinestats.csv-spec

@@ -399,24 +399,22 @@ emp_no:integer |avg_worked_seconds:long|avg_avg_worked_seconds:double|languages:
 10023          |330870342              |3.181719481E8                |null             |5                    |1                     |5                     |F
 ;
 
-// TODO: INLINE STATS unit test needed for this one
-// https://github.com/elastic/elasticsearch/issues/113727
-pushDownSort_To_LeftSideOnly-Ignore
-required_capability: inline_stats
+pushDownSort_To_LeftSideOnly
+required_capability: inline_stats_preceeded_by_sort
 
 from employees
 | sort emp_no
-| inline stats avg = avg(salary) by languages
+| inline stats avg = round(avg(salary)) by languages
 | limit 5
 | keep emp_no, avg, languages, gender
 ;
 
- emp_no:integer| avg:double    |languages:integer|gender:keyword     
-10001          |57305.0        |2                |M              
-10002          |46272.5        |5                |F              
-10003          |61805.0        |4                |M              
-10004          |46272.5        |5                |M              
-10005          |63528.0        |1                |M
+ emp_no:integer| avg:double    |languages:integer|gender:keyword
+10001          |48179.0        |2                |M
+10002          |41681.0        |5                |F
+10003          |47733.0        |4                |M
+10004          |41681.0        |5                |M
+10005          |50577.0        |1                |M
 ;
 
 byMultivaluedSimple
@@ -4416,6 +4414,331 @@ c:long
 1
 ;
 
+sortBeforeInlinestats1
+required_capability: inline_stats_preceeded_by_sort
+ 
+ROW salary = 12300, emp_no = 5, gender = "F"
+| EVAL salaryK = salary/1000
+| SORT salaryK DESC
+| INLINE STATS sum = SUM(salaryK) BY gender
+| KEEP emp_no
+;
+
+emp_no:integer
+5
+;
+
+sortBeforeInlinestats2
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| SORT emp_no
+| EVAL salaryK = salary/1000
+| INLINE STATS count = COUNT(*) BY salaryK
+| KEEP emp_no, still_hired, count
+| LIMIT 5
+;
+
+emp_no:integer |still_hired:boolean|count:long
+10001          |true               |1
+10002          |true               |3
+10003          |false              |2
+10004          |true               |2
+10005          |true               |1
+;
+
+// TODO: fails with: java.lang.AssertionError: expected no concrete indices without data node plan
+sortBeforeInlinestats3-Ignore
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| SORT languages DESC
+| EVAL salaryK = salary/1000
+| INLINE STATS count = COUNT(*) BY salaryK
+| SORT emp_no
+| INLINE STATS min = MIN(MV_COUNT(languages)) BY salaryK
+| KEEP emp_no, still_hired, count
+| LIMIT 5
+;
+
+emp_no:integer |still_hired:boolean|count:long
+10001          |true               |1
+10002          |true               |3
+10003          |false              |2
+10004          |true               |2
+10005          |true               |1
+;
+
+// same as `afterLookup`, swapped SORT position
+sortBeforeInlinestatsAndLookupJoin
+required_capability: inline_stats_preceeded_by_sort
+required_capability: join_lookup_v12
+
+FROM airports
+| EVAL backup_scalerank = scalerank
+| RENAME scalerank AS language_code
+// | SORT abbrev DESC // this would work here, but not in a FORK config (GenerativeForkIT), which disables column pruning
+| LOOKUP JOIN languages_lookup ON language_code
+| RENAME language_name as scalerank
+| DROP language_code
+| INLINE STATS count=COUNT(*) BY scalerank
+| SORT abbrev DESC
+| KEEP abbrev, *scalerank
+| LIMIT 5
+;
+
+abbrev:keyword |backup_scalerank:integer| scalerank:keyword
+null           |8                       |null
+null           |8                       |null
+null           |8                       |null
+ZRH            |3                       |Spanish
+ZNZ            |4                       |German
+;
+
+// same as `shadowingAggregateByNextGrouping`, swapped SORT position
+sortBeforeDoubleInlinestats
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| KEEP gender, languages, emp_no, salary
+| SORT emp_no
+| INLINE STATS gender = count_distinct(gender) BY languages
+| INLINE STATS avg(salary) BY gender
+| LIMIT 3
+;
+
+emp_no:integer |salary:integer |languages:integer|avg(salary):double|gender:long
+10001          |57305          |2                |48248.55          |2
+10002          |56371          |5                |48248.55          |2
+10003          |61805          |4                |48248.55          |2
+;
+
+inlineStatsAfterSortDoubled
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| SORT emp_no DESC // going to be dropped
+| INLINE STATS min = MIN(salary) BY languages
+| SORT salary ASC
+| INLINE STATS avg = AVG(salary) BY emp_no
+| WHERE emp_no > 10050
+| KEEP emp_no, avg, salary
+| LIMIT 5
+;
+
+emp_no:integer |avg:double |salary:integer
+10092          |25976.0    |25976
+10057          |27215.0    |27215
+10084          |28035.0    |28035
+10068          |28941.0    |28941
+10060          |29175.0    |29175
+;
+
+inlineStatsAfterSortShadowed
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| EVAL s1 = salary + 1
+| SORT s1
+| WHERE s1 > 50000
+| EVAL ls = languages::string
+| EVAL s1 = salary // s1 is shadowed, not dropped
+| INLINE STATS cd = COUNT_DISTINCT(ls)
+| WHERE emp_no < 10006
+| KEEP emp_no, ls, cd, s1
+;
+
+emp_no:integer |ls:keyword     |cd:long        |s1:integer
+10002          |5              |5              |56371
+10001          |2              |5              |57305
+10003          |4              |5              |61805
+10005          |1              |5              |63528
+;
+
+inlineStatsAfterSortDropped
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| EVAL s1 = salary + 1
+| SORT s1
+| WHERE s1 > 72000
+| DROP s1
+| INLINE STATS cd = COUNT_DISTINCT(languages)
+| KEEP emp_no, cd, languages, salary
+;
+
+emp_no:integer |cd:long        |languages:integer|salary:integer
+10099          |4              |2                |73578
+10019          |4              |1                |73717
+10027          |4              |null             |73851
+10007          |4              |4                |74572
+10045          |4              |3                |74970
+10029          |4              |null             |74999
+;
+
+mixedShadowingInlineStatsAfterSort
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| KEEP salary, emp_no, first_name
+| SORT salary, emp_no
+| INLINE STATS salary = COUNT(*) BY emp_no
+| LIMIT 5
+;
+
+first_name:keyword|salary:long    |emp_no:integer
+Guoxiang          |1              |10015
+null              |1              |10035
+Valdiodio         |1              |10092
+Florian           |1              |10048
+Ebbe              |1              |10057
+;
+
+inlineStatsWithAggExpressionAfterSortAndRename
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| KEEP emp_no, salary, languages
+| SORT emp_no
+| RENAME emp_no AS emp_idx
+| INLINE STATS sum = COUNT(salary) + COUNT(salary) BY languages
+| LIMIT 5
+;
+
+emp_idx:integer|salary:integer |sum:long       |languages:integer
+10001          |57305          |38             |2
+10002          |56371          |42             |5
+10003          |61805          |36             |4
+10004          |36174          |42             |5
+10005          |63528          |30             |1
+;
+
+inlineStatsAfterEnrichAndSort
+required_capability: inline_stats_preceeded_by_sort
+required_capability: enrich_load
+
+FROM employees
+| KEEP emp_no, first_name, salary, languages
+| EVAL language_name = first_name // key in this test is to have a field overwritten by the enrich, and sorting by it before
+| SORT language_name
+| EVAL language_code = languages::KEYWORD
+| ENRICH languages_policy
+| RENAME salary AS sal
+| INLINE STATS cd = COUNT(emp_no) BY languages
+| LIMIT 7
+;
+
+emp_no:integer |first_name:keyword|sal:integer    |language_code:keyword | language_name:keyword |cd:long        |languages:integer
+10059          |Alejandro         |44307          |2                     |French                 |19             |2
+10091          |Amabile           |38645          |3                     |Spanish                |17             |3
+10006          |Anneke            |60335          |3                     |Spanish                |17             |3
+10062          |Anoosh            |65030          |3                     |Spanish                |17             |3
+10094          |Arumugam          |66817          |5                     |null                   |21             |5
+10049          |Basil             |37853          |5                     |null                   |21             |5
+10058          |Berhard           |38376          |3                     |Spanish                |17             |3
+;
+
+inlineStatsAfterSortDropped
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| KEEP salary, emp_no, first_name, gender
+| EVAL old_salary = salary
+| SORT old_salary, emp_no
+| DROP old_salary
+| INLINE STATS salary = COUNT(*) BY gender
+| LIMIT 5
+;
+
+emp_no:integer |first_name:keyword |salary:long    |gender:keyword
+10015          |Guoxiang           |10             |null
+10035          |null               |57             |M
+10092          |Valdiodio          |33             |F
+10048          |Florian            |57             |M
+10057          |Ebbe               |33             |F
+;
+
+inlineStatsAfterSortDroppedAndResort
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| KEEP salary, emp_no, first_name, gender
+| EVAL old_salary = salary
+| SORT old_salary, emp_no
+| DROP old_salary
+| EVAL salary = salary / 1000
+| RENAME salary AS s
+| SORT s, first_name
+| INLINE STATS s = COUNT(*) BY gender
+| LIMIT 5
+;
+
+emp_no:integer |first_name:keyword |s:long         |gender:keyword
+10015          |Guoxiang           |10             |null
+10092          |Valdiodio          |33             |F
+10035          |null               |57             |M
+10048          |Florian            |57             |M
+10057          |Ebbe               |33             |F
+;
+
+inlineStatsDoubledAfterSortDropped
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| EVAL s1 = salary + 1
+| SORT s1
+| WHERE s1 > 72000
+| DROP s1
+| INLINE STATS sum = SUM(languages)
+| INLINE STATS cd = COUNT_DISTINCT(languages)
+| KEEP emp_no, cd, languages, salary, sum
+;
+
+emp_no:integer |cd:long        |languages:integer|salary:integer |sum:long
+10099          |4              |2                |73578          |10
+10019          |4              |1                |73717          |10
+10027          |4              |null             |73851          |10
+10007          |4              |4                |74572          |10
+10045          |4              |3                |74970          |10
+10029          |4              |null             |74999          |10
+;
+
+inlineStatsAfterStats
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| KEEP salary, emp_no, first_name, gender
+| SORT salary
+| STATS salary = MAX(salary) BY gender
+| INLINE STATS s = COUNT(*) BY gender
+| LIMIT 5
+| SORT salary
+;
+
+salary:integer |s:long         |gender:keyword     
+73717          |1              |null           
+74572          |1              |F              
+74999          |1              |M              
+;
+
+inlineStatsAfterStatsAndSort
+required_capability: inline_stats_preceeded_by_sort
+
+FROM employees
+| KEEP salary, emp_no, first_name, gender
+| SORT salary
+| STATS salary = MAX(salary) BY gender
+| SORT salary DESC
+| INLINE STATS s = COUNT(*) BY gender
+| LIMIT 5
+;
+
+salary:integer |s:long         |gender:keyword     
+74999          |1              |M              
+74572          |1              |F              
+73717          |1              |null           
+;
+
 fixClassCastBugWithSeveralCounts
 required_capability: inline_stats
 required_capability: fix_stats_classcast_exception

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -983,6 +983,11 @@ public enum Cap {
          */
         INLINE_STATS,
 
+        /**
+         * Added support for having INLINE STATS preceded by a SORT clause, now executable in certain cases.
+         */
+        INLINE_STATS_PRECEEDED_BY_SORT,
+
         /**
          * Support partial_results
          */

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizer.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.DeduplicateAggs;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.ExtractAggregateCommonFilter;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.FoldNull;
+import org.elasticsearch.xpack.esql.optimizer.rules.logical.HoistOrderByBeforeInlineJoin;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.HoistRemoteEnrichLimit;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.HoistRemoteEnrichTopN;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.LiteralsOnTheRight;
@@ -225,6 +226,7 @@ protected static Batch<LogicalPlan> operators() {
             new PruneRedundantSortClauses(),
             new PruneLeftJoinOnNullMatchingField(),
             new PruneInlineJoinOnEmptyRightSide(),
+            new HoistOrderByBeforeInlineJoin(),
             new PruneEmptyAggregates()
         );
     }