Decouple PPL logical plans from UDT temporal types

Logical RelNode/RexNode trees now use standard Calcite DATE/TIME(9)/
TIMESTAMP(9) types for date columns. A new TemporalUdtRewriteShuttle
runs at the prepare-statement boundary (OpenSearchCalcitePreparingStmt
.implement and OpenSearchRelRunners.run) and rewrites the standard
temporal types back to UDTs (ExprDateType/ExprTimeType/ExprTimeStampType)
just before physical execution, so Linq4j keeps receiving the
VARCHAR-backed representation.

This commit folds in the prior "Decouple Calcite PPL planning from
ExprType" change as well: Calcite-side PPL planning (RelNode/RexNode
code, coercion, type checking, and UDF implementations) operates on
RelDataType throughout instead of bouncing through ExprType.

Type system:
- OpenSearchTypeFactory.convertExprTypeToRelDataType returns standard
  TIMESTAMP(9)/DATE/TIME(9) for the corresponding ExprCoreType. IP and
  BINARY remain UDT.
- New helper isStandardTemporalType.
- PPLOperandTypes constants renamed DATE_UDT/TIME_UDT/TIMESTAMP_UDT to
  DATE_T/TIME_T/TIMESTAMP_T.
- UserDefinedFunctionUtils NULLABLE_*_UDT renamed to NULLABLE_*_T and
  repointed to standard temporal RelDataTypes (IP and BINARY constants
  unchanged).
- UDTs are recognised via instanceof rather than getExprType(); cloneWith
  preserves UDT identity through createTypeWithNullability.

Cast emission and pushdown:
- CalciteRexNodeVisitor.visitCast emits standard temporal types for
  AST DATE/TIME/TIMESTAMP cast targets.
- ExtendedRexBuilder.makeCast routes IP separately; standard temporal
  targets dispatch to PPLBuiltinOperators.DATE/TIME/TIMESTAMP with the
  standard target type so the call's RelDataType stays standard.
- PredicateAnalyzer.isTimestamp/isDate accept both standard SqlTypeName
  and UDT, and read literal values via getValueAs(String.class) so
  TimestampString/DateString round-trip without ClassCastException.

Coercion + type-checker:
- RelDataType-typed common-type resolver with a CoercionTag widening DAG.
- CoercionUtils' DATE+TIME -> TIMESTAMP resolver emits standard
  TIMESTAMP(9).
- PPLTypeChecker gains a temporalKind helper used in typesMatch and
  isComparable so standard and UDT temporal pairs match across forms.
- getRelDataTypes(family) returns standard temporal types for
  DATETIME/TIMESTAMP/DATE/TIME families (BINARY family stays UDT).
- UDF return-type inference (AddSubDate/Weekday/LastDay/TimestampDiff
  /Format/TimestampAdd/Extract/Span/WidthBucket) recognise both
  standard and UDT temporal operand types.

Shuttle implementation:
- Atomic rebuild path for Project/Filter/Calc/Aggregate/Values nodes
  (Calcite's default copy-then-validate path doesn't work because each
  half of the rewrite would briefly violate row-type consistency).
- Filter rebuild preserves variablesSet so correlated subqueries keep
  their CorrelationId binding.
- TemporalSchemaRewritable marker interface lets OpenSearch table
  scans rewrite their schema in place rather than wrap in a
  LogicalProject(CAST), so Calcite Linq4j codegen reads String values
  matching what OpenSearchExprValueFactory delivers at runtime.

Other:
- DatetimeUdtNormalizeRule (api/datetime extension) recognises both
  UDT and standard temporal RexCalls and forces precision to MAX so
  unified-API consumers see TIMESTAMP(9).
- Calcite IT golden files updated to reflect logical plans now showing
  TIMESTAMP(9) instead of EXPR_TIMESTAMP VARCHAR (physical plans still
  show UDT post-shuttle).

Tests:
- New unit: TemporalUdtRewriteShuttleTest, CalcitePPLLogicalPlanStandardTemporalTest.
- Updated CoercionUtilsTest, OpenSearchTypeFactoryTest. All module unit
  tests pass. Full Calcite IT suite (including ExplainIT) green.

Signed-off-by: Peng Huo <penghuo@gmail.com>

Author: penghuo

api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeExtension.java

@@ -14,8 +14,9 @@
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.opensearch.sql.api.spec.LanguageSpec.LanguageExtension;
-import org.opensearch.sql.calcite.type.AbstractExprRelDataType;
-import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT;
+import org.opensearch.sql.calcite.type.ExprDateType;
+import org.opensearch.sql.calcite.type.ExprTimeStampType;
+import org.opensearch.sql.calcite.type.ExprTimeType;
 
 /** Datetime language extension that normalizes datetime UDT types to standard Calcite types. */
 public class DatetimeExtension implements LanguageExtension {
@@ -30,20 +31,21 @@ public List<RelShuttle> postAnalysisRules() {
   @Getter
   @RequiredArgsConstructor
   enum UdtMapping {
-    DATE(ExprUDT.EXPR_DATE, SqlTypeName.DATE),
-    TIME(ExprUDT.EXPR_TIME, SqlTypeName.TIME),
-    TIMESTAMP(ExprUDT.EXPR_TIMESTAMP, SqlTypeName.TIMESTAMP);
+    DATE(ExprDateType.class, SqlTypeName.DATE),
+    TIME(ExprTimeType.class, SqlTypeName.TIME),
+    TIMESTAMP(ExprTimeStampType.class, SqlTypeName.TIMESTAMP);
 
-    private final ExprUDT udtType;
+    private final Class<?> udtClass;
     private final SqlTypeName stdType;
 
     /** Matches a UDT RelDataType to its mapping, or empty if not a datetime UDT. */
     static Optional<UdtMapping> fromUdtType(RelDataType type) {
-      if (!(type instanceof AbstractExprRelDataType<?> e)) {
-        return Optional.empty();
-      }
-      ExprUDT udt = e.getUdt();
-      return Arrays.stream(values()).filter(u -> u.udtType == udt).findFirst();
+      return Arrays.stream(values()).filter(u -> u.udtClass.isInstance(type)).findFirst();
+    }
+
+    /** Returns true if the given SqlTypeName is a standard datetime type. */
+    static boolean isDatetimeType(SqlTypeName typeName) {
+      return Arrays.stream(values()).anyMatch(u -> u.stdType == typeName);
     }
   }
 }

api/src/main/java/org/opensearch/sql/api/spec/datetime/DatetimeUdtNormalizeRule.java

@@ -36,21 +36,29 @@ public RelNode visit(RelNode other) {
           @Override
           public RexNode visitCall(RexCall call) {
             call = (RexCall) super.visitCall(call);
-            Optional<UdtMapping> mapping = UdtMapping.fromUdtType(call.getType());
-            if (mapping.isEmpty()) {
+            RelDataType callType = call.getType();
+            Optional<UdtMapping> mapping = UdtMapping.fromUdtType(callType);
+            SqlTypeName stdTypeName;
+            if (mapping.isPresent()) {
+              // Normalize UDT return type to standard Calcite DATE/TIME/TIMESTAMP
+              stdTypeName = mapping.get().getStdType();
+            } else if (UdtMapping.isDatetimeType(callType.getSqlTypeName())) {
+              // Normalize standard temporal types to canonical max precision
+              stdTypeName = callType.getSqlTypeName();
+            } else {
               return call;
             }
 
-            // Normalize UDT return type to standard Calcite DATE/TIME/TIMESTAMP
-            UdtMapping m = mapping.get();
-            SqlTypeName stdTypeName = m.getStdType();
             RelDataType baseType =
                 stdTypeName.allowsPrec()
                     ? typeFactory.createSqlType(
                         stdTypeName, typeFactory.getTypeSystem().getMaxPrecision(stdTypeName))
                     : typeFactory.createSqlType(stdTypeName);
             RelDataType stdType =
-                typeFactory.createTypeWithNullability(baseType, call.getType().isNullable());
+                typeFactory.createTypeWithNullability(baseType, callType.isNullable());
+            if (stdType.equals(callType)) {
+              return call;
+            }
             return call.clone(stdType, call.getOperands());
           }
         });

api/src/test/java/org/opensearch/sql/api/UnifiedQueryPlannerTest.java

@@ -147,20 +147,6 @@ public void invalidTableIsRethrownAsSemanticCheckException() {
         .assertCauseType(CalciteException.class);
   }
 
-  @Test
-  public void assertionErrorIsWrappedAsSemanticCheckException() {
-    // Remove when the underlying Calcite assertion is fixed.
-    givenInvalidQuery(
-            """
-            source = catalog.employees
-            | eval ts = timestamp('2024-01-01')
-            | stats max(ts)
-            """)
-        .assertErrorType(SemanticCheckException.class)
-        .assertErrorMessageEquals("Failed to plan query: invalid plan structure")
-        .assertCauseType(AssertionError.class);
-  }
-
   /**
    * Without the {@code PATTERN_*} defaults in {@link UnifiedQueryContext}, a bare {@code patterns
    * <field>} (no explicit {@code method=}/{@code mode=}) dies at parse time with {@code

core/src/main/java/org/opensearch/sql/calcite/CalciteRexNodeVisitor.java

@@ -82,6 +82,7 @@
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit;
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit.SystemLimitType;
 import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory;
+import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT;
 import org.opensearch.sql.calcite.utils.PlanUtils;
 import org.opensearch.sql.calcite.utils.SubsearchUtils;
 import org.opensearch.sql.common.utils.StringUtils;
@@ -819,10 +820,30 @@ private RelNode resolveSubqueryPlan(
   @Override
   public RexNode visitCast(Cast node, CalcitePlanContext context) {
     RexNode expr = analyze(node.getExpression(), context);
-    RelDataType type =
-        OpenSearchTypeFactory.convertExprTypeToRelDataType(node.getDataType().getCoreType());
     RelDataType nullableType =
-        context.rexBuilder.getTypeFactory().createTypeWithNullability(type, true);
+        switch (node.getDataType()) {
+          case TYPE_ERROR ->
+              throw new IllegalArgumentException("Unsupported AST DataType: " + node.getDataType());
+          case NULL -> TYPE_FACTORY.createSqlType(SqlTypeName.NULL, true);
+          case INTEGER -> TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER, true);
+          case LONG -> TYPE_FACTORY.createSqlType(SqlTypeName.BIGINT, true);
+          case SHORT -> TYPE_FACTORY.createSqlType(SqlTypeName.SMALLINT, true);
+          case FLOAT -> TYPE_FACTORY.createSqlType(SqlTypeName.REAL, true);
+          case DOUBLE, DECIMAL -> TYPE_FACTORY.createSqlType(SqlTypeName.DOUBLE, true);
+          case STRING -> TYPE_FACTORY.createSqlType(SqlTypeName.VARCHAR, true);
+          case BOOLEAN -> TYPE_FACTORY.createSqlType(SqlTypeName.BOOLEAN, true);
+          case DATE -> TYPE_FACTORY.createSqlType(SqlTypeName.DATE, true);
+          case TIME ->
+              TYPE_FACTORY.createTypeWithNullability(
+                  TYPE_FACTORY.createSqlType(SqlTypeName.TIME, 9), true);
+          case TIMESTAMP ->
+              TYPE_FACTORY.createTypeWithNullability(
+                  TYPE_FACTORY.createSqlType(SqlTypeName.TIMESTAMP, 9), true);
+          case INTERVAL ->
+              throw new IllegalArgumentException(
+                  "INTERVAL must carry a unit; cannot map to RelDataType directly.");
+          case IP -> TYPE_FACTORY.createUDT(ExprUDT.EXPR_IP, true);
+        };
     // call makeCast() instead of cast() because the saft parameter is true could avoid exception.
     return context.rexBuilder.makeCast(nullableType, expr, true, true);
   }

core/src/main/java/org/opensearch/sql/calcite/ExtendedRexBuilder.java

@@ -20,11 +20,8 @@
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.type.SqlTypeUtil;
 import org.opensearch.sql.ast.expression.SpanUnit;
-import org.opensearch.sql.calcite.type.AbstractExprRelDataType;
-import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory;
-import org.opensearch.sql.data.type.ExprCoreType;
+import org.opensearch.sql.calcite.type.ExprIPType;
 import org.opensearch.sql.exception.ExpressionEvaluationException;
-import org.opensearch.sql.exception.SemanticCheckException;
 import org.opensearch.sql.expression.function.PPLBuiltinOperators;
 
 public class ExtendedRexBuilder extends RexBuilder {
@@ -146,35 +143,39 @@ public RexNode makeCast(
         //            SqlStdOperatorTable.NOT_EQUALS,
         //            ImmutableList.of(exp, makeZeroLiteral(sourceType)));
       }
-    } else if (OpenSearchTypeFactory.isUserDefinedType(type)) {
+    } else if (type instanceof ExprIPType) {
       if (RexLiteral.isNullLiteral(exp)) {
         return super.makeCast(pos, type, exp, matchNullability, safe, format);
       }
-      var udt = ((AbstractExprRelDataType<?>) type).getUdt();
-      var argExprType = OpenSearchTypeFactory.convertRelDataTypeToExprType(sourceType);
-      return switch (udt) {
-        case EXPR_DATE -> makeCall(type, PPLBuiltinOperators.DATE, List.of(exp));
-        case EXPR_TIME -> makeCall(type, PPLBuiltinOperators.TIME, List.of(exp));
-        case EXPR_TIMESTAMP -> makeCall(type, PPLBuiltinOperators.TIMESTAMP, List.of(exp));
-        case EXPR_IP -> {
-          if (argExprType == ExprCoreType.IP) {
-            yield exp;
-          } else if (argExprType == ExprCoreType.STRING) {
-            yield makeCall(type, PPLBuiltinOperators.IP, List.of(exp));
-          }
-          // Throwing error inside implementation will be suppressed by Calcite, thus
-          // throwing 500 error. Therefore, we throw error here to ensure the error
-          // information is displayed properly.
-          throw new ExpressionEvaluationException(
-              String.format(
-                  Locale.ROOT,
-                  "Cannot convert %s to IP, only STRING and IP types are supported",
-                  argExprType));
-        }
-        default ->
-            throw new SemanticCheckException(
-                String.format(Locale.ROOT, "Cannot cast from %s to %s", argExprType, udt.name()));
-      };
+      if (sourceType instanceof ExprIPType) {
+        return exp;
+      } else if (SqlTypeUtil.isCharacter(sourceType)) {
+        return makeCall(type, PPLBuiltinOperators.IP, List.of(exp));
+      }
+      // Throwing error inside implementation will be suppressed by Calcite, thus
+      // throwing 500 error. Therefore, we throw error here to ensure the error
+      // information is displayed properly.
+      throw new ExpressionEvaluationException(
+          String.format(
+              Locale.ROOT,
+              "Cannot convert %s to IP, only STRING and IP types are supported",
+              sourceType));
+    } else if (sqlType == SqlTypeName.DATE
+        || sqlType == SqlTypeName.TIME
+        || sqlType == SqlTypeName.TIMESTAMP) {
+      // Standard SQL temporal target. Lower to PPL UDF so the runtime String → temporal parsing
+      // matches v2 semantics. The shuttle in OpenSearchCalcitePreparingStmt rewrites these to UDT
+      // at execution time.
+      if (RexLiteral.isNullLiteral(exp)) {
+        return super.makeCast(pos, type, exp, matchNullability, safe, format);
+      }
+      if (sqlType == SqlTypeName.DATE) {
+        return makeCall(type, PPLBuiltinOperators.DATE, List.of(exp));
+      } else if (sqlType == SqlTypeName.TIME) {
+        return makeCall(type, PPLBuiltinOperators.TIME, List.of(exp));
+      } else {
+        return makeCall(type, PPLBuiltinOperators.TIMESTAMP, List.of(exp));
+      }
     }
     // Use a custom operator when casting floating point or decimal number to a character type.
     // This patch is necessary because in Calcite, 0.0F is cast to 0E0, decimal 0.x to x

core/src/main/java/org/opensearch/sql/calcite/plan/rel/TemporalSchemaRewritable.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.calcite.plan.rel;
+
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+
+/**
+ * Marker interface for OpenSearch RelNodes whose row type can be rewritten in place.
+ *
+ * <p>The {@link TemporalUdtRewriteShuttle} uses this interface to convert a scan's schema from
+ * standard Calcite temporal types ({@code TIMESTAMP}, {@code DATE}, {@code TIME}) to the equivalent
+ * UDT-backed types ({@code ExprTimeStampType}/{@code ExprDateType}/{@code ExprTimeType}, all
+ * VARCHAR-backed at runtime) just before physical execution.
+ *
+ * <p>Wrapping a TableScan in a {@code LogicalProject(CAST...)} is not sufficient: Calcite's Linq4j
+ * codegen reads the TableScan's row type to generate field accessors, and any standard temporal
+ * type triggers {@code (Long) row[i]} casts at runtime, which fail when the runtime actually
+ * delivers VARCHAR values via {@code OpenSearchExprValueFactory}. Implementations of this interface
+ * must therefore return a NEW RelNode whose row type itself reflects the rewritten UDT schema (not
+ * a wrapper that hides it from Calcite).
+ */
+public interface TemporalSchemaRewritable {
+  /**
+   * Return a copy of this RelNode whose row type is replaced by {@code rowType}. The structural
+   * shape of the returned node should otherwise mirror {@code this}.
+   */
+  RelNode withRowType(RelDataType rowType);
+}