[Calcite Engine] Support alias type field (#3538)

* [Calcite Engine] Support alias type field

Signed-off-by: Heng Qian <qianheng@amazon.com>

* Add standalone IT to ensure w/wo push down test cases

Signed-off-by: Heng Qian <qianheng@amazon.com>

---------

Signed-off-by: Heng Qian <qianheng@amazon.com>

Author: qianheng-aws

integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLBasicIT.java

@@ -5,6 +5,7 @@
 
 package org.opensearch.sql.calcite.standalone;
 
+import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ALIAS;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK;
 import static org.opensearch.sql.util.MatcherUtils.rows;
 import static org.opensearch.sql.util.MatcherUtils.schema;
@@ -35,6 +36,7 @@ public void init() throws IOException {
     client().performRequest(request3);
 
     loadIndex(Index.BANK);
+    loadIndex(Index.DATA_TYPE_ALIAS);
   }
 
   @Test
@@ -517,4 +519,15 @@ public void testXor() {
                 TEST_INDEX_BANK));
     verifyDataRows(result, rows("Elinor", 36));
   }
+
+  @Test
+  public void testAliasDataType() {
+    JSONObject result =
+        executeQuery(
+            String.format(
+                "source=%s | where alias_col > 1 | fields original_col, alias_col ",
+                TEST_INDEX_ALIAS));
+    verifySchema(result, schema("original_col", "integer"), schema("alias_col", "integer"));
+    verifyDataRows(result, rows(2, 2), rows(3, 3));
+  }
 }

integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java

@@ -837,7 +837,7 @@ public enum Index {
         TestsConstants.TEST_INDEX_ALIAS,
         "alias",
         getAliasIndexMapping(),
-        "src/test/resources/work_information.json"),
+        "src/test/resources/alias.json"),
     DUPLICATION_NULLABLE(
         TestsConstants.TEST_INDEX_DUPLICATION_NULLABLE,
         "duplication_nullable",

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

@@ -11,7 +11,9 @@
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ALIAS;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_DATATYPE_NONNUMERIC;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_DATATYPE_NUMERIC;
+import static org.opensearch.sql.util.MatcherUtils.rows;
 import static org.opensearch.sql.util.MatcherUtils.schema;
+import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
 import static org.opensearch.sql.util.MatcherUtils.verifySchema;
 
 import java.io.IOException;
@@ -85,8 +87,9 @@ public void test_alias_data_type() throws IOException {
     JSONObject result =
         executeQuery(
             String.format(
-                "source=%s | where alias_col > 1 " + "| fields original_col, alias_col ",
+                "source=%s | where alias_col > 1 | fields original_col, alias_col ",
                 TEST_INDEX_ALIAS));
     verifySchema(result, schema("original_col", "int"), schema("alias_col", "int"));
+    verifyDataRows(result, rows(2, 2), rows(3, 3));
   }
 }

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java

@@ -10,7 +10,9 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import org.apache.calcite.linq4j.AbstractEnumerable;
@@ -81,6 +83,9 @@ public class OpenSearchIndex extends OpenSearchTable {
   /** The cached ExprType of fields. */
   private Map<String, ExprType> cachedFieldTypes = null;
 
+  /** The cached mapping of alias type field to its original path. */
+  private Map<String, String> aliasMapping = null;
+
   /** The cached max result window setting of index. */
   private Integer cachedMaxResultWindow = null;
 
@@ -148,6 +153,22 @@ public Map<String, ExprType> getReservedFieldTypes() {
     return METADATAFIELD_TYPE_MAP;
   }
 
+  public Map<String, String> getAliasMapping() {
+    if (cachedFieldOpenSearchTypes == null) {
+      cachedFieldOpenSearchTypes =
+          new OpenSearchDescribeIndexRequest(client, indexName).getFieldTypes();
+    }
+    if (aliasMapping == null) {
+      aliasMapping =
+          cachedFieldOpenSearchTypes.entrySet().stream()
+              .filter(entry -> entry.getValue().getOriginalPath().isPresent())
+              .collect(
+                  Collectors.toUnmodifiableMap(
+                      Entry::getKey, entry -> entry.getValue().getOriginalPath().get()));
+    }
+    return aliasMapping;
+  }
+
   /**
    * Get parsed mapping info.
    *

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

@@ -94,10 +94,16 @@ public Result implement(EnumerableRelImplementor implementor, Prefer pref) {
       public Enumerator<Object> enumerator() {
         return new OpenSearchIndexEnumerator(
             osIndex.getClient(),
-            List.copyOf(getRowType().getFieldNames()),
+            getFieldPath(),
             requestBuilder.getMaxResponseSize(),
             osIndex.buildRequest(requestBuilder));
       }
     };
   }
+
+  private List<String> getFieldPath() {
+    return getRowType().getFieldNames().stream()
+        .map(f -> osIndex.getAliasMapping().getOrDefault(f, f))
+        .toList();
+  }
 }