Coerce any to specific types (#4753)

ykmr1224 · web-flow · commit ac06da4ae519 · 2025-11-06T15:30:01.000-08:00
Signed-off-by: Tomoyuki Morita &lt;moritato@amazon.com&gt;
diff --git a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java
@@ -247,16 +247,17 @@ public RelNode visitRegex(Regex node, CalcitePlanContext context) {
     RexNode fieldRex = rexVisitor.analyze(node.getField(), context);
     RexNode patternRex = rexVisitor.analyze(node.getPattern(), context);
 
-    if (!SqlTypeFamily.CHARACTER.contains(fieldRex.getType())) {
+    if (!SqlTypeFamily.CHARACTER.contains(fieldRex.getType())
+        && !SqlTypeName.ANY.equals(fieldRex.getType().getSqlTypeName())) {
       throw new IllegalArgumentException(
           String.format(
               "Regex command requires field of string type, but got %s for field '%s'",
               fieldRex.getType().getSqlTypeName(), node.getField().toString()));
     }
 
     RexNode regexCondition =
-        context.rexBuilder.makeCall(
-            org.apache.calcite.sql.fun.SqlLibraryOperators.REGEXP_CONTAINS, fieldRex, patternRex);
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder, BuiltinFunctionName.REGEX_MATCH, fieldRex, patternRex);
 
     if (node.isNegated()) {
       regexCondition = context.rexBuilder.makeCall(SqlStdOperatorTable.NOT, regexCondition);
diff --git a/core/src/main/java/org/opensearch/sql/expression/function/CoercionUtils.java b/core/src/main/java/org/opensearch/sql/expression/function/CoercionUtils.java
@@ -17,8 +17,10 @@
 import java.util.function.BinaryOperator;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
+import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.commons.lang3.tuple.Pair;
 import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory;
 import org.opensearch.sql.data.type.ExprCoreType;
@@ -106,6 +108,14 @@ public final class CoercionUtils {
 
   private static @Nullable RexNode cast(RexBuilder builder, ExprType targetType, RexNode arg) {
     ExprType argType = OpenSearchTypeFactory.convertRelDataTypeToExprType(arg.getType());
+
+    // Special handling for ANY type (dynamic fields)
+    if (isAnyType(arg.getType())) {
+      // ANY type can be cast to any target type
+      return builder.makeCast(
+          OpenSearchTypeFactory.convertExprTypeToRelDataType(targetType), arg, true, true);
+    }
+
     if (!argType.shouldCast(targetType)) {
       return arg;
     }
@@ -176,6 +186,10 @@ public static boolean hasString(List<RexNode> rexNodeList) {
 
   private static final List<CoercionRule> COMMON_COERCION_RULES =
       List.of(
+          // ANY type coercion: if one side is ANY, use the other type
+          CoercionRule.of(
+              (left, right) -> isAnyType(left) || isAnyType(right),
+              (left, right) -> isAnyType(left) ? right : left),
           CoercionRule.of(
               (left, right) -> areDateAndTime(left, right),
               (left, right) -> ExprCoreType.TIMESTAMP),
@@ -212,6 +226,7 @@ static CoercionRule of(
 
   private static final int IMPOSSIBLE_WIDENING = Integer.MAX_VALUE;
   private static final int TYPE_EQUAL = 0;
+  private static final int ANY_TYPE_DISTANCE = 1;
 
   private static int distance(ExprType type1, ExprType type2) {
     return distance(type1, type2, TYPE_EQUAL);
@@ -222,6 +237,9 @@ private static int distance(ExprType type1, ExprType type2, int distance) {
       return distance;
     } else if (type1 == UNKNOWN) {
       return IMPOSSIBLE_WIDENING;
+    } else if (isAnyType(type1)) {
+      // ANY type (from dynamic fields) can coerce to any other type with distance 1
+      return distance + ANY_TYPE_DISTANCE;
     } else if (type1 == ExprCoreType.STRING && type2 == ExprCoreType.DOUBLE) {
       return 1;
     } else {
@@ -232,6 +250,20 @@ private static int distance(ExprType type1, ExprType type2, int distance) {
     }
   }
 
+  private static boolean isAnyType(RelDataType type) {
+    return type.getSqlTypeName() == SqlTypeName.ANY;
+  }
+
+  private static boolean isAnyType(ExprType type) {
+    // UNDEFINED is the ExprType representation of SqlTypeName.ANY
+    // but we need to distinguish it from actual UNDEFINED (NULL type)
+    // In the context of dynamic fields, UNDEFINED with no parents represents ANY type
+    if (type == ExprCoreType.UNDEFINED) {
+      return type.getParent().isEmpty();
+    }
+    return false;
+  }
+
   /**
    * The max type among two types. The max is defined as follow if type1 could widen to type2, then
    * max is type2, vice versa if type1 couldn't widen to type2 and type2 could't widen to type1,
diff --git a/core/src/test/java/org/opensearch/sql/expression/function/CoercionUtilsTest.java b/core/src/test/java/org/opensearch/sql/expression/function/CoercionUtilsTest.java
@@ -16,6 +16,7 @@
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -32,6 +33,11 @@ private static RexNode nullLiteral(ExprCoreType type) {
     return REX_BUILDER.makeNullLiteral(OpenSearchTypeFactory.convertExprTypeToRelDataType(type));
   }
 
+  private static RexNode anyTypedLiteral() {
+    return REX_BUILDER.makeNullLiteral(
+        OpenSearchTypeFactory.TYPE_FACTORY.createSqlType(SqlTypeName.ANY));
+  }
+
   private static Stream<Arguments> commonWidestTypeArguments() {
     return Stream.of(
         Arguments.of(STRING, INTEGER, DOUBLE),
@@ -82,6 +88,26 @@ void castArgumentsReturnsNullWhenNoCompatibleSignatureExists() {
     assertNull(CoercionUtils.castArguments(REX_BUILDER, typeChecker, arguments));
   }
 
+  @Test
+  void coerceAnyToString() {
+    testAnyToSpecificTypeCoercion(STRING);
+    testAnyToSpecificTypeCoercion(INTEGER);
+    testAnyToSpecificTypeCoercion(DOUBLE);
+    testAnyToSpecificTypeCoercion(BOOLEAN);
+    testAnyToSpecificTypeCoercion(ExprCoreType.TIMESTAMP);
+  }
+
+  void testAnyToSpecificTypeCoercion(ExprCoreType toType) {
+    PPLTypeChecker typeChecker = new StubTypeChecker(List.of(List.of(toType)));
+    List<RexNode> arguments = List.of(anyTypedLiteral());
+
+    List<RexNode> result = CoercionUtils.castArguments(REX_BUILDER, typeChecker, arguments);
+
+    assertEquals(1, result.size());
+    assertEquals(
+        toType, OpenSearchTypeFactory.convertRelDataTypeToExprType(result.getFirst().getType()));
+  }
+
   private static class StubTypeChecker implements PPLTypeChecker {
     private final List<List<ExprType>> signatures;
 
diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalciteDynamicFieldsCommandIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalciteDynamicFieldsCommandIT.java
@@ -124,16 +124,6 @@ public void testRegex() throws IOException {
     String query =
         source(
             TEST_INDEX_DYNAMIC, "regex department=\"Eng.*\" | fields account_number, department");
-    assertThrows(RuntimeException.class, () -> executeQuery(query));
-  }
-
-  @Test
-  public void testCastAndRegex() throws IOException {
-    String query =
-        source(
-            TEST_INDEX_DYNAMIC,
-            "eval department=CAST(department AS string) | regex department=\"Eng.*\" | fields"
-                + " account_number, department");
     JSONObject result = executeQuery(query);
 
     verifySchema(result, schema("account_number", "bigint"), schema("department", "string"));
@@ -145,40 +135,39 @@ public void testRex() throws IOException {
     String query =
         source(
             TEST_INDEX_DYNAMIC,
-            "rex field=firstname \"(?<initial>[A-Z])\" | fields firstname, initial | head 1");
+            "rex field=department '(?<initial>[A-Z])'" + " | fields department, initial | head 1");
     JSONObject result = executeQuery(query);
 
-    verifySchema(result, schema("firstname", "string"), schema("initial", "string"));
-    verifyDataRows(result, rows("John", "J"));
+    verifySchema(result, schema("department", "string"), schema("initial", "string"));
+    verifyDataRows(result, rows("Engineering", "E"));
   }
 
   @Test
-  public void testCastAndRex() throws IOException {
+  public void testRexWithNumeric() throws IOException {
     String query =
         source(
             TEST_INDEX_DYNAMIC,
-            "eval department=CAST(department AS string) | rex field=department '(?<initial>[A-Z])'"
-                + " | fields department, initial | head 1");
+            "rex field=salary '(?<initial>[0-9])'" + " | fields salary, initial | head 1");
     JSONObject result = executeQuery(query);
 
-    verifySchema(result, schema("department", "string"), schema("initial", "string"));
-    verifyDataRows(result, rows("Engineering", "E"));
+    verifySchema(result, schema("salary", "int"), schema("initial", "string"));
+    verifyDataRows(result, rows(75000, "7"));
   }
 
   @Test
-  public void testCastAndParse() throws IOException {
+  public void testParse() throws IOException {
     String query =
         source(
             TEST_INDEX_DYNAMIC,
-            "eval department=CAST(department AS string) | fields department | parse department"
-                + " '(?<initial>[A-Z]).*' | fields department, initial | head 1");
+            "parse department '(?<initial>[A-Z]).*' | fields department, initial | head 1");
+
     assertExplainYaml(
         query,
         "calcite:\n"
             + "  logical: |\n"
             + "    LogicalSystemLimit(fetch=[200], type=[QUERY_SIZE_LIMIT])\n"
             + "      LogicalSort(fetch=[1])\n"
-            + "        LogicalProject(department=[SAFE_CAST(ITEM($9, 'department'))],"
+            + "        LogicalProject(department=[ITEM($9, 'department')],"
             + " initial=[ITEM(PARSE(SAFE_CAST(ITEM($9, 'department')),"
             + " '(?<initial>[A-Z]).*':VARCHAR, 'regex':VARCHAR), 'initial':VARCHAR)])\n"
             + "          CalciteLogicalIndexScan(table=[[OpenSearch, test_dynamic_fields]])\n"
@@ -188,7 +177,7 @@ public void testCastAndParse() throws IOException {
             + " expr#11=[ITEM($t9, $t10)], expr#12=[SAFE_CAST($t11)],"
             + " expr#13=['(?<initial>[A-Z]).*':VARCHAR], expr#14=['regex':VARCHAR],"
             + " expr#15=[PARSE($t12, $t13, $t14)], expr#16=['initial':VARCHAR], expr#17=[ITEM($t15,"
-            + " $t16)], department=[$t12], initial=[$t17])\n"
+            + " $t16)], department=[$t11], initial=[$t17])\n"
             + "        EnumerableLimit(fetch=[1])\n"
             + "          CalciteEnumerableIndexScan(table=[[OpenSearch, test_dynamic_fields]])\n");
 
@@ -235,28 +224,41 @@ public void testRevers() throws IOException {
 
   @Test
   public void testBin() throws IOException {
-    String query = source(TEST_INDEX_DYNAMIC, "bin salary span=10000 | head 1");
+    String query =
+        source(TEST_INDEX_DYNAMIC, "bin salary span=10000 | fields account_number, salary");
     JSONObject result = executeQuery(query);
 
-    verifySchema(
-        result,
-        schema("account_number", "bigint"),
-        schema("firstname", "string"),
-        schema("lastname", "string"),
-        schema("salary", "string"),
-        schema("city", "string"),
-        schema("department", "string"),
-        schema("json", "string"));
+    assertExplainYaml(
+        query,
+        "calcite:\n"
+            + "  logical: |\n"
+            + "    LogicalSystemLimit(fetch=[200], type=[QUERY_SIZE_LIMIT])\n"
+            + "      LogicalProject(account_number=[$0], salary=[SPAN_BUCKET(ITEM($9, 'salary'),"
+            + " 10000)])\n"
+            + "        CalciteLogicalIndexScan(table=[[OpenSearch, test_dynamic_fields]])\n"
+            + "  physical: |\n"
+            + "    EnumerableCalc(expr#0..9=[{inputs}], expr#10=['salary'], expr#11=[ITEM($t9,"
+            + " $t10)], expr#12=[10000], expr#13=[SPAN_BUCKET($t11, $t12)], account_number=[$t0],"
+            + " salary=[$t13])\n"
+            + "      EnumerableLimit(fetch=[200])\n"
+            + "        CalciteEnumerableIndexScan(table=[[OpenSearch, test_dynamic_fields]])\n");
+
+    verifySchema(result, schema("account_number", "bigint"), schema("salary", "string"));
     verifyDataRows(
-        result, rows(1, "John", "Doe", "70000-80000", "NYC", "{\"n\":1}", "Engineering"));
+        result,
+        rows(1, "70000-80000"),
+        rows(2, "60000-70000"),
+        rows(3, null),
+        rows(4, "80000-90000"),
+        rows(5, "60000-70000"));
   }
 
   @Test
-  public void testCastAndPatterns() throws IOException {
+  public void testPatterns() throws IOException {
     String query =
         source(
             TEST_INDEX_DYNAMIC,
-            "eval department=CAST(department as string) | patterns department method=simple_pattern"
+            "patterns department method=simple_pattern"
                 + " | fields department, patterns_field | head 1");
     JSONObject result = executeQuery(query);
 
@@ -265,12 +267,12 @@ public void testCastAndPatterns() throws IOException {
   }
 
   @Test
-  public void testCastAndPatternsWithAggregation() throws IOException {
+  public void testPatternsWithAggregation() throws IOException {
     // TODO:
     String query =
         source(
             TEST_INDEX_DYNAMIC,
-            "eval department=CAST(department as string) | patterns department mode=aggregation"
+            "patterns department mode=aggregation"
                 + " method=simple_pattern | fields patterns_field, pattern_count, sample_logs |"
                 + " head 1");
     JSONObject result = executeQuery(query);
@@ -490,34 +492,29 @@ public void testEval() throws IOException {
     String query =
         source(
             TEST_INDEX_DYNAMIC,
-            "eval salary = cast(salary as int) * 2 | fields firstname,"
-                + " lastname, salary | head 1");
+            "eval salary = 1 + salary * 2 | fields account_number, salary | head 1");
 
     assertExplainYaml(
         query,
         "calcite:\n"
             + "  logical: |\n"
             + "    LogicalSystemLimit(fetch=[200], type=[QUERY_SIZE_LIMIT])\n"
             + "      LogicalSort(fetch=[1])\n"
-            + "        LogicalProject(firstname=[$1], lastname=[$2], salary=[*(SAFE_CAST(ITEM($9,"
-            + " 'salary')), 2)])\n"
+            + "        LogicalProject(account_number=[$0], salary=[+(1, *(SAFE_CAST(ITEM($9,"
+            + " 'salary')), 2.0E0:DOUBLE))])\n"
             + "          CalciteLogicalIndexScan(table=[[OpenSearch, test_dynamic_fields]])\n"
             + "  physical: |\n"
             + "    EnumerableLimit(fetch=[200])\n"
-            + "      EnumerableCalc(expr#0..9=[{inputs}], expr#10=['salary'], expr#11=[ITEM($t9,"
-            + " $t10)], expr#12=[SAFE_CAST($t11)], expr#13=[2], expr#14=[*($t12, $t13)],"
-            + " firstname=[$t1], lastname=[$t2], salary=[$t14])\n"
+            + "      EnumerableCalc(expr#0..9=[{inputs}], expr#10=[1], expr#11=['salary'],"
+            + " expr#12=[ITEM($t9, $t11)], expr#13=[SAFE_CAST($t12)], expr#14=[2.0E0:DOUBLE],"
+            + " expr#15=[*($t13, $t14)], expr#16=[+($t10, $t15)], account_number=[$t0],"
+            + " salary=[$t16])\n"
             + "        EnumerableLimit(fetch=[1])\n"
-            + "          CalciteEnumerableIndexScan(table=[[OpenSearch, test_dynamic_fields]])\n"
-            + "");
+            + "          CalciteEnumerableIndexScan(table=[[OpenSearch, test_dynamic_fields]])\n");
 
     JSONObject result = executeQuery(query);
 
-    verifySchema(
-        result,
-        schema("firstname", "string"),
-        schema("lastname", "string"),
-        schema("salary", "int"));
+    verifySchema(result, schema("account_number", "bigint"), schema("salary", "double"));
   }
 
   private void createTestIndexWithUnmappedFields() throws IOException {
