diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java
index 04dc2b0e74b..d4850951eab 100644
--- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java
+++ b/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)));
   }
diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteTimechartCommandIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteTimechartCommandIT.java
index 73396ab31b9..7bd32bb2af2 100644
--- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteTimechartCommandIT.java
+++ b/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)");
diff --git a/integ-test/src/test/java/org/opensearch/sql/ppl/DedupCommandIT.java b/integ-test/src/test/java/org/opensearch/sql/ppl/DedupCommandIT.java
index 45ecf02af8f..1507764b020 100644
--- a/integ-test/src/test/java/org/opensearch/sql/ppl/DedupCommandIT.java
+++ b/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 =
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/chart_null_str.yaml b/integ-test/src/test/resources/expectedOutput/calcite/chart_null_str.yaml
index 726eeedc429..9776f5dafc1 100644
--- a/integ-test/src/test/resources/expectedOutput/calcite/chart_null_str.yaml
+++ b/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)])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_dedup_text_type_no_push.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_dedup_text_type_no_push.yaml
deleted file mode 100644
index e7f26a14a96..00000000000
--- a/integ-test/src/test/resources/expectedOutput/calcite/explain_dedup_text_type_no_push.yaml
+++ /dev/null
@@ -1,13 +0,0 @@
-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: |
-    EnumerableLimit(fetch=[10000])
-      EnumerableCalc(expr#0..13=[{inputs}], expr#14=[1], expr#15=[<=($t13, $t14)], proj#0..12=[{exprs}], $condition=[$t15])
-        EnumerableWindow(window#0=[window(partition {11} rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
-          CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[PROJECT->[account_number, firstname, address, birthdate, gender, city, lastname, balance, employer, state, age, email, male], FILTER->IS NOT NULL($11)], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","query":{"exists":{"field":"email","boost":1.0}},"_source":{"includes":["account_number","firstname","address","birthdate","gender","city","lastname","balance","employer","state","age","email","male"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_dedup_text_type_push.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_dedup_text_type_push.yaml
new file mode 100644
index 00000000000..11aa4be7da2
--- /dev/null
+++ b/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)])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_timechart.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_timechart.yaml
index 0818c18eabb..bbb9c602016 100644
--- a/integ-test/src/test/resources/expectedOutput/calcite/explain_timechart.yaml
+++ b/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)])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_timechart_count.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_timechart_count.yaml
index f26cb9e5822..a8f49b59ee4 100644
--- a/integ-test/src/test/resources/expectedOutput/calcite/explain_timechart_count.yaml
+++ b/integ-test/src/test/resources/expectedOutput/calcite/explain_timechart_count.yaml
@@ -23,13 +23,8 @@ calcite:
       EnumerableAggregate(group=[{0, 1}], count()=[$SUM0($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], count()=[$t2])
           EnumerableMergeJoin(condition=[=($1, $3)], joinType=[left])
-            EnumerableSort(sort0=[$1], dir0=[ASC])
-              EnumerableCalc(expr#0..2=[{inputs}], @timestamp=[$t1], host=[$t0], count()=[$t2])
-                EnumerableAggregate(group=[{0, 1}], count()=[COUNT()])
-                  EnumerableCalc(expr#0..1=[{inputs}], expr#2=[1], expr#3=['m'], expr#4=[SPAN($t0, $t2, $t3)], host=[$t1], @timestamp0=[$t4])
-                    CalciteEnumerableIndexScan(table=[[OpenSearch, events]], PushDownContext=[[PROJECT->[@timestamp, host], FILTER->IS NOT NULL($0)], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","query":{"exists":{"field":"@timestamp","boost":1.0}},"_source":{"includes":["@timestamp","host"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
+            CalciteEnumerableIndexScan(table=[[OpenSearch, events]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 1},count()=COUNT()), PROJECT->[@timestamp0, host, count()], SORT->[1]], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","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"}}}]}}}}, 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__=[COUNT()])
-                    CalciteEnumerableIndexScan(table=[[OpenSearch, events]], PushDownContext=[[PROJECT->[@timestamp, host], FILTER->IS NOT NULL($0), PROJECT->[host, @timestamp], FILTER->IS NOT NULL($0)], OpenSearchRequestBuilder(sourceBuilder={"from":0,"timeout":"1m","query":{"bool":{"filter":[{"exists":{"field":"@timestamp","boost":1.0}},{"exists":{"field":"host","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"_source":{"includes":["host","@timestamp"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
+                  CalciteEnumerableIndexScan(table=[[OpenSearch, events]], PushDownContext=[[FILTER->IS NOT NULL($0), FILTER->IS NOT NULL($0), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},__grand_total__=COUNT())], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","query":{"bool":{"filter":[{"exists":{"field":"@timestamp","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"}}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
diff --git a/integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues/text_agg_pushdown.yml b/integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues/text_agg_pushdown.yml
new file mode 100644
index 00000000000..f13798a7729
--- /dev/null
+++ b/integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues/text_agg_pushdown.yml
@@ -0,0 +1,282 @@
+setup:
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled: true
+
+  # Primary index — appId is text-only, no keyword sub-field. Before the fix this
+  # scenario silently degraded to a full _source scan with client-side aggregation.
+  - do:
+      indices.create:
+        index: logs_text_only
+        body:
+          settings:
+            number_of_shards: 1
+            number_of_replicas: 0
+          mappings:
+            properties:
+              appId:
+                type: text
+              level:
+                type: keyword
+
+  - do:
+      bulk:
+        index: logs_text_only
+        refresh: true
+        body:
+          - '{"index": {"_id": "1"}}'
+          - '{"appId": "app-1", "level": "info"}'
+          - '{"index": {"_id": "2"}}'
+          - '{"appId": "app-1", "level": "warn"}'
+          - '{"index": {"_id": "3"}}'
+          - '{"appId": "app-2", "level": "info"}'
+          - '{"index": {"_id": "4"}}'
+          - '{"appId": "app-3", "level": "info"}'
+          - '{"index": {"_id": "5"}}'
+          - '{"appId": "app-1", "level": "info"}'
+
+  # Companion index — appId is text with a keyword sub-field. Sits under the
+  # same logs_text* pattern so multi-index queries also exercise the
+  # conflicting-mapping case where the merged type loses its keyword sub-field.
+  - do:
+      indices.create:
+        index: logs_text_with_keyword
+        body:
+          settings:
+            number_of_shards: 1
+            number_of_replicas: 0
+          mappings:
+            properties:
+              appId:
+                type: text
+                fields:
+                  keyword:
+                    type: keyword
+                    ignore_above: 256
+
+  - do:
+      bulk:
+        index: logs_text_with_keyword
+        refresh: true
+        body:
+          - '{"index": {}}'
+          - '{"appId": "app-1"}'
+          - '{"index": {}}'
+          - '{"appId": "app-1"}'
+          - '{"index": {}}'
+          - '{"appId": "app-2"}'
+
+---
+teardown:
+  - do:
+      indices.delete:
+        index: logs_text_only,logs_text_with_keyword
+        ignore_unavailable: true
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled: false
+
+---
+"top command on a text field returns correct counts":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=logs_text_only | top 10 appId
+  - match: { total: 3 }
+  - match:
+      schema:
+        - { name: appId, type: string }
+        - { name: count, type: bigint }
+  - match:
+      datarows:
+        - [ "app-1", 3 ]
+        - [ "app-2", 1 ]
+        - [ "app-3", 1 ]
+
+---
+"stats by text field returns correct counts":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=logs_text_only | stats count() by appId
+  - match: { total: 3 }
+  - match:
+      schema:
+        - { name: "count()", type: bigint }
+        - { name: appId, type: string }
+  - match:
+      datarows:
+        - [ 3, "app-1" ]
+        - [ 1, "app-2" ]
+        - [ 1, "app-3" ]
+
+---
+"top on text field pushes down as composite terms with _source script":
+  # AGGREGATION pushdown means the physical plan carries a composite_buckets
+  # aggregation whose terms source uses a script (not a field), because appId
+  # is a text field with no .keyword sub-field. Presence of both AGGREGATION
+  # in PushDownContext and the composite/script bucket confirms the DSL is
+  # pushed to OpenSearch instead of falling back to an unbounded scan.
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl.explain:
+        body:
+          query: source=logs_text_only | top 10 appId
+  - match: { "calcite.physical": "/AGGREGATION/" }
+  - match: { "calcite.physical": "/composite_buckets/" }
+  - match: { "calcite.physical": "/\"appId\":\\{\"terms\":\\{\"script\"/" }
+
+---
+"stats by text field pushes down as composite terms with _source script":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl.explain:
+        body:
+          query: source=logs_text_only | stats count() by appId
+  - match: { "calcite.physical": "/AGGREGATION/" }
+  - match: { "calcite.physical": "/composite_buckets/" }
+  - match: { "calcite.physical": "/\"appId\":\\{\"terms\":\\{\"script\"/" }
+
+---
+"count(text_field) on text-only pushes down as scripted value_count":
+  # count(FIELD) reaches the metric path with a bare RexInputRef (Calcite does not
+  # insert a numeric cast around COUNT). For a text field without .keyword, the
+  # metric aggregation must still push down by using a script value source that
+  # reads the field from _source, not fall back to an unbounded scan.
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl.explain:
+        body:
+          query: source=logs_text_only | stats count(appId)
+  - match: { "calcite.physical": "/AGGREGATION/" }
+  - match: { "calcite.physical": "/\"count\\(appId\\)\":\\{\"value_count\":\\{\"script\"/" }
+
+---
+"count(text_field) on text-only returns correct count":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=logs_text_only | stats count(appId)
+  - match: { total: 1 }
+  - match:
+      schema:
+        - { name: "count(appId)", type: bigint }
+  - match:
+      datarows:
+        - [ 5 ]
+
+---
+"text field with .keyword sub-field pushes down using the sub-field":
+  # A text field that carries `.keyword` must be rewritten to `<field>.keyword`
+  # in the DSL, not routed through the script path. Baseline that guarantees the
+  # fix did NOT alter behavior for keyword-backed fields.
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl.explain:
+        body:
+          query: source=logs_text_with_keyword | stats count() by appId
+  - match: { "calcite.physical": "/AGGREGATION/" }
+  - match: { "calcite.physical": "/composite_buckets/" }
+  - match: { "calcite.physical": "/\"appId\":\\{\"terms\":\\{\"field\":\"appId\\.keyword\"/" }
+
+---
+"multi-index stats across text-only + text+keyword pushes down via _source script":
+  # `source=logs_text*` spans a text-only and a text+keyword index. The
+  # merged `appId` field loses its .keyword sub-field (because one member has
+  # none). Before the fix this fell back to a full _source scan and client-side
+  # aggregation; after the fix the composite terms bucket uses the _source-
+  # reading script for the merged field type.
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl.explain:
+        body:
+          query: source=logs_text* | stats count() by appId
+  - match: { "calcite.physical": "/AGGREGATION/" }
+  - match: { "calcite.physical": "/composite_buckets/" }
+  - match: { "calcite.physical": "/\"appId\":\\{\"terms\":\\{\"script\"/" }
+
+---
+"multi-index stats returns correct merged counts":
+  # logs_text_only:         app-1=3, app-2=1, app-3=1
+  # logs_text_with_keyword: app-1=2, app-2=1
+  # merged:                 app-1=5, app-2=2, app-3=1
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=logs_text* | stats count() by appId
+  - match: { total: 3 }
+  - match:
+      schema:
+        - { name: "count()", type: bigint }
+        - { name: appId, type: string }
+  - match:
+      datarows:
+        - [ 5, "app-1" ]
+        - [ 2, "app-2" ]
+        - [ 1, "app-3" ]
+
+---
+"multi-index top returns correct merged counts":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=logs_text* | top 10 appId
+  - match: { total: 3 }
+  - match:
+      schema:
+        - { name: appId, type: string }
+        - { name: count, type: bigint }
+  - match:
+      datarows:
+        - [ "app-1", 5 ]
+        - [ "app-2", 2 ]
+        - [ "app-3", 1 ]
diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java
index 775b0278683..54561029e5e 100644
--- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java
+++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/AggregateAnalyzer.java
@@ -158,7 +158,13 @@ <T extends CompositeValuesSourceBuilder<T>> T build(RexNode node, T sourceBuilde
     <T> T build(RexNode node, Function<String, T> fieldBuilder, Function<Script, T> scriptBuilder) {
       if (node == null) return fieldBuilder.apply(METADATA_FIELD);
       else if (node instanceof RexInputRef ref) {
-        return fieldBuilder.apply(inferNamedField(node).getReferenceForTermQuery());
+        String fieldRef = inferNamedField(node).getReferenceForTermQuery();
+        // Text field with no .keyword sub-field is not aggregatable directly. Fall back to a
+        // Calcite script that reads the value from _source.
+        if (fieldRef == null) {
+          return scriptBuilder.apply(inferScript(node).getScript());
+        }
+        return fieldBuilder.apply(fieldRef);
       } else if (node instanceof RexCall || node instanceof RexLiteral) {
         return scriptBuilder.apply(inferScript(node).getScript());
       }
@@ -175,7 +181,7 @@ NamedFieldExpression inferNamedField(RexNode node) {
     }
 
     ScriptQueryExpression inferScript(RexNode node) {
-      if (node instanceof RexCall || node instanceof RexLiteral) {
+      if (node instanceof RexCall || node instanceof RexLiteral || node instanceof RexInputRef) {
         return new ScriptQueryExpression(
             node, rowType, fieldTypes, cluster, Collections.emptyMap());
       }
diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/request/AggregateAnalyzerTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/request/AggregateAnalyzerTest.java
index 4779332abac..cfa4de6f695 100644
--- a/opensearch/src/test/java/org/opensearch/sql/opensearch/request/AggregateAnalyzerTest.java
+++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/request/AggregateAnalyzerTest.java
@@ -7,7 +7,6 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -301,33 +300,81 @@ void analyze_groupBy() throws ExpressionNotAnalyzableException {
   }
 
   @Test
-  void analyze_aggCall_TextWithoutKeyword() {
+  void analyze_aggCall_TextWithoutKeyword_countPushesDownAsScript()
+      throws ExpressionNotAnalyzableException {
+    // count(FIELD) on a text field with no .keyword sub-field must not fall back to an
+    // unbounded client-side scan. The metric aggregation is built with a script value source
+    // that reads the value from _source, matching the pattern used by TermQuery/LikeQuery.
+    Hook.CURRENT_TIME.addThread((Consumer<Holder<Long>>) h -> h.set(0L));
+
+    SchemaPlus root = Frameworks.createRootSchema(true);
+    root.add(
+        "test",
+        new AbstractTable() {
+          @Override
+          public RelDataType getRowType(RelDataTypeFactory tf) {
+            return rowType;
+          }
+        });
+    RelBuilder rb =
+        RelBuilder.create(Frameworks.newConfigBuilder().defaultSchema(root).build()).scan("test");
+
     AggregateCall aggCall =
         AggregateCall.create(
-            SqlStdOperatorTable.SUM,
+            SqlStdOperatorTable.COUNT,
             false,
             false,
             false,
             ImmutableList.of(),
-            ImmutableList.of(0),
+            // arg #2 in the row type is `c` — text without .keyword sub-field
+            ImmutableList.of(2),
             -1,
             null,
             RelCollations.EMPTY,
-            typeFactory.createSqlType(SqlTypeName.INTEGER),
-            "sum");
-    Aggregate aggregate = createMockAggregate(List.of(aggCall), ImmutableBitSet.of());
-    Project project = createMockProject(List.of(2));
+            typeFactory.createSqlType(SqlTypeName.BIGINT),
+            "cnt");
+
+    RelNode rel = rb.aggregate(rb.groupKey(), List.of(aggCall)).build();
+    Aggregate aggregate = (Aggregate) rel;
+
     AggregateAnalyzer.AggregateBuilderHelper helper =
-        new AggregateAnalyzer.AggregateBuilderHelper(rowType, fieldTypes, null, true, BUCKET_SIZE);
-    ExpressionNotAnalyzableException exception =
-        assertThrows(
-            ExpressionNotAnalyzableException.class,
-            () -> AggregateAnalyzer.analyze(aggregate, project, List.of("sum"), helper));
-    assertEquals("[field] must not be null: [sum]", exception.getCause().getMessage());
+        new AggregateAnalyzer.AggregateBuilderHelper(
+            rowType, fieldTypes, aggregate.getCluster(), true, BUCKET_SIZE);
+    Pair<List<AggregationBuilder>, OpenSearchAggregationResponseParser> result =
+        AggregateAnalyzer.analyze(aggregate, null, List.of("cnt"), helper);
+
+    String dsl = result.getLeft().toString();
+    // The value_count metric must use a script value source (not a raw field) on `c`.
+    assertTrue(
+        dsl.contains("\"cnt\":{\"value_count\":{\"script\":{"),
+        "expected value_count metric on 'c' to use a script, but got: " + dsl);
+    assertTrue(
+        dsl.contains("\"lang\":\"opensearch_compounded_script\""),
+        "expected compounded script lang, but got: " + dsl);
+    assertTrue(
+        dsl.contains("\"DIGESTS\":[\"c\"]"),
+        "expected script to reference field 'c' via DIGESTS, but got: " + dsl);
   }
 
   @Test
-  void analyze_groupBy_TextWithoutKeyword() {
+  void analyze_groupBy_TextWithoutKeyword() throws ExpressionNotAnalyzableException {
+    // Grouping by a text field with no .keyword sub-field must not fall back to an unbounded
+    // client-side scan. Instead, the composite terms bucket is built with a script value source
+    // that reads the field from _source, matching the pattern used by TermQuery/LikeQuery.
+    Hook.CURRENT_TIME.addThread((Consumer<Holder<Long>>) h -> h.set(0L));
+
+    SchemaPlus root = Frameworks.createRootSchema(true);
+    root.add(
+        "test",
+        new AbstractTable() {
+          @Override
+          public RelDataType getRowType(RelDataTypeFactory tf) {
+            return rowType;
+          }
+        });
+    RelBuilder rb =
+        RelBuilder.create(Frameworks.newConfigBuilder().defaultSchema(root).build()).scan("test");
+
     AggregateCall aggCall =
         AggregateCall.create(
             SqlStdOperatorTable.COUNT,
@@ -341,16 +388,32 @@ void analyze_groupBy_TextWithoutKeyword() {
             RelCollations.EMPTY,
             typeFactory.createSqlType(SqlTypeName.INTEGER),
             "cnt");
-    List<String> outputFields = List.of("c", "cnt");
-    Aggregate aggregate = createMockAggregate(List.of(aggCall), ImmutableBitSet.of(0));
-    Project project = createMockProject(List.of(2));
+
+    RelNode rel = rb.aggregate(rb.groupKey(ImmutableBitSet.of(2)), List.of(aggCall)).build();
+    Aggregate aggregate = (Aggregate) rel;
+    Project project = null;
+
     AggregateAnalyzer.AggregateBuilderHelper helper =
-        new AggregateAnalyzer.AggregateBuilderHelper(rowType, fieldTypes, null, true, BUCKET_SIZE);
-    ExpressionNotAnalyzableException exception =
-        assertThrows(
-            ExpressionNotAnalyzableException.class,
-            () -> AggregateAnalyzer.analyze(aggregate, project, outputFields, helper));
-    assertEquals("[field] must not be null", exception.getCause().getMessage());
+        new AggregateAnalyzer.AggregateBuilderHelper(
+            rowType, fieldTypes, aggregate.getCluster(), true, BUCKET_SIZE);
+    Pair<List<AggregationBuilder>, OpenSearchAggregationResponseParser> result =
+        AggregateAnalyzer.analyze(aggregate, project, List.of("c", "cnt"), helper);
+
+    String dsl = result.getLeft().toString();
+    // Composite bucket for `c` must use a script value source (not a raw field).
+    assertTrue(
+        dsl.contains(
+            "\"composite_buckets\":{\"composite\":{\"size\":1000,\"sources\":["
+                + "{\"c\":{\"terms\":{\"script\":{"),
+        "expected composite terms bucket on 'c' to use a script, but got: " + dsl);
+    // The script must be tagged as a Calcite compounded script and reference the `c` field.
+    assertTrue(
+        dsl.contains("\"lang\":\"opensearch_compounded_script\""),
+        "expected compounded script lang, but got: " + dsl);
+    assertTrue(
+        dsl.contains("\"DIGESTS\":[\"c\"]"),
+        "expected script to reference field 'c' via DIGESTS, but got: " + dsl);
+    assertInstanceOf(BucketAggregationParser.class, result.getRight());
   }
 
   @Test