From d779cc2fec5c49e738489d7df8089c1cc089f9c0 Mon Sep 17 00:00:00 2001 From: Louis Chu Date: Thu, 30 Jul 2026 22:11:54 +0800 Subject: [PATCH] Emit list() for mvcombine so it runs on the analytics engine mvcombine lowered to Calcite ARRAY_AGG with an explicit IS NOT NULL filter. The analytics-engine (DataFusion) route cannot execute that shape: Calcite ARRAY_AGG infers a nullable-element ARRAY while the engine's array_agg operator infers a NOT NULL element, and the two cannot be reconciled during the aggregate rewrite. Switch performArrayAggAggregation to emit the existing PPL list() aggregate (PPLBuiltinOperators.LIST), which the analytics engine already supports. ListAggFunction drops null values internally, so the explicit IS NOT NULL filter is no longer needed. Behavior note: the combined field is now ARRAY (list() stringifies elements) and is capped at 100 values per group, matching list()'s documented contract. CalcitePPLMvCombineTest logical-plan and SparkSQL expectations updated for the LIST aggregate. Document the list()-based output (strings, 100-value cap, unordered) in mvcombine.md. Also update CalciteExplainIT explain_mvcombine.yaml fixtures (pushdown on/off) for the LIST plan (drop the IS NOT NULL projection; the identity EnumerableCalc is eliminated on the pushdown path). Signed-off-by: Louis Chu --- .../sql/calcite/CalciteRelNodeVisitor.java | 10 ++--- docs/user/ppl/cmd/mvcombine.md | 4 ++ .../calcite/explain_mvcombine.yaml | 9 ++-- .../explain_mvcombine.yaml | 8 ++-- .../ppl/calcite/CalcitePPLMvCombineTest.java | 41 ++++++++----------- 5 files changed, 33 insertions(+), 39 deletions(-) diff --git a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java index b7b52e16d82..77124136056 100644 --- a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java +++ b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java @@ -73,7 +73,6 @@ import org.apache.calcite.rex.RexVisitorImpl; import org.apache.calcite.rex.RexWindowBounds; import org.apache.calcite.sql.SqlKind; -import org.apache.calcite.sql.fun.SqlLibraryOperators; import org.apache.calcite.sql.fun.SqlStdOperatorTable; import org.apache.calcite.sql.fun.SqlTrimFunction; import org.apache.calcite.sql.type.ArraySqlType; @@ -4498,13 +4497,10 @@ private void performArrayAggAggregation( RelBuilder relBuilder, int targetIndex, String targetName, List groupExprs) { final RexNode targetRef = relBuilder.field(targetIndex); - final RexNode notNullTarget = relBuilder.isNotNull(targetRef); - + // list() rather than Calcite ARRAY_AGG: list() carries the ARRAY contract the + // analytics-engine (DataFusion) route supports, and drops nulls itself (no filter needed). final RelBuilder.AggCall aggCall = - relBuilder - .aggregateCall(SqlLibraryOperators.ARRAY_AGG, targetRef) - .filter(notNullTarget) - .as(targetName); + relBuilder.aggregateCall(PPLBuiltinOperators.LIST, targetRef).as(targetName); relBuilder.aggregate(relBuilder.groupKey(groupExprs), aggCall); } diff --git a/docs/user/ppl/cmd/mvcombine.md b/docs/user/ppl/cmd/mvcombine.md index 9b3e07399f2..c768f35da2a 100644 --- a/docs/user/ppl/cmd/mvcombine.md +++ b/docs/user/ppl/cmd/mvcombine.md @@ -108,6 +108,10 @@ Error: Query returned no data ======= +## Limitations + +Values are combined using the `list()` aggregation: the combined field is a multivalue array of **strings** (non-string values are rendered as strings), at most **100** values are retained per group, and the order of values within the array is not guaranteed. + ## Related commands - [`nomv`](nomv.md) -- Converts a multivalue field into a single-value string diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_mvcombine.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_mvcombine.yaml index c387d566e2c..7e33df0bbbf 100644 --- a/integ-test/src/test/resources/expectedOutput/calcite/explain_mvcombine.yaml +++ b/integ-test/src/test/resources/expectedOutput/calcite/explain_mvcombine.yaml @@ -2,11 +2,10 @@ calcite: logical: | LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT]) LogicalProject(state=[$0], city=[$1], age=[$2]) - LogicalAggregate(group=[{0, 1}], age=[ARRAY_AGG($2) FILTER $3]) - LogicalProject(state=[$7], city=[$5], age=[$8], $f3=[IS NOT NULL($8)]) + LogicalAggregate(group=[{0, 1}], age=[LIST($2)]) + LogicalProject(state=[$7], city=[$5], age=[$8]) CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]]) physical: | EnumerableLimit(fetch=[10000]) - EnumerableAggregate(group=[{0, 1}], age=[ARRAY_AGG($2) FILTER $3]) - EnumerableCalc(expr#0..2=[{inputs}], expr#3=[IS NOT NULL($t2)], proj#0..3=[{exprs}]) - CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[state, city, age]], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","_source":{"includes":["state","city","age"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)]) + EnumerableAggregate(group=[{0, 1}], age=[LIST($2)]) + CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[state, city, age]], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","_source":{"includes":["state","city","age"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)]) diff --git a/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_mvcombine.yaml b/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_mvcombine.yaml index 35f2d79e7c8..c411ff8889b 100644 --- a/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_mvcombine.yaml +++ b/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_mvcombine.yaml @@ -2,11 +2,11 @@ calcite: logical: | LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT]) LogicalProject(state=[$0], city=[$1], age=[$2]) - LogicalAggregate(group=[{0, 1}], age=[ARRAY_AGG($2) FILTER $3]) - LogicalProject(state=[$7], city=[$5], age=[$8], $f3=[IS NOT NULL($8)]) + LogicalAggregate(group=[{0, 1}], age=[LIST($2)]) + LogicalProject(state=[$7], city=[$5], age=[$8]) CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]]) physical: | EnumerableLimit(fetch=[10000]) - EnumerableAggregate(group=[{0, 1}], age=[ARRAY_AGG($2) FILTER $3]) - EnumerableCalc(expr#0..16=[{inputs}], expr#17=[IS NOT NULL($t8)], state=[$t7], city=[$t5], age=[$t8], $f3=[$t17]) + EnumerableAggregate(group=[{0, 1}], age=[LIST($2)]) + EnumerableCalc(expr#0..16=[{inputs}], state=[$t7], city=[$t5], age=[$t8]) CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]]) diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMvCombineTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMvCombineTest.java index 6e6460a2365..17fd75955e0 100644 --- a/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMvCombineTest.java +++ b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMvCombineTest.java @@ -85,14 +85,13 @@ public void testMvCombineBasic() { String expectedLogical = "LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n" + " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n" - + " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n" - + " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n" - + " LogicalFilter(condition=[=($0, 'basic')])\n" - + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; + + " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n" + + " LogicalFilter(condition=[=($0, 'basic')])\n" + + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; verifyLogical(root, expectedLogical); String expectedSparkSql = - "SELECT `case`, `ip`, ARRAY_AGG(`packets`) FILTER (WHERE `packets` IS NOT NULL) `packets`\n" + "SELECT `case`, `ip`, `LIST`(`packets`) `packets`\n" + "FROM `scott`.`MVCOMBINE_DATA`\n" + "WHERE `case` = 'basic'\n" + "GROUP BY `case`, `ip`\n" @@ -114,14 +113,13 @@ public void testMvCombineWithNullTargetValues() { String expectedLogical = "LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n" + " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n" - + " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n" - + " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n" - + " LogicalFilter(condition=[=($0, 'nulls')])\n" - + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; + + " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n" + + " LogicalFilter(condition=[=($0, 'nulls')])\n" + + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; verifyLogical(root, expectedLogical); String expectedSparkSql = - "SELECT `case`, `ip`, ARRAY_AGG(`packets`) FILTER (WHERE `packets` IS NOT NULL) `packets`\n" + "SELECT `case`, `ip`, `LIST`(`packets`) `packets`\n" + "FROM `scott`.`MVCOMBINE_DATA`\n" + "WHERE `case` = 'nulls'\n" + "GROUP BY `case`, `ip`\n" @@ -143,14 +141,13 @@ public void testMvCombineWithDelimOption_SplunkSyntaxOrder() { String expectedLogical = "LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n" + " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n" - + " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n" - + " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n" - + " LogicalFilter(condition=[=($0, 'basic')])\n" - + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; + + " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n" + + " LogicalFilter(condition=[=($0, 'basic')])\n" + + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; verifyLogical(root, expectedLogical); String expectedSparkSql = - "SELECT `case`, `ip`, ARRAY_AGG(`packets`) FILTER (WHERE `packets` IS NOT NULL) `packets`\n" + "SELECT `case`, `ip`, `LIST`(`packets`) `packets`\n" + "FROM `scott`.`MVCOMBINE_DATA`\n" + "WHERE `case` = 'basic'\n" + "GROUP BY `case`, `ip`\n" @@ -188,10 +185,9 @@ public void testMvCombineSingleRow() { String expectedLogical = "LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n" + " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n" - + " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n" - + " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n" - + " LogicalFilter(condition=[=($0, 'single')])\n" - + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; + + " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n" + + " LogicalFilter(condition=[=($0, 'single')])\n" + + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; verifyLogical(root, expectedLogical); } @@ -209,10 +205,9 @@ public void testMvCombineEmptyResult() { String expectedLogical = "LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n" + " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n" - + " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n" - + " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n" - + " LogicalFilter(condition=[=($0, 'no_such_case')])\n" - + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; + + " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n" + + " LogicalFilter(condition=[=($0, 'no_such_case')])\n" + + " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n"; verifyLogical(root, expectedLogical); }