[FEATURE] Support Limit pushdown through aggregation.

[penghuo]

Is your feature request related to a problem?

  @Test
  public void testFilterAndAggPushDownExplain() throws IOException {
    String expected = loadExpectedPlan("explain_filter_agg_push.json");
    assertJsonEqualsIgnoreId(
        expected,
        explainQueryToString(
            "source=opensearch-sql_test_index_account"
                + "| where age > 30 "
                + "| stats avg(age) AS avg_age by state, city"));
  }

• Output

{
  "calcite": {
    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(avg_age=[$2], state=[$0], city=[$1])\n    LogicalAggregate(group=[{0, 1}], avg_age=[AVG($2)])\n      LogicalProject(state=[$7], city=[$5], age=[$8])\n        LogicalFilter(condition=[>($8, 30)])\n          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableCalc(expr#0..2=[{inputs}], avg_age=[$t2], state=[$t0], city=[$t1])\n    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[city, state, age], FILTER->>($2, 30), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 1},avg_age=AVG($2))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"query\":{\"range\":{\"age\":{\"from\":30,\"to\":null,\"include_lower\":false,\"include_upper\":true,\"boost\":1.0}}},\"_source\":{\"includes\":[\"city\",\"state\",\"age\"],\"excludes\":[]},\"sort\":[{\"_doc\":{\"order\":\"asc\"}}],\"aggregations\":{\"composite_buckets\":{\"composite\":{\"size\":1000,\"sources\":[{\"state\":{\"terms\":{\"field\":\"state.keyword\",\"missing_bucket\":true,\"missing_order\":\"first\",\"order\":\"asc\"}}},{\"city\":{\"terms\":{\"field\":\"city.keyword\",\"missing_bucket\":true,\"missing_order\":\"first\",\"order\":\"asc\"}}}]},\"aggregations\":{\"avg_age\":{\"avg\":{\"field\":\"age\"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
  }
}

What solution would you like?
The LogicalSystemLimit(fetch=[10000] should push down to aggregation. e.g. {\"composite\":{\"size\":10000,\"

What alternatives have you considered?
n/a

Do you have any additional context?
n/a

[qianheng-aws]

We restrict the max bucket size by 1000 via hard code. Do we plan to increase it if the fetch size in limit operator exceeds 1000?

v2:

sql/opensearch/src/main/java/org/opensearch/sql/opensearch/storage/script/aggregation/AggregationQueryBuilder.java

Line 48 in 5316c0a

public static final int AGGREGATION_BUCKET_SIZE = 1000;

Calcite:

sql/opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java

Line 88 in 5316c0a

public static final int AGGREGATION_BUCKET_SIZE = 1000;

[LantaoJin]

For aggregation query, size will be 0. Push down the LogicalSystemLimit(fetch=[10000]) to size won't get any help for query result excepts performance downgrade. right?

[qianheng-aws]

For aggregation query, size will be 0. Push down the LogicalSystemLimit(fetch=[10000]) to size won't get any help for query result excepts performance downgrade. right?

@LantaoJin I think we need to push down the limit into AggregationBuilder instead of QueryBuilder. AggregationBuilder has its own size to determine the final output rows