Push down aggregation on text field without .keyword sub-field

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

@@ -2366,9 +2366,11 @@ public void testDedupWithExpr() throws IOException {
   }
 
   @Test
-  public void testDedupTextTypeNotPushdown() throws IOException {
+  public void testDedupTextTypePushdown() throws IOException {
+    // A text field with no .keyword sub-field is aggregated by reading its value from _source via
+    // a Calcite script; dedup therefore pushes down as a composite terms + top_hits aggregation.
     enabledOnlyWhenPushdownIsEnabled();
-    String expected = loadExpectedPlan("explain_dedup_text_type_no_push.yaml");
+    String expected = loadExpectedPlan("explain_dedup_text_type_push.yaml");
     assertYamlEqualsIgnoreId(
         expected, explainQueryYaml(String.format("source=%s | dedup email", TEST_INDEX_BANK)));
   }

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

@@ -85,6 +85,26 @@ public void testTimechartWithCustomTimeField() throws IOException {
     verifyDataRows(result, rows("2017-01-01 00:00:00", 2), rows("2018-01-01 00:00:00", 5));
   }
 
+  @Test
+  public void testTimechartAvgByTextHost() throws IOException {
+    // `host` is mapped as text with no .keyword sub-field, so this timechart pushes down via the
+    // text-field aggregation path (composite terms(script over _source) + date_histogram, with
+    // avg(cpu_usage) as a numeric metric on the composite).
+    JSONObject result = executeQuery("source=events | timechart span=1m avg(cpu_usage) by host");
+    verifySchema(
+        result,
+        schema("@timestamp", "timestamp"),
+        schema("host", "string"),
+        schema("avg(cpu_usage)", "double"));
+    verifyDataRows(
+        result,
+        rows("2024-07-01 00:00:00", "web-01", 45.2),
+        rows("2024-07-01 00:01:00", "web-02", 38.7),
+        rows("2024-07-01 00:02:00", "web-01", 55.3),
+        rows("2024-07-01 00:03:00", "db-01", 42.1),
+        rows("2024-07-01 00:04:00", "web-02", 41.8));
+  }
+
   @Test
   public void testTimechartWithMinuteSpanNoGroupBy() throws IOException {
     JSONObject result = executeQuery("source=events | timechart span=1m avg(cpu_usage)");

integ-test/src/test/java/org/opensearch/sql/ppl/DedupCommandIT.java

@@ -58,6 +58,27 @@ public void testAllowMoreDuplicates() throws IOException {
     verifyDataRows(result, rows(true), rows(true), rows(false), rows(false));
   }
 
+  @Test
+  public void testDedupOnTextField() throws IOException {
+    // `email` is mapped as text with no .keyword sub-field, so dedup runs via the text-field
+    // aggregation pushdown path (composite terms + top_hits with the field read from _source).
+    // Assert not just the dedup key set but also the associated projected columns per row, so
+    // the top_hits round-trip is exercised end-to-end.
+    JSONObject result =
+        executeQuery(
+            String.format(
+                "source=%s | dedup email | fields email, firstname, balance", TEST_INDEX_BANK));
+    verifyDataRows(
+        result,
+        rows("amberduke@pyrami.com", "Amber JOHnny", 39225),
+        rows("hattiebond@netagy.com", "Hattie", 5686),
+        rows("nanettebates@quility.com", "Nanette", 32838),
+        rows("daleadams@boink.com", "Dale", 4180),
+        rows("elinorratliff@scentric.com", "Elinor", 16418),
+        rows("virginiaayala@filodyne.com", "Virginia", 40540),
+        rows("dillardmcpherson@quailcom.com", "Dillard", 48086));
+  }
+
   @Test
   public void testKeepEmptyDedup() throws IOException {
     JSONObject result =

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

@@ -25,15 +25,11 @@ calcite:
           EnumerableCalc(expr#0..4=[{inputs}], expr#5=[IS NULL($t1)], expr#6=['nil'], expr#7=[10], expr#8=[<=($t4, $t7)], expr#9=['OTHER'], expr#10=[CASE($t5, $t6, $t8, $t1, $t9)], gender=[$t0], age=[$t10], avg(balance)=[$t2])
             EnumerableMergeJoin(condition=[=($1, $3)], joinType=[left])
               EnumerableSort(sort0=[$1], dir0=[ASC])
-                EnumerableCalc(expr#0..3=[{inputs}], expr#4=[SAFE_CAST($t1)], expr#5=[0], expr#6=[=($t3, $t5)], expr#7=[null:BIGINT], expr#8=[CASE($t6, $t7, $t2)], expr#9=[CAST($t8):DOUBLE], expr#10=[/($t9, $t3)], gender=[$t0], age=[$t4], avg(balance)=[$t10])
-                  EnumerableAggregate(group=[{0, 2}], agg#0=[$SUM0($1)], agg#1=[COUNT($1)])
-                    EnumerableCalc(expr#0..2=[{inputs}], expr#3=[10], expr#4=[null:NULL], expr#5=[SPAN($t2, $t3, $t4)], gender=[$t1], balance=[$t0], age0=[$t5])
-                      CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank_with_null_values]], PushDownContext=[[PROJECT->[balance, gender, age], FILTER->AND(IS NOT NULL($1), IS NOT NULL($0))], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","query":{"bool":{"must":[{"exists":{"field":"gender","boost":1.0}},{"exists":{"field":"balance","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"_source":{"includes":["balance","gender","age"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
+                EnumerableCalc(expr#0..2=[{inputs}], expr#3=[SAFE_CAST($t1)], gender=[$t0], age=[$t3], avg(balance)=[$t2])
+                  CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank_with_null_values]], PushDownContext=[[FILTER->AND(IS NOT NULL($1), IS NOT NULL($0)), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 2},avg(balance)=AVG($1))], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","query":{"bool":{"must":[{"exists":{"field":"gender","boost":1.0}},{"exists":{"field":"balance","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"gender":{"terms":{"script":{"source":"{\"langType\":\"calcite\",\"script\":\"rO0ABXQAaXsKICAiZHluYW1pY1BhcmFtIjogMCwKICAidHlwZSI6IHsKICAgICJ0eXBlIjogIlZBUkNIQVIiLAogICAgIm51bGxhYmxlIjogdHJ1ZSwKICAgICJwcmVjaXNpb24iOiAtMQogIH0KfQ==\"}","lang":"opensearch_compounded_script","params":{"utcTimestamp": 0,"SOURCES":[1],"DIGESTS":["gender"]}},"missing_bucket":true,"missing_order":"first","order":"asc"}}},{"age0":{"histogram":{"field":"age","missing_bucket":true,"missing_order":"first","order":"asc","interval":10.0}}}]},"aggregations":{"avg(balance)":{"avg":{"field":"balance"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
               EnumerableSort(sort0=[$0], dir0=[ASC])
                 EnumerableCalc(expr#0..2=[{inputs}], age=[$t0], _row_number_chart_=[$t2])
                   EnumerableWindow(window#0=[window(order by [1 DESC-nulls-last] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
                     EnumerableAggregate(group=[{0}], __grand_total__=[SUM($1)])
-                      EnumerableCalc(expr#0..3=[{inputs}], expr#4=[SAFE_CAST($t1)], expr#5=[0], expr#6=[=($t3, $t5)], expr#7=[null:BIGINT], expr#8=[CASE($t6, $t7, $t2)], expr#9=[CAST($t8):DOUBLE], expr#10=[/($t9, $t3)], expr#11=[IS NOT NULL($t4)], age=[$t4], avg(balance)=[$t10], $condition=[$t11])
-                        EnumerableAggregate(group=[{0, 2}], agg#0=[$SUM0($1)], agg#1=[COUNT($1)])
-                          EnumerableCalc(expr#0..2=[{inputs}], expr#3=[10], expr#4=[null:NULL], expr#5=[SPAN($t2, $t3, $t4)], gender=[$t1], balance=[$t0], age0=[$t5])
-                            CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank_with_null_values]], PushDownContext=[[PROJECT->[balance, gender, age], FILTER->AND(IS NOT NULL($1), IS NOT NULL($0))], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","query":{"bool":{"must":[{"exists":{"field":"gender","boost":1.0}},{"exists":{"field":"balance","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"_source":{"includes":["balance","gender","age"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
+                      EnumerableCalc(expr#0..1=[{inputs}], expr#2=[SAFE_CAST($t0)], expr#3=[IS NOT NULL($t2)], age=[$t2], avg(balance)=[$t1], $condition=[$t3])
+                        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank_with_null_values]], PushDownContext=[[FILTER->AND(IS NOT NULL($1), IS NOT NULL($0)), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 2},avg(balance)=AVG($1)), PROJECT->[age0, avg(balance)]], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","query":{"bool":{"must":[{"exists":{"field":"gender","boost":1.0}},{"exists":{"field":"balance","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"gender":{"terms":{"script":{"source":"{\"langType\":\"calcite\",\"script\":\"rO0ABXQAaXsKICAiZHluYW1pY1BhcmFtIjogMCwKICAidHlwZSI6IHsKICAgICJ0eXBlIjogIlZBUkNIQVIiLAogICAgIm51bGxhYmxlIjogdHJ1ZSwKICAgICJwcmVjaXNpb24iOiAtMQogIH0KfQ==\"}","lang":"opensearch_compounded_script","params":{"utcTimestamp": 0,"SOURCES":[1],"DIGESTS":["gender"]}},"missing_bucket":true,"missing_order":"first","order":"asc"}}},{"age0":{"histogram":{"field":"age","missing_bucket":true,"missing_order":"first","order":"asc","interval":10.0}}}]},"aggregations":{"avg(balance)":{"avg":{"field":"balance"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])

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

@@ -0,0 +1,10 @@
+calcite:
+  logical: |
+    LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
+      LogicalProject(account_number=[$0], firstname=[$1], address=[$2], birthdate=[$3], gender=[$4], city=[$5], lastname=[$6], balance=[$7], employer=[$8], state=[$9], age=[$10], email=[$11], male=[$12])
+        LogicalFilter(condition=[<=($19, 1)])
+          LogicalProject(account_number=[$0], firstname=[$1], address=[$2], birthdate=[$3], gender=[$4], city=[$5], lastname=[$6], balance=[$7], employer=[$8], state=[$9], age=[$10], email=[$11], male=[$12], _id=[$13], _index=[$14], _score=[$15], _maxscore=[$16], _sort=[$17], _routing=[$18], _row_number_dedup_=[ROW_NUMBER() OVER (PARTITION BY $11)])
+            LogicalFilter(condition=[IS NOT NULL($11)])
+              CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]])
+  physical: |
+    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=LogicalProject#,group={0},agg#0=LITERAL_AGG(1)), LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"email":{"terms":{"script":{"source":"{\"langType\":\"calcite\",\"script\":\"rO0ABXQAaXsKICAiZHluYW1pY1BhcmFtIjogMCwKICAidHlwZSI6IHsKICAgICJ0eXBlIjogIlZBUkNIQVIiLAogICAgIm51bGxhYmxlIjogdHJ1ZSwKICAgICJwcmVjaXNpb24iOiAtMQogIH0KfQ==\"}","lang":"opensearch_compounded_script","params":{"utcTimestamp": 0,"SOURCES":[1],"DIGESTS":["email"]}},"missing_bucket":false,"order":"asc"}}}]},"aggregations":{"$f1":{"top_hits":{"from":0,"size":1,"version":false,"seq_no_primary_term":false,"explain":false,"fields":[{"field":"email"},{"field":"account_number"},{"field":"firstname"},{"field":"address"},{"field":"birthdate"},{"field":"gender"},{"field":"city"},{"field":"lastname"},{"field":"balance"},{"field":"employer"},{"field":"state"},{"field":"age"},{"field":"male"}]}}}}}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])

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

@@ -19,21 +19,14 @@ calcite:
                           LogicalFilter(condition=[AND(IS NOT NULL($1), IS NOT NULL($7))])
                             CalciteLogicalIndexScan(table=[[OpenSearch, events]])
   physical: |
-    CalciteEnumerableTopK(sort0=[$0], sort1=[$1], dir0=[ASC], dir1=[ASC], fetch=[10000])
-      EnumerableCalc(expr#0..3=[{inputs}], expr#4=[0], expr#5=[=($t3, $t4)], expr#6=[null:DOUBLE], expr#7=[CASE($t5, $t6, $t2)], expr#8=[/($t7, $t3)], proj#0..1=[{exprs}], avg(cpu_usage)=[$t8])
+    EnumerableCalc(expr#0..3=[{inputs}], expr#4=[0], expr#5=[=($t3, $t4)], expr#6=[null:DOUBLE], expr#7=[CASE($t5, $t6, $t2)], expr#8=[/($t7, $t3)], proj#0..1=[{exprs}], avg(cpu_usage)=[$t8])
+      CalciteEnumerableTopK(sort0=[$0], sort1=[$1], dir0=[ASC], dir1=[ASC], fetch=[10000])
         EnumerableAggregate(group=[{0, 1}], agg#0=[$SUM0($2)], agg#1=[COUNT($2)])
           EnumerableCalc(expr#0..4=[{inputs}], expr#5=[IS NULL($t1)], expr#6=['NULL'], expr#7=[10], expr#8=[<=($t4, $t7)], expr#9=['OTHER'], expr#10=[CASE($t5, $t6, $t8, $t1, $t9)], @timestamp=[$t0], host=[$t10], avg(cpu_usage)=[$t2])
             EnumerableMergeJoin(condition=[=($1, $3)], joinType=[left])
-              EnumerableSort(sort0=[$1], dir0=[ASC])
-                EnumerableCalc(expr#0..3=[{inputs}], expr#4=[0], expr#5=[=($t3, $t4)], expr#6=[null:DOUBLE], expr#7=[CASE($t5, $t6, $t2)], expr#8=[/($t7, $t3)], @timestamp=[$t1], host=[$t0], avg(cpu_usage)=[$t8])
-                  EnumerableAggregate(group=[{0, 2}], agg#0=[$SUM0($1)], agg#1=[COUNT($1)])
-                    EnumerableCalc(expr#0..2=[{inputs}], expr#3=[1], expr#4=['m'], expr#5=[SPAN($t0, $t3, $t4)], host=[$t1], cpu_usage=[$t2], @timestamp0=[$t5])
-                      CalciteEnumerableIndexScan(table=[[OpenSearch, events]], PushDownContext=[[PROJECT->[@timestamp, host, cpu_usage], FILTER->AND(IS NOT NULL($0), IS NOT NULL($2))], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","query":{"bool":{"must":[{"exists":{"field":"@timestamp","boost":1.0}},{"exists":{"field":"cpu_usage","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"_source":{"includes":["@timestamp","host","cpu_usage"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
+              CalciteEnumerableIndexScan(table=[[OpenSearch, events]], PushDownContext=[[FILTER->AND(IS NOT NULL($0), IS NOT NULL($2)), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 2},avg(cpu_usage)=AVG($1)), PROJECT->[@timestamp0, host, avg(cpu_usage)], SORT->[1]], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","query":{"bool":{"must":[{"exists":{"field":"@timestamp","boost":1.0}},{"exists":{"field":"cpu_usage","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"host":{"terms":{"script":{"source":"{\"langType\":\"calcite\",\"script\":\"rO0ABXQAaXsKICAiZHluYW1pY1BhcmFtIjogMCwKICAidHlwZSI6IHsKICAgICJ0eXBlIjogIlZBUkNIQVIiLAogICAgIm51bGxhYmxlIjogdHJ1ZSwKICAgICJwcmVjaXNpb24iOiAtMQogIH0KfQ==\"}","lang":"opensearch_compounded_script","params":{"utcTimestamp": 0,"SOURCES":[1],"DIGESTS":["host"]}},"missing_bucket":true,"missing_order":"last","order":"asc"}}},{"@timestamp0":{"date_histogram":{"field":"@timestamp","missing_bucket":false,"order":"asc","fixed_interval":"1m"}}}]},"aggregations":{"avg(cpu_usage)":{"avg":{"field":"cpu_usage"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
               EnumerableSort(sort0=[$0], dir0=[ASC])
                 EnumerableCalc(expr#0..2=[{inputs}], host=[$t0], _row_number_chart_=[$t2])
                   EnumerableWindow(window#0=[window(order by [1 DESC-nulls-last] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
                     EnumerableAggregate(group=[{0}], __grand_total__=[SUM($1)])
-                      EnumerableCalc(expr#0..3=[{inputs}], expr#4=[0], expr#5=[=($t3, $t4)], expr#6=[null:DOUBLE], expr#7=[CASE($t5, $t6, $t2)], expr#8=[/($t7, $t3)], host=[$t0], avg(cpu_usage)=[$t8])
-                        EnumerableAggregate(group=[{0, 2}], agg#0=[$SUM0($1)], agg#1=[COUNT($1)])
-                          EnumerableCalc(expr#0..2=[{inputs}], expr#3=[1], expr#4=['m'], expr#5=[SPAN($t2, $t3, $t4)], proj#0..1=[{exprs}], @timestamp0=[$t5])
-                            CalciteEnumerableIndexScan(table=[[OpenSearch, events]], PushDownContext=[[PROJECT->[@timestamp, host, cpu_usage], FILTER->AND(IS NOT NULL($0), IS NOT NULL($2)), PROJECT->[host, cpu_usage, @timestamp], FILTER->IS NOT NULL($0)], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","query":{"bool":{"filter":[{"bool":{"must":[{"exists":{"field":"@timestamp","boost":1.0}},{"exists":{"field":"cpu_usage","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},{"exists":{"field":"host","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"_source":{"includes":["host","cpu_usage","@timestamp"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
+                      CalciteEnumerableIndexScan(table=[[OpenSearch, events]], PushDownContext=[[FILTER->AND(IS NOT NULL($0), IS NOT NULL($2)), FILTER->IS NOT NULL($0), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 2},avg(cpu_usage)=AVG($1)), PROJECT->[host, avg(cpu_usage)]], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","query":{"bool":{"filter":[{"bool":{"must":[{"exists":{"field":"@timestamp","boost":1.0}},{"exists":{"field":"cpu_usage","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},{"exists":{"field":"host","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"host":{"terms":{"script":{"source":"{\"langType\":\"calcite\",\"script\":\"rO0ABXQAaXsKICAiZHluYW1pY1BhcmFtIjogMCwKICAidHlwZSI6IHsKICAgICJ0eXBlIjogIlZBUkNIQVIiLAogICAgIm51bGxhYmxlIjogdHJ1ZSwKICAgICJwcmVjaXNpb24iOiAtMQogIH0KfQ==\"}","lang":"opensearch_compounded_script","params":{"utcTimestamp": 0,"SOURCES":[1],"DIGESTS":["host"]}},"missing_bucket":true,"missing_order":"first","order":"asc"}}},{"@timestamp0":{"date_histogram":{"field":"@timestamp","missing_bucket":false,"order":"asc","fixed_interval":"1m"}}}]},"aggregations":{"avg(cpu_usage)":{"avg":{"field":"cpu_usage"}}}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])