Use table scan rowType in filter pushdown could fix rename issue (#4670)

Author: LantaoJin

integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues/4563_4664.yml

@@ -0,0 +1,63 @@
+setup:
+  - do:
+      indices.create:
+        index: test
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled : true
+
+  - do:
+      bulk:
+        index: test
+        refresh: true
+        body:
+          - '{"index": {}}'
+          - '{"status":"200","service":"api","value":100,"time":"2025-01-01T00:00:00Z"}'
+          - '{"index": {}}'
+          - '{"status":"500","service":"web","value":200,"time":"2025-01-02T00:00:00Z"}'
+          - '{"index": {}}'
+          - '{"status":"200","service":"db","value":150,"time":"2025-01-03T00:00:00Z"}'
+          - '{"index": {}}'
+          - '{"status":"404","service":"api","value":50,"time":"2025-01-03T00:01:00Z"}'
+
+---
+teardown:
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled : false
+
+---
+"4563: Test rename then dedup":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=test | rename status as http_status | dedup http_status | fields http_status
+
+  - match: { total: 3 }
+  - match: { schema: [{"name": "http_status", "type": "string"}] }
+  - match: { datarows: [["200"], ["500"], ["404"]] }
+
+---
+"4664: Test rename then filter":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=test | rename status as http_status | where http_status = '404' | fields http_status
+
+  - match: { total: 1 }
+  - match: { schema: [{"name": "http_status", "type": "string"}] }
+  - match: { datarows: [["404"]] }

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java

@@ -102,6 +102,11 @@ protected AbstractCalciteIndexScan buildScan(
         cluster, traitSet, hints, table, osIndex, schema, pushDownContext);
   }
 
+  public CalciteLogicalIndexScan copy() {
+    return new CalciteLogicalIndexScan(
+        getCluster(), traitSet, hints, table, osIndex, schema, pushDownContext.clone());
+  }
+
   public CalciteLogicalIndexScan copyWithNewSchema(RelDataType schema) {
     // Do shallow copy for requestBuilder, thus requestBuilder among different plans produced in the
     // optimization process won't affect each other.
@@ -131,8 +136,7 @@ public void register(RelOptPlanner planner) {
 
   public AbstractRelNode pushDownFilter(Filter filter) {
     try {
-      RelDataType rowType = filter.getRowType();
-      CalciteLogicalIndexScan newScan = this.copyWithNewSchema(filter.getRowType());
+      RelDataType rowType = this.getRowType();
       List<String> schema = this.getRowType().getFieldNames();
       Map<String, ExprType> fieldTypes =
           this.osIndex.getAllFieldTypes().entrySet().stream()
@@ -142,6 +146,7 @@ public AbstractRelNode pushDownFilter(Filter filter) {
           PredicateAnalyzer.analyzeExpression(
               filter.getCondition(), schema, fieldTypes, rowType, getCluster());
       // TODO: handle the case where condition contains a score function
+      CalciteLogicalIndexScan newScan = this.copy();
       newScan.pushDownContext.add(
           queryExpression.getScriptCount() > 0 ? PushDownType.SCRIPT : PushDownType.FILTER,
           new FilterDigest(