feat(api): Add datetime UDT extension for unified query API

Bridge the type mismatch between PPL UDT types (String-based) and standard
Calcite types (int/long-based) in the unified query API path by contributing
two post-analysis rules to PPL's LanguageSpec:

1. DatetimeUdtNormalizeRule rewrites UDT calls — replaces UDT return types
   with standard Calcite types and wraps UDF implementors to convert values
   at input (int/long -> String) and output (String -> int/long).
2. DatetimeUdtOutputCastRule wraps the plan root with a projection that
   casts remaining datetime output columns to VARCHAR so the wire format
   matches PPL's String datetime contract.

Both rules are registered via DatetimeUdtExtension, which encapsulates the
ordering invariant (normalize before cast). The extension plugs into the
LanguageExtension mechanism introduced in #5360 via a new postAnalysisRules
hook, applied once at the top of UnifiedQueryPlanner.plan() after the
language-specific strategy returns.

Applied only on the PPL path; zero impact on the SQL or OpenSearch plugin
paths.

Signed-off-by: Chen Dai <daichen@amazon.com>

Author: dai-chen

api/src/main/java/org/opensearch/sql/api/UnifiedQueryPlanner.java

@@ -60,7 +60,15 @@ public UnifiedQueryPlanner(UnifiedQueryContext context) {
    */
   public RelNode plan(String query) {
     try {
-      return context.measure(ANALYZE, () -> strategy.plan(query));
+      return context.measure(
+          ANALYZE,
+          () -> {
+            RelNode plan = strategy.plan(query);
+            for (var rule : context.getLangSpec().postAnalysisRules()) {
+              plan = rule.apply(plan);
+            }
+            return plan;
+          });
     } catch (SyntaxCheckException | UnsupportedOperationException e) {
       throw e;
     } catch (Exception e) {

api/src/main/java/org/opensearch/sql/api/spec/LanguageSpec.java

@@ -7,6 +7,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlOperatorTable;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
@@ -17,8 +18,8 @@
 
 /**
  * Language specification defining the dialect the engine accepts. Provides parser configuration,
- * validator configuration, and composable {@link LanguageExtension}s that contribute operators and
- * post-parse rewrite rules.
+ * validator configuration, and composable {@link LanguageExtension}s that contribute operators,
+ * post-parse rewrite rules, and post-analysis rewrite rules.
  *
  * <p>Implementations define a complete language surface — for example, {@link UnifiedSqlSpec}
  * provides ANSI and extended SQL modes. A future PPL spec would implement this same interface once
@@ -27,8 +28,18 @@
 public interface LanguageSpec {
 
   /**
-   * A composable language extension that contributes operators and post-parse rewrite rules. All
-   * methods have defaults so extensions only override what they need.
+   * A RelNode rewrite rule applied after analysis and before execution. Takes a logical plan and
+   * returns a rewritten plan.
+   */
+  @FunctionalInterface
+  interface PostAnalysisRule {
+    RelNode apply(RelNode plan);
+  }
+
+  /**
+   * A composable language extension that contributes operators, post-parse rewrite rules, and
+   * post-analysis rewrite rules. All methods have defaults so extensions only override what they
+   * need.
    */
   interface LanguageExtension {
 
@@ -47,6 +58,15 @@ default SqlOperatorTable operators() {
     default List<SqlVisitor<SqlNode>> postParseRules() {
       return List.of();
     }
+
+    /**
+     * RelNode rewrite rules applied after analysis and before execution. Rules within a single
+     * extension are applied in list order; extensions that depend on ordering should return their
+     * rules together from one extension rather than relying on cross-extension ordering.
+     */
+    default List<PostAnalysisRule> postAnalysisRules() {
+      return List.of();
+    }
   }
 
   /**
@@ -62,9 +82,9 @@ default List<SqlVisitor<SqlNode>> postParseRules() {
   SqlValidator.Config validatorConfig();
 
   /**
-   * Language extensions registered with this spec. Each extension contributes operators and
-   * post-parse rewrite rules that are composed by {@link #operatorTable()} and {@link
-   * #postParseRules()}.
+   * Language extensions registered with this spec. Each extension contributes operators, post-parse
+   * rewrite rules, and post-analysis rewrite rules composed by {@link #operatorTable()}, {@link
+   * #postParseRules()}, and {@link #postAnalysisRules()}.
    */
   List<LanguageExtension> extensions();
 
@@ -86,4 +106,12 @@ default SqlOperatorTable operatorTable() {
   default List<SqlVisitor<SqlNode>> postParseRules() {
     return extensions().stream().flatMap(ext -> ext.postParseRules().stream()).toList();
   }
+
+  /**
+   * All post-analysis RelNode rewrite rules from registered extensions, flattened in registration
+   * order. Applied to the logical plan after analysis and before execution.
+   */
+  default List<PostAnalysisRule> postAnalysisRules() {
+    return extensions().stream().flatMap(ext -> ext.postAnalysisRules().stream()).toList();
+  }
 }

api/src/main/java/org/opensearch/sql/api/spec/UnifiedPplSpec.java

@@ -10,6 +10,7 @@
 import lombok.NoArgsConstructor;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.calcite.sql.validate.SqlValidator;
+import org.opensearch.sql.api.spec.datetime.DatetimeUdtExtension;
 
 /**
  * PPL language specification.
@@ -37,6 +38,6 @@ public SqlValidator.Config validatorConfig() {
 
   @Override
   public List<LanguageExtension> extensions() {
-    return List.of();
+    return List.of(new DatetimeUdtExtension());
   }
 }

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

@@ -0,0 +1,92 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.api.spec.datetime;
+
+import java.util.List;
+import java.util.Optional;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.apache.calcite.avatica.util.DateTimeUtils;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.linq4j.tree.Expressions;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.opensearch.sql.api.spec.LanguageSpec.LanguageExtension;
+import org.opensearch.sql.api.spec.LanguageSpec.PostAnalysisRule;
+import org.opensearch.sql.calcite.type.AbstractExprRelDataType;
+import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT;
+
+/**
+ * Normalizes datetime UDT operations in the logical plan and casts remaining datetime output
+ * columns to VARCHAR so the wire output matches PPL's String datetime contract.
+ *
+ * <p>Contributes two ordered post-analysis rules: {@link DatetimeUdtNormalizeRule} rewrites UDT
+ * calls; {@link DatetimeUdtOutputCastRule} wraps the root with a varchar projection. The cast
+ * depends on the normalized row type, so both rules live in a single extension to keep their
+ * ordering encapsulated.
+ */
+public class DatetimeUdtExtension implements LanguageExtension {
+
+  @Override
+  public List<PostAnalysisRule> postAnalysisRules() {
+    return List.of(new DatetimeUdtNormalizeRule(), new DatetimeUdtOutputCastRule());
+  }
+
+  /** Maps a datetime UDT to its standard Calcite equivalent with value conversion methods. */
+  @Getter
+  @RequiredArgsConstructor
+  enum UdtMapping {
+    DATE(ExprUDT.EXPR_DATE, SqlTypeName.DATE, "dateStringToUnixDate", "unixDateToString"),
+    TIME(ExprUDT.EXPR_TIME, SqlTypeName.TIME, "timeStringToUnixDate", "unixTimeToString"),
+    TIMESTAMP(
+        ExprUDT.EXPR_TIMESTAMP,
+        SqlTypeName.TIMESTAMP,
+        "timestampStringToUnixDate",
+        "unixTimestampToString");
+
+    private final ExprUDT udtType;
+    private final SqlTypeName stdType;
+    private final String toStdMethod;
+    private final String fromStdMethod;
+
+    /** Matches a UDT type to its mapping. */
+    static Optional<UdtMapping> fromUdtType(RelDataType type) {
+      if (!(type instanceof AbstractExprRelDataType<?> e)) return Optional.empty();
+      ExprUDT udt = e.getUdt();
+      for (UdtMapping u : values()) {
+        if (u.udtType == udt) return Optional.of(u);
+      }
+      return Optional.empty();
+    }
+
+    /** Matches a standard Calcite type to its mapping. */
+    static Optional<UdtMapping> fromStdType(RelDataType type) {
+      SqlTypeName name = type.getSqlTypeName();
+      for (UdtMapping u : values()) {
+        if (u.stdType == name) return Optional.of(u);
+      }
+      return Optional.empty();
+    }
+
+    RelDataType toStdType(RexBuilder rexBuilder, boolean nullable) {
+      return rexBuilder
+          .getTypeFactory()
+          .createTypeWithNullability(rexBuilder.getTypeFactory().createSqlType(stdType), nullable);
+    }
+
+    /** UDT value (String) → standard value (int/long). */
+    Expression toStdValue(Expression result) {
+      return Expressions.call(
+          DateTimeUtils.class, toStdMethod, Expressions.call(result, "toString"));
+    }
+
+    /** Standard value (int/long) → UDT value (String). */
+    Expression fromStdValue(Expression operand) {
+      return Expressions.call(DateTimeUtils.class, fromStdMethod, operand);
+    }
+  }
+}

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

@@ -0,0 +1,115 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.api.spec.datetime;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import org.apache.calcite.adapter.enumerable.NotNullImplementor;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.rel.RelHomogeneousShuttle;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlReturnTypeInference;
+import org.apache.calcite.sql.validate.SqlUserDefinedFunction;
+import org.opensearch.sql.api.spec.LanguageSpec.PostAnalysisRule;
+import org.opensearch.sql.api.spec.datetime.DatetimeUdtExtension.UdtMapping;
+import org.opensearch.sql.expression.function.ImplementorUDF;
+import org.opensearch.sql.expression.function.ImplementorUDF.ImplementableUDFunction;
+import org.opensearch.sql.expression.function.UDFOperandMetadata;
+
+/**
+ * Normalizes UDT types in the plan by replacing UDT return types with standard Calcite types
+ * (signature) and wrapping UDF implementors to convert between UDT and standard values
+ * (implementation).
+ */
+public class DatetimeUdtNormalizeRule implements PostAnalysisRule {
+
+  @Override
+  public RelNode apply(RelNode plan) {
+    RexBuilder rexBuilder = plan.getCluster().getRexBuilder();
+    return plan.accept(
+        new RelHomogeneousShuttle() {
+          @Override
+          public RelNode visit(RelNode other) {
+            return super.visit(other).accept(new UdtRexShuttle(rexBuilder));
+          }
+        });
+  }
+
+  private static class UdtRexShuttle extends RexShuttle {
+
+    private final RexBuilder rexBuilder;
+
+    UdtRexShuttle(RexBuilder rexBuilder) {
+      this.rexBuilder = rexBuilder;
+    }
+
+    @Override
+    public RexNode visitCall(RexCall call) {
+      RexCall visited = (RexCall) super.visitCall(call);
+
+      if (!(visited.getOperator() instanceof SqlUserDefinedFunction udf
+          && udf.getFunction() instanceof ImplementableUDFunction func)) {
+        return visited;
+      }
+
+      Optional<UdtMapping> returnType = UdtMapping.fromUdtType(visited.getType());
+      if (returnType.isEmpty()
+          && visited.getOperands().stream()
+              .noneMatch(op -> UdtMapping.fromStdType(op.getType()).isPresent())) {
+        return visited;
+      }
+
+      RelDataType normReturnType = normalizeReturnType(rexBuilder, visited, returnType);
+      NotNullImplementor normFuncImpl = normalizeImplementation(visited, func, returnType);
+      SqlUserDefinedFunction normUdf =
+          new ImplementorUDF(normFuncImpl, func.getNullPolicy()) {
+            @Override
+            public SqlReturnTypeInference getReturnTypeInference() {
+              return returnType
+                  .<SqlReturnTypeInference>map(u -> ReturnTypes.explicit(normReturnType))
+                  .orElse(udf.getReturnTypeInference());
+            }
+
+            @Override
+            public UDFOperandMetadata getOperandMetadata() {
+              return (UDFOperandMetadata) udf.getOperandTypeChecker();
+            }
+          }.toUDF(udf.getName(), udf.isDeterministic());
+      return rexBuilder.makeCall(normReturnType, normUdf, visited.getOperands());
+    }
+  }
+
+  /** Replace UDT return type with standard Calcite type. */
+  private static RelDataType normalizeReturnType(
+      RexBuilder rexBuilder, RexCall call, Optional<UdtMapping> returnUdt) {
+    return returnUdt
+        .map(u -> u.toStdType(rexBuilder, call.getType().isNullable()))
+        .orElse(call.getType());
+  }
+
+  /** Wrap implementor to convert inputs (standard → UDT) and output (UDT → standard). */
+  private static NotNullImplementor normalizeImplementation(
+      RexCall originalCall, ImplementableUDFunction func, Optional<UdtMapping> returnUdt) {
+    return (translator, rexCall, operands) -> {
+      List<Expression> converted = new ArrayList<>(operands.size());
+      for (int i = 0; i < operands.size(); i++) {
+        Expression operand = operands.get(i);
+        RelDataType opType = originalCall.getOperands().get(i).getType();
+        converted.add(
+            UdtMapping.fromStdType(opType).map(u -> u.fromStdValue(operand)).orElse(operand));
+      }
+      Expression result = func.getNotNullImplementor().implement(translator, rexCall, converted);
+      return returnUdt.map(u -> u.toStdValue(result)).orElse(result);
+    };
+  }
+}

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

@@ -0,0 +1,56 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.api.spec.datetime;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.logical.LogicalProject;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.opensearch.sql.api.spec.LanguageSpec.PostAnalysisRule;
+import org.opensearch.sql.api.spec.datetime.DatetimeUdtExtension.UdtMapping;
+
+/**
+ * Wraps the plan with a projection that casts standard datetime output columns to VARCHAR so the
+ * wire output matches PPL's String datetime contract. Runs after {@link DatetimeUdtNormalizeRule}
+ * has converted UDT columns to their standard datetime types.
+ */
+public class DatetimeUdtOutputCastRule implements PostAnalysisRule {
+
+  @Override
+  public RelNode apply(RelNode plan) {
+    List<RelDataTypeField> fields = plan.getRowType().getFieldList();
+    boolean anyDatetime =
+        fields.stream().anyMatch(f -> UdtMapping.fromStdType(f.getType()).isPresent());
+    if (!anyDatetime) {
+      return plan;
+    }
+
+    RexBuilder rexBuilder = plan.getCluster().getRexBuilder();
+    RelDataType varcharType = rexBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR);
+    List<RexNode> projects = new ArrayList<>(fields.size());
+    List<String> names = new ArrayList<>(fields.size());
+    for (RelDataTypeField field : fields) {
+      RexNode ref = rexBuilder.makeInputRef(plan, field.getIndex());
+      if (UdtMapping.fromStdType(field.getType()).isPresent()) {
+        RelDataType nullableVarchar =
+            rexBuilder
+                .getTypeFactory()
+                .createTypeWithNullability(varcharType, field.getType().isNullable());
+        projects.add(rexBuilder.makeCast(nullableVarchar, ref));
+      } else {
+        projects.add(ref);
+      }
+      names.add(field.getName());
+    }
+    return LogicalProject.create(plan, List.of(), projects, names, Set.of());
+  }
+}