Make addcoltotals/addtotals summary row backend-deterministic

addcoltotals (and addtotals with col=true) lowers to UNION ALL(data,
single SUM row). SQL UNION ALL has no ordering guarantee: the v2 engine
preserves input order incidentally, but the DataFusion backend on the
analytics-engine route does not, so the summary row landed in a
non-deterministic position. The command contract and user docs require the
summary row to be last.

Separately, the synthesized column for a new `labelfield` was typed as a
bounded VARCHAR(n), which renders as a Substrait VarChar(n) literal that the
DataFusion substrait consumer rejects ("Unsupported literal_type: VarChar").

Both fixes are in CalciteRelNodeVisitor.buildAddRowTotalAggregate:
- Tag each union branch with a constant ordering key (0 = data rows,
  1 = summary row), sort on it, then drop the helper column, so the summary
  row is deterministically last regardless of backend. Reuses the existing
  streamstats sort-then-projectExcept idiom; adds PlanUtils constant
  ORDER_COLUMN_FOR_ADDCOLTOTALS.
- Type the synthesized label column as an unbounded VARCHAR, matching how
  OpenSearchTypeFactory maps STRING fields.

Regenerated the affected CalcitePPLAddColTotalsTest / CalcitePPLAddTotalsTest
plan and Spark-SQL expectations and the explain_add_col_totals /
explain_add_totals YAML plans; result rows are unchanged.

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

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

@@ -16,6 +16,7 @@
 import static org.opensearch.sql.ast.tree.Sort.SortOrder.DESC;
 import static org.opensearch.sql.calcite.plan.rule.PPLDedupConvertRule.buildDedupNotNull;
 import static org.opensearch.sql.calcite.plan.rule.PPLDedupConvertRule.buildDedupOrNull;
+import static org.opensearch.sql.calcite.utils.PlanUtils.ORDER_COLUMN_FOR_ADDCOLTOTALS;
 import static org.opensearch.sql.calcite.utils.PlanUtils.ROW_NUMBER_COLUMN_FOR_MAIN;
 import static org.opensearch.sql.calcite.utils.PlanUtils.ROW_NUMBER_COLUMN_FOR_RARE_TOP;
 import static org.opensearch.sql.calcite.utils.PlanUtils.ROW_NUMBER_COLUMN_FOR_STREAMSTATS;
@@ -3370,13 +3371,12 @@ public RelNode buildAddRowTotalAggregate(
     RelNode originalData = context.relBuilder.peek();
     List<String> fieldNames = originalData.getRowType().getFieldNames();
     boolean foundLabelField = false;
-    int labelLength =
-        (labelField != null) && (labelField.length() > label.length())
-            ? labelField.length()
-            : label.length();
 
+    // Type the synthesized label column as an unbounded VARCHAR, matching how OpenSearchTypeFactory
+    // maps STRING fields. A length-bounded VARCHAR(n) would render as a Substrait VarChar(n)
+    // literal, which the DataFusion backend (analytics-engine route) does not support.
     RelDataType labelVarcharType =
-        context.relBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR, labelLength);
+        context.relBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR);
 
     // If no specific fields specified, use all numeric fields
     if (fieldsToAggregate.isEmpty()) {
@@ -3483,10 +3483,24 @@ public RelNode buildAddRowTotalAggregate(
       // Project the totals row with proper field order and labels
       context.relBuilder.project(selectList);
       RelNode totalsRow = context.relBuilder.build();
-      // 4. Union original data with totals row
+      // 4. Union original data with totals row.
+      // UNION ALL does not guarantee row order, and some backends (e.g. DataFusion on the
+      // analytics-engine route) do not preserve input order, so the summary row can land
+      // anywhere. Tag each branch with a constant ordering key (0 = data rows, 1 = summary row),
+      // union, sort on it so the summary row is deterministically last, then drop the helper.
       context.relBuilder.push(originalData);
+      context.relBuilder.projectPlus(
+          context.relBuilder.alias(context.relBuilder.literal(0), ORDER_COLUMN_FOR_ADDCOLTOTALS));
+      RelNode dataWithOrder = context.relBuilder.build();
       context.relBuilder.push(totalsRow);
-      context.relBuilder.union(true); // Use UNION ALL to preserve order
+      context.relBuilder.projectPlus(
+          context.relBuilder.alias(context.relBuilder.literal(1), ORDER_COLUMN_FOR_ADDCOLTOTALS));
+      RelNode totalsWithOrder = context.relBuilder.build();
+      context.relBuilder.push(dataWithOrder);
+      context.relBuilder.push(totalsWithOrder);
+      context.relBuilder.union(true); // UNION ALL
+      context.relBuilder.sort(context.relBuilder.field(ORDER_COLUMN_FOR_ADDCOLTOTALS));
+      context.relBuilder.projectExcept(context.relBuilder.field(ORDER_COLUMN_FOR_ADDCOLTOTALS));
     }
     return context.relBuilder.peek();
   }

core/src/main/java/org/opensearch/sql/calcite/utils/PlanUtils.java

@@ -90,6 +90,12 @@ public interface PlanUtils {
   String ROW_NUMBER_COLUMN_FOR_TRANSPOSE = "_row_number_transpose_";
   String VALUE_COLUMN_FOR_TRANSPOSE = "_value_transpose_";
 
+  /**
+   * Ordering key for the addcoltotals UNION ALL: 0 on the original rows, 1 on the summary row, so a
+   * sort on it keeps the summary row last regardless of backend union ordering.
+   */
+  String ORDER_COLUMN_FOR_ADDCOLTOTALS = "_addcoltotals_order_";
+
   static SpanUnit intervalUnitToSpanUnit(IntervalUnit unit) {
     return switch (unit) {
       case MICROSECOND -> SpanUnit.MICROSECOND;

integ-test/src/test/resources/expectedOutput/calcite/explain_add_col_totals.yaml

@@ -2,17 +2,21 @@ calcite:
   logical: |
     LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
       LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10])
-        LogicalUnion(all=[true])
-          LogicalSort(fetch=[5])
-            CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
-          LogicalProject(account_number=[null:BIGINT], firstname=[null:VARCHAR], address=[null:VARCHAR], balance=[$0], gender=[null:VARCHAR], city=[null:VARCHAR], employer=[null:VARCHAR], state=[null:VARCHAR], age=[$1], email=[null:VARCHAR], lastname=[null:VARCHAR], _id=[null:VARCHAR], _index=[null:VARCHAR], _score=[null:REAL], _maxscore=[null:REAL], _sort=[null:BIGINT], _routing=[null:VARCHAR])
-            LogicalAggregate(group=[{}], balance=[SUM($3)], age=[SUM($8)])
+        LogicalSort(sort0=[$17], dir0=[ASC])
+          LogicalUnion(all=[true])
+            LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], _id=[$11], _index=[$12], _score=[$13], _maxscore=[$14], _sort=[$15], _routing=[$16], _addcoltotals_order_=[0])
               LogicalSort(fetch=[5])
                 CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
+            LogicalProject(account_number=[null:BIGINT], firstname=[null:VARCHAR], address=[null:VARCHAR], balance=[$0], gender=[null:VARCHAR], city=[null:VARCHAR], employer=[null:VARCHAR], state=[null:VARCHAR], age=[$1], email=[null:VARCHAR], lastname=[null:VARCHAR], _id=[null:VARCHAR], _index=[null:VARCHAR], _score=[null:REAL], _maxscore=[null:REAL], _sort=[null:BIGINT], _routing=[null:VARCHAR], _addcoltotals_order_=[1])
+              LogicalAggregate(group=[{}], balance=[SUM($3)], age=[SUM($8)])
+                LogicalSort(fetch=[5])
+                  CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
   physical: |
-    EnumerableLimit(fetch=[10000])
-      EnumerableUnion(all=[true])
-        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname], LIMIT->5], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":5,"timeout":"1m","_source":{"includes":["account_number","firstname","address","balance","gender","city","employer","state","age","email","lastname"]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])
-        EnumerableCalc(expr#0..1=[{inputs}], expr#2=[null:BIGINT], expr#3=[null:VARCHAR], account_number=[$t2], firstname=[$t3], address=[$t3], balance=[$t0], gender=[$t3], city=[$t3], employer=[$t3], state=[$t3], age=[$t1], email=[$t3], lastname=[$t3])
-          EnumerableAggregate(group=[{}], balance=[SUM($0)], age=[SUM($1)])
-            CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[balance, age], LIMIT->5], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":5,"timeout":"1m","_source":{"includes":["balance","age"]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])
+    EnumerableCalc(expr#0..11=[{inputs}], proj#0..10=[{exprs}])
+      CalciteEnumerableTopK(sort0=[$11], dir0=[ASC], fetch=[10000])
+        EnumerableUnion(all=[true])
+          EnumerableCalc(expr#0..10=[{inputs}], expr#11=[0], proj#0..11=[{exprs}])
+            CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname], LIMIT->5], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":5,"timeout":"1m","_source":{"includes":["account_number","firstname","address","balance","gender","city","employer","state","age","email","lastname"]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])
+          EnumerableCalc(expr#0..1=[{inputs}], expr#2=[null:BIGINT], expr#3=[null:VARCHAR], expr#4=[1], account_number=[$t2], firstname=[$t3], address=[$t3], balance=[$t0], gender=[$t3], city=[$t3], employer=[$t3], state=[$t3], age=[$t1], email=[$t3], lastname=[$t3], _addcoltotals_order_=[$t4])
+            EnumerableAggregate(group=[{}], balance=[SUM($0)], age=[SUM($1)])
+              CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[balance, age], LIMIT->5], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":5,"timeout":"1m","_source":{"includes":["balance","age"]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])

integ-test/src/test/resources/expectedOutput/calcite/explain_add_totals.yaml

@@ -2,21 +2,22 @@ calcite:
   logical: |
     LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
       LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], CustomSum=[$17], all_emp_total=[$18])
-        LogicalUnion(all=[true])
-          LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], _id=[$11], _index=[$12], _score=[$13], _maxscore=[$14], _sort=[$15], _routing=[$16], CustomSum=[+($3, $8)], all_emp_total=[null:VARCHAR(13)])
-            LogicalSort(fetch=[5])
-              CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
-          LogicalProject(account_number=[null:BIGINT], firstname=[null:VARCHAR], address=[null:VARCHAR], balance=[$0], gender=[null:VARCHAR], city=[null:VARCHAR], employer=[null:VARCHAR], state=[null:VARCHAR], age=[$1], email=[null:VARCHAR], lastname=[null:VARCHAR], _id=[null:VARCHAR], _index=[null:VARCHAR], _score=[null:REAL], _maxscore=[null:REAL], _sort=[null:BIGINT], _routing=[null:VARCHAR], CustomSum=[null:BIGINT], all_emp_total=['ColTotal':VARCHAR(13)])
-            LogicalAggregate(group=[{}], balance=[SUM($0)], age=[SUM($1)])
-              LogicalProject(balance=[$3], age=[$8])
-                LogicalSort(fetch=[5])
-                  CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
+        LogicalSort(sort0=[$19], dir0=[ASC])
+          LogicalUnion(all=[true])
+            LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], _id=[$11], _index=[$12], _score=[$13], _maxscore=[$14], _sort=[$15], _routing=[$16], CustomSum=[+($3, $8)], all_emp_total=[null:VARCHAR], _addcoltotals_order_=[0])
+              LogicalSort(fetch=[5])
+                CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
+            LogicalProject(account_number=[null:BIGINT], firstname=[null:VARCHAR], address=[null:VARCHAR], balance=[$0], gender=[null:VARCHAR], city=[null:VARCHAR], employer=[null:VARCHAR], state=[null:VARCHAR], age=[$1], email=[null:VARCHAR], lastname=[null:VARCHAR], _id=[null:VARCHAR], _index=[null:VARCHAR], _score=[null:REAL], _maxscore=[null:REAL], _sort=[null:BIGINT], _routing=[null:VARCHAR], CustomSum=[null:BIGINT], all_emp_total=['ColTotal':VARCHAR], _addcoltotals_order_=[1])
+              LogicalAggregate(group=[{}], balance=[SUM($0)], age=[SUM($1)])
+                LogicalProject(balance=[$3], age=[$8])
+                  LogicalSort(fetch=[5])
+                    CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
   physical: |
-    EnumerableLimit(fetch=[10000])
-      EnumerableUnion(all=[true])
-        EnumerableCalc(expr#0..10=[{inputs}], expr#11=[+($t3, $t8)], expr#12=[null:VARCHAR(13)], proj#0..12=[{exprs}])
-          CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname], LIMIT->5, LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":5,"timeout":"1m","_source":{"includes":["account_number","firstname","address","balance","gender","city","employer","state","age","email","lastname"]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])
-        EnumerableLimit(fetch=[10000])
-          EnumerableCalc(expr#0..1=[{inputs}], expr#2=[null:BIGINT], expr#3=[null:VARCHAR], expr#4=['ColTotal':VARCHAR(13)], account_number=[$t2], firstname=[$t3], address=[$t3], balance=[$t0], gender=[$t3], city=[$t3], employer=[$t3], state=[$t3], age=[$t1], email=[$t3], lastname=[$t3], CustomSum=[$t2], all_emp_total=[$t4])
+    EnumerableCalc(expr#0..13=[{inputs}], proj#0..12=[{exprs}])
+      CalciteEnumerableTopK(sort0=[$13], dir0=[ASC], fetch=[10000])
+        EnumerableUnion(all=[true])
+          EnumerableCalc(expr#0..10=[{inputs}], expr#11=[+($t3, $t8)], expr#12=[null:VARCHAR], expr#13=[0], proj#0..13=[{exprs}])
+            CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname], LIMIT->5], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":5,"timeout":"1m","_source":{"includes":["account_number","firstname","address","balance","gender","city","employer","state","age","email","lastname"]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])
+          EnumerableCalc(expr#0..1=[{inputs}], expr#2=[null:BIGINT], expr#3=[null:VARCHAR], expr#4=['ColTotal':VARCHAR], expr#5=[1], account_number=[$t2], firstname=[$t3], address=[$t3], balance=[$t0], gender=[$t3], city=[$t3], employer=[$t3], state=[$t3], age=[$t1], email=[$t3], lastname=[$t3], CustomSum=[$t2], all_emp_total=[$t4], _addcoltotals_order_=[$t5])
             EnumerableAggregate(group=[{}], balance=[SUM($0)], age=[SUM($1)])
               CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[balance, age], LIMIT->5], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":5,"timeout":"1m","_source":{"includes":["balance","age"]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])