Extend rewrite to dc / distinct_count via canonical-name translation

PPL eventstats accepts three aliases for the cardinality aggregation —
dc, distinct_count, distinct_count_approx — all resolving to
BuiltinFunctionName.DISTINCT_COUNT_APPROX. The stats command only
accepts distinct_count_approx, so AGGREGATION_FUNC_MAPPING registers
only that name; the other two are window-only aliases in
WINDOW_FUNC_MAPPING.

The previous predicate required intersection of both maps, which
rejected dc and distinct_count. They fell through to the legacy
RexOver lowering — which is the exact buggy "EnumerableWindow over a
row-fetching scan" shape opensearch-project#5483 was filed against. Fix was incomplete.

Replace the intersection check with: name is in ofWindowFunction AND
its canonical aggregation name (BuiltinFunctionName.name().toLowerCase,
e.g. "distinct_count_approx") is in ofAggregation. Translate the same
way in stripWindowFunctionForAggregate so aggVisitor sees the
registered name. For names already in both maps (count/sum/avg/etc.)
the canonical name equals the user-typed name, so the lookup is a
no-op — no behavior change for the cases that already worked.

ROW_NUMBER still falls through because its canonical name "row_number"
isn't in the aggregation map. Same for percentile / take / first /
last / median / list / values — all rejected by the canonical-name
lookup.

Verified locally:
- eventstats dc(state)               → cardinality agg, size:0
- eventstats distinct_count(state) by gender
  → composite over gender + nested cardinality on state, size:0

Regenerated explain_eventstats_dc.json and
explain_eventstats_distinct_count.json with the new shape (composite
size 1000 to match CI). Both tests are pushdown-only
(enabledOnlyWhenPushdownIsEnabled() + loadFromFile hardcoded to
calcite/), so no calcite_no_pushdown/ variants needed.

Signed-off-by: Jialiang Liang <jiallian@amazon.com>

Author: RyanL1997

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

@@ -39,6 +39,7 @@
 import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -2246,9 +2247,19 @@ private static boolean canRewriteWindowAsAggregateJoin(Window node) {
         return false;
       }
       String funcName = extractAggregateFunctionName(wf.getFunction());
-      if (funcName == null
-          || BuiltinFunctionName.ofWindowFunction(funcName).isEmpty()
-          || BuiltinFunctionName.ofAggregation(funcName).isEmpty()) {
+      if (funcName == null) {
+        return false;
+      }
+      // The user-typed name must resolve to a window function, and the windowable
+      // {@link BuiltinFunctionName}'s canonical name (e.g. {@code distinct_count_approx} for the
+      // {@code dc} / {@code distinct_count} aliases) must also be registered as a regular
+      // aggregation so {@code aggVisitor} can resolve it for the right-side LogicalAggregate.
+      Optional<BuiltinFunctionName> windowName = BuiltinFunctionName.ofWindowFunction(funcName);
+      if (windowName.isEmpty()) {
+        return false;
+      }
+      String canonical = canonicalAggregationName(windowName.get());
+      if (BuiltinFunctionName.ofAggregation(canonical).isEmpty()) {
         return false;
       }
       if (!wf.getSortList().isEmpty()) {
@@ -2279,6 +2290,19 @@ private static String extractAggregateFunctionName(UnresolvedExpression fn) {
     return null;
   }
 
+  /**
+   * Maps a windowable {@link BuiltinFunctionName} to the function name registered in {@code
+   * BuiltinFunctionName.AGGREGATION_FUNC_MAPPING}. Most names match directly (e.g. {@code COUNT} →
+   * {@code "count"}), but the {@code dc} and {@code distinct_count} aliases are window-only — they
+   * map to {@link BuiltinFunctionName#DISTINCT_COUNT_APPROX}, whose registered aggregation name is
+   * {@code "distinct_count_approx"}. Translating to the canonical name lets the right-side
+   * aggregate go through {@code aggVisitor}'s standard {@link BuiltinFunctionName#ofAggregation}
+   * lookup with no special case for the alias.
+   */
+  private static String canonicalAggregationName(BuiltinFunctionName windowName) {
+    return windowName.name().toLowerCase(Locale.ROOT);
+  }
+
   /**
    * A bare, simple (non-dotted) field reference. Dotted paths like {@code doc.user.city} are
    * rejected here because the join condition references the partition key by field name via {@code
@@ -2321,12 +2345,19 @@ private UnresolvedExpression stripWindowFunctionForAggregate(UnresolvedExpressio
         return fn;
       }
       if (fn instanceof Function f) {
+        // Translate window-only aliases (e.g. `dc`, `distinct_count`) to the canonical
+        // aggregation name (`distinct_count_approx`) so aggVisitor's ofAggregation lookup
+        // resolves them. For names already in the aggregation map (count, sum, avg, max, min,
+        // stddev*, var*, earliest, latest, distinct_count_approx, pattern), this is a no-op.
+        String funcName =
+            BuiltinFunctionName.ofWindowFunction(f.getFuncName())
+                .map(CalciteRelNodeVisitor::canonicalAggregationName)
+                .orElse(f.getFuncName());
         List<UnresolvedExpression> args = f.getFuncArgs();
         UnresolvedExpression field = args.isEmpty() ? null : args.get(0);
         List<UnresolvedExpression> argList =
             args.size() <= 1 ? List.of() : args.subList(1, args.size());
-        AggregateFunction agg = new AggregateFunction(f.getFuncName(), field, argList);
-        return agg;
+        return new AggregateFunction(funcName, field, argList);
       }
       return fn;
     }

integ-test/src/test/resources/expectedOutput/calcite/explain_eventstats_dc.json

@@ -1,6 +1,6 @@
 {
   "calcite": {
-    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], distinct_states=[DISTINCT_COUNT_APPROX($7) OVER ()])\n    CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
-    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableWindow(window#0=[window(aggs [DISTINCT_COUNT_APPROX($7)])])\n    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"account_number\",\"firstname\",\"address\",\"balance\",\"gender\",\"city\",\"employer\",\"state\",\"age\",\"email\",\"lastname\"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
+    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], distinct_states=[$17])\n    LogicalJoin(condition=[true], joinType=[inner])\n      CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n      LogicalAggregate(group=[{}], distinct_states=[DISTINCT_COUNT_APPROX($0)])\n        LogicalProject(state=[$7])\n          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
+    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableNestedLoopJoin(condition=[true], joinType=[inner])\n    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"account_number\",\"firstname\",\"address\",\"balance\",\"gender\",\"city\",\"employer\",\"state\",\"age\",\"email\",\"lastname\"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#48:LogicalAggregate.NONE.[](input=RelSubset#47,group={},distinct_states=DISTINCT_COUNT_APPROX($0))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"aggregations\":{\"distinct_states\":{\"cardinality\":{\"field\":\"state.keyword\"}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
   }
 }

integ-test/src/test/resources/expectedOutput/calcite/explain_eventstats_distinct_count.json

@@ -1,6 +1,6 @@
 {
   "calcite": {
-    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], distinct_states=[DISTINCT_COUNT_APPROX($7) OVER (PARTITION BY $4)])\n    CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
-    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableWindow(window#0=[window(partition {4} aggs [DISTINCT_COUNT_APPROX($7)])])\n    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"account_number\",\"firstname\",\"address\",\"balance\",\"gender\",\"city\",\"employer\",\"state\",\"age\",\"email\",\"lastname\"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
+    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], distinct_states=[$18])\n    LogicalJoin(condition=[IS NOT DISTINCT FROM($4, $17)], joinType=[inner])\n      CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n      LogicalAggregate(group=[{0}], distinct_states=[DISTINCT_COUNT_APPROX($1)])\n        LogicalProject(gender=[$4], state=[$7])\n          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
+    "physical": "EnumerableCalc(expr#0..12=[{inputs}], proj#0..10=[{exprs}], distinct_states=[$t12])\n  EnumerableLimit(fetch=[10000])\n    EnumerableNestedLoopJoin(condition=[IS NOT DISTINCT FROM($4, $11)], joinType=[inner])\n      CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"account_number\",\"firstname\",\"address\",\"balance\",\"gender\",\"city\",\"employer\",\"state\",\"age\",\"email\",\"lastname\"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n      CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#163:LogicalAggregate.NONE.[](input=RelSubset#162,group={0},distinct_states=DISTINCT_COUNT_APPROX($1))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"aggregations\":{\"composite_buckets\":{\"composite\":{\"size\":1000,\"sources\":[{\"gender\":{\"terms\":{\"field\":\"gender.keyword\",\"missing_bucket\":true,\"missing_order\":\"first\",\"order\":\"asc\"}}}]},\"aggregations\":{\"distinct_states\":{\"cardinality\":{\"field\":\"state.keyword\"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
   }
 }