Skip to content

Commit 5007125

Browse files
noChargernoCharger
committed
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<VARCHAR> (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. Signed-off-by: Louis Chu <lingzhichu.clz@gmail.com>
1 parent de95ffb commit 5007125

3 files changed

Lines changed: 27 additions & 30 deletions

File tree

‎core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java‎

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,6 @@
7373
import org.apache.calcite.rex.RexVisitorImpl;
7474
import org.apache.calcite.rex.RexWindowBounds;
7575
import org.apache.calcite.sql.SqlKind;
76-
import org.apache.calcite.sql.fun.SqlLibraryOperators;
7776
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
7877
import org.apache.calcite.sql.fun.SqlTrimFunction;
7978
import org.apache.calcite.sql.type.ArraySqlType;
@@ -4498,13 +4497,12 @@ private void performArrayAggAggregation(
44984497
RelBuilder relBuilder, int targetIndex, String targetName, List<RexNode> groupExprs) {
44994498

45004499
final RexNode targetRef = relBuilder.field(targetIndex);
4501-
final RexNode notNullTarget = relBuilder.isNotNull(targetRef);
4502-
4500+
// mvcombine collects each group's values into a multivalue field. Emit the PPL list()
4501+
// aggregate (PPLBuiltinOperators.LIST) instead of Calcite ARRAY_AGG so the plan carries the
4502+
// ARRAY<VARCHAR> type contract the analytics engine (DataFusion) route already supports;
4503+
// ListAggFunction drops nulls internally, so no explicit filter is needed.
45034504
final RelBuilder.AggCall aggCall =
4504-
relBuilder
4505-
.aggregateCall(SqlLibraryOperators.ARRAY_AGG, targetRef)
4506-
.filter(notNullTarget)
4507-
.as(targetName);
4505+
relBuilder.aggregateCall(PPLBuiltinOperators.LIST, targetRef).as(targetName);
45084506

45094507
relBuilder.aggregate(relBuilder.groupKey(groupExprs), aggCall);
45104508
}

‎docs/user/ppl/cmd/mvcombine.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,10 @@ Error: Query returned no data
108108

109109
=======
110110

111+
## Limitations
112+
113+
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.
114+
111115
## Related commands
112116

113117
- [`nomv`](nomv.md) -- Converts a multivalue field into a single-value string

‎ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMvCombineTest.java‎

Lines changed: 18 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -85,14 +85,13 @@ public void testMvCombineBasic() {
8585
String expectedLogical =
8686
"LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n"
8787
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n"
88-
+ " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n"
89-
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n"
90-
+ " LogicalFilter(condition=[=($0, 'basic')])\n"
91-
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
88+
+ " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n"
89+
+ " LogicalFilter(condition=[=($0, 'basic')])\n"
90+
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
9291
verifyLogical(root, expectedLogical);
9392

9493
String expectedSparkSql =
95-
"SELECT `case`, `ip`, ARRAY_AGG(`packets`) FILTER (WHERE `packets` IS NOT NULL) `packets`\n"
94+
"SELECT `case`, `ip`, `LIST`(`packets`) `packets`\n"
9695
+ "FROM `scott`.`MVCOMBINE_DATA`\n"
9796
+ "WHERE `case` = 'basic'\n"
9897
+ "GROUP BY `case`, `ip`\n"
@@ -114,14 +113,13 @@ public void testMvCombineWithNullTargetValues() {
114113
String expectedLogical =
115114
"LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n"
116115
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n"
117-
+ " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n"
118-
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n"
119-
+ " LogicalFilter(condition=[=($0, 'nulls')])\n"
120-
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
116+
+ " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n"
117+
+ " LogicalFilter(condition=[=($0, 'nulls')])\n"
118+
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
121119
verifyLogical(root, expectedLogical);
122120

123121
String expectedSparkSql =
124-
"SELECT `case`, `ip`, ARRAY_AGG(`packets`) FILTER (WHERE `packets` IS NOT NULL) `packets`\n"
122+
"SELECT `case`, `ip`, `LIST`(`packets`) `packets`\n"
125123
+ "FROM `scott`.`MVCOMBINE_DATA`\n"
126124
+ "WHERE `case` = 'nulls'\n"
127125
+ "GROUP BY `case`, `ip`\n"
@@ -143,14 +141,13 @@ public void testMvCombineWithDelimOption_SplunkSyntaxOrder() {
143141
String expectedLogical =
144142
"LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n"
145143
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n"
146-
+ " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n"
147-
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n"
148-
+ " LogicalFilter(condition=[=($0, 'basic')])\n"
149-
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
144+
+ " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n"
145+
+ " LogicalFilter(condition=[=($0, 'basic')])\n"
146+
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
150147
verifyLogical(root, expectedLogical);
151148

152149
String expectedSparkSql =
153-
"SELECT `case`, `ip`, ARRAY_AGG(`packets`) FILTER (WHERE `packets` IS NOT NULL) `packets`\n"
150+
"SELECT `case`, `ip`, `LIST`(`packets`) `packets`\n"
154151
+ "FROM `scott`.`MVCOMBINE_DATA`\n"
155152
+ "WHERE `case` = 'basic'\n"
156153
+ "GROUP BY `case`, `ip`\n"
@@ -188,10 +185,9 @@ public void testMvCombineSingleRow() {
188185
String expectedLogical =
189186
"LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n"
190187
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n"
191-
+ " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n"
192-
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n"
193-
+ " LogicalFilter(condition=[=($0, 'single')])\n"
194-
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
188+
+ " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n"
189+
+ " LogicalFilter(condition=[=($0, 'single')])\n"
190+
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
195191
verifyLogical(root, expectedLogical);
196192
}
197193

@@ -209,10 +205,9 @@ public void testMvCombineEmptyResult() {
209205
String expectedLogical =
210206
"LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])\n"
211207
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2])\n"
212-
+ " LogicalAggregate(group=[{0, 1}], packets=[ARRAY_AGG($2) FILTER $3])\n"
213-
+ " LogicalProject(case=[$0], ip=[$1], packets=[$2], $f3=[IS NOT NULL($2)])\n"
214-
+ " LogicalFilter(condition=[=($0, 'no_such_case')])\n"
215-
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
208+
+ " LogicalAggregate(group=[{0, 1}], packets=[LIST($2)])\n"
209+
+ " LogicalFilter(condition=[=($0, 'no_such_case')])\n"
210+
+ " LogicalTableScan(table=[[scott, MVCOMBINE_DATA]])\n";
216211
verifyLogical(root, expectedLogical);
217212
}
218213

0 commit comments

Comments
 (0)