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); }