enable pushdown distinct count approx

Signed-off-by: xinyual <xinyual@amazon.com>

Author: xinyual

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java

@@ -619,6 +619,19 @@ public void testExplainOnAggregationWithSumEnhancement() throws IOException {
                 TEST_INDEX_BANK)));
   }
 
+  @Test
+  public void testStatsDistinctCountApproxFunctionExplainWithPushDown() throws IOException {
+    enabledOnlyWhenPushdownIsEnabled();
+    String query =
+        "source=opensearch-sql_test_index_account | stats distinct_count_approx(state) as"
+            + " distinct_states by gender";
+    var result = explainQueryToString(query);
+    String expected =
+        loadFromFile(
+            "expectedOutput/calcite/explain_agg_with_distinct_count_approx_enhancement.json");
+    assertJsonEqualsIgnoreId(expected, result);
+  }
+
   @Test
   public void testExplainRegexMatchInWhereWithScriptPushdown() throws IOException {
     enabledOnlyWhenPushdownIsEnabled();

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

@@ -0,0 +1,6 @@
+{
+  "calcite":{
+    "logical":"LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(distinct_states=[$1], gender=[$0])\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":"CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},distinct_states=DISTINCT_COUNT_APPROX($1)), PROJECT->[distinct_states, gender], LIMIT->10000], 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"
+  }
+}

opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java

@@ -5,6 +5,8 @@
 
 package org.opensearch.sql.opensearch.executor;
 
+import static org.opensearch.sql.expression.function.BuiltinFunctionName.DISTINCT_COUNT_APPROX;
+
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.sql.PreparedStatement;
@@ -55,8 +57,6 @@
 import org.opensearch.sql.planner.physical.PhysicalPlan;
 import org.opensearch.sql.storage.TableScanOperator;
 
-import static org.opensearch.sql.expression.function.BuiltinFunctionName.DISTINCT_COUNT_APPROX;
-
 /** OpenSearch execution engine implementation. */
 public class OpenSearchExecutionEngine implements ExecutionEngine {
   private static final Logger logger = LogManager.getLogger(OpenSearchExecutionEngine.class);

opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java

@@ -495,11 +495,11 @@ yield switch (functionName) {
             }
             yield Pair.of(aggBuilder, new SinglePercentileParser(aggFieldName));
           }
-          case DISTINCT_COUNT_APPROX ->  Pair.of(
-                  helper.build(
-                          !args.isEmpty() ? args.getFirst() : null,
-                          AggregationBuilders.cardinality(aggFieldName)),
-                  new SingleValueParser(aggFieldName));
+          case DISTINCT_COUNT_APPROX -> Pair.of(
+              helper.build(
+                  !args.isEmpty() ? args.getFirst() : null,
+                  AggregationBuilders.cardinality(aggFieldName)),
+              new SingleValueParser(aggFieldName));
           default -> throw new AggregateAnalyzer.AggregateAnalyzerException(
               String.format("Unsupported push-down aggregator %s", aggCall.getAggregation()));
         };