Preserve datetime schema-column types in AnalyticsExecutionEngine

`DatetimeOutputCastRule` (api/spec/datetime) wraps every datetime-typed
root column in `CAST(... AS VARCHAR)` so the analytics-engine backend
can emit PPL's documented space-separator format via `to_char`
(`DatetimeOutputCastRewriter` on the OpenSearch side). The downside is
that `plan.getRowType()` then advertises those columns as VARCHAR — the
response schema reports `"string"` for what callers expect as
`"timestamp"`. The legacy / Calcite-reference path doesn't have this
divergence: the wire value is the formatted string while the schema
column type stays as the original datetime type.

Walk the outermost `Project` (descending through a single `Sort`, the
`LogicalSystemLimit` system query-size wrapper) and, for any slot of
shape `CAST(<datetime> AS VARCHAR)`, swap in the inner `RelDataType`
for the schema column. Values still come back as the strings DataFusion
emitted — only the schema column type is restored.

Falls back to the slot's reported type for non-cast slots, non-datetime
sources, or when the root isn't a Project (raw scan / aggregate fragment).

Surfaced by `verifySchema(... schema("@timestamp", "timestamp") ...)`
assertions in `CalciteSearchCommandIT` time-modifier tests; this commit
removes the schema-type divergence for those queries (the remaining
data-row mismatches are downstream Lucene-secondary issues filed
separately).

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

core/src/main/java/org/opensearch/sql/executor/analytics/AnalyticsExecutionEngine.java

@@ -6,13 +6,20 @@
 package org.opensearch.sql.executor.analytics;
 
 import java.util.ArrayList;
+import java.util.EnumSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import org.apache.calcite.plan.RelOptUtil;
 import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.opensearch.analytics.exec.QueryPlanExecutor;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.sql.ast.statement.ExplainMode;
@@ -87,7 +94,7 @@ public void onResponse(Iterable<Object[]> rows) {
             try {
               List<RelDataTypeField> fields = plan.getRowType().getFieldList();
               List<ExprValue> results = convertRows(rows, fields);
-              Schema schema = buildSchema(fields);
+              Schema schema = buildSchema(fields, recoverOriginalDatetimeTypes(plan));
               execMetric.set(System.nanoTime() - execStart);
               listener.onResponse(new QueryResponse(schema, results, Cursor.None));
             } catch (Exception e) {
@@ -132,11 +139,11 @@ private List<ExprValue> convertRows(Iterable<Object[]> rows, List<RelDataTypeFie
     return results;
   }
 
-  private Schema buildSchema(List<RelDataTypeField> fields) {
+  private Schema buildSchema(List<RelDataTypeField> fields, List<RelDataType> reportedTypes) {
     List<Schema.Column> columns = new ArrayList<>();
-    for (RelDataTypeField field : fields) {
-      ExprType exprType = convertType(field.getType());
-      columns.add(new Schema.Column(field.getName(), null, exprType));
+    for (int i = 0; i < fields.size(); i++) {
+      ExprType exprType = convertType(reportedTypes.get(i));
+      columns.add(new Schema.Column(fields.get(i).getName(), null, exprType));
     }
     return new Schema(columns);
   }
@@ -148,4 +155,68 @@ private ExprType convertType(RelDataType type) {
       return org.opensearch.sql.data.type.ExprCoreType.UNKNOWN;
     }
   }
+
+  /**
+   * Datetime SqlTypeNames produced by the SQL-plugin engine pipeline. Used by {@link
+   * #recoverOriginalDatetimeTypes} to detect {@code DatetimeOutputCastRule}'s {@code
+   * CAST(<datetime> AS VARCHAR)} output slots.
+   */
+  private static final EnumSet<SqlTypeName> DATETIME_TYPES =
+      EnumSet.of(
+          SqlTypeName.TIMESTAMP,
+          SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE,
+          SqlTypeName.DATE,
+          SqlTypeName.TIME);
+
+  /**
+   * Recovers the pre-cast datetime types for the schema. {@code DatetimeOutputCastRule} (api
+   * post-analysis pass) wraps every datetime-typed root column in {@code CAST(... AS VARCHAR)}
+   * so the analytics-engine backend can emit PPL's documented space-separator format via {@code
+   * to_char} ({@code DatetimeOutputCastRewriter} on the OpenSearch sandbox side). The downside
+   * is that {@code plan.getRowType()} then advertises those columns as VARCHAR — the response
+   * schema would report {@code "string"} for what tests (and callers) still expect as {@code
+   * "timestamp"}. The legacy / Calcite-reference path doesn't have this divergence: the wire
+   * value is a formatted string while the schema column type is the original datetime type.
+   *
+   * <p>This walk inspects the OUTERMOST {@link Project} (descending through a single {@link
+   * Sort}, the {@code LogicalSystemLimit} wrapper) and, for any slot of the exact shape {@code
+   * CAST(<datetime> AS VARCHAR)}, swaps in the inner {@link RelDataType} for the schema. Values
+   * still come back as the strings DataFusion emitted; only the schema column type is restored.
+   *
+   * <p>Falls back to the slot's reported type for non-cast slots, non-datetime sources, or when
+   * the root isn't a Project (raw scan / aggregate fragment).
+   */
+  private static List<RelDataType> recoverOriginalDatetimeTypes(RelNode plan) {
+    List<RelDataTypeField> fields = plan.getRowType().getFieldList();
+    List<RelDataType> reportedTypes = new ArrayList<>(fields.size());
+    for (RelDataTypeField field : fields) {
+      reportedTypes.add(field.getType());
+    }
+    RelNode root = plan;
+    if (root instanceof Sort sort && sort.getInput() instanceof Project) {
+      root = sort.getInput();
+    }
+    if (!(root instanceof Project project)) {
+      return reportedTypes;
+    }
+    List<RexNode> slots = project.getProjects();
+    int n = Math.min(slots.size(), reportedTypes.size());
+    for (int i = 0; i < n; i++) {
+      RexNode slot = slots.get(i);
+      if (!(slot instanceof RexCall call)) {
+        continue;
+      }
+      if (call.getKind() != SqlKind.CAST) {
+        continue;
+      }
+      if (call.getType().getSqlTypeName() != SqlTypeName.VARCHAR) {
+        continue;
+      }
+      RexNode source = call.getOperands().get(0);
+      if (DATETIME_TYPES.contains(source.getType().getSqlTypeName())) {
+        reportedTypes.set(i, source.getType());
+      }
+    }
+    return reportedTypes;
+  }
 }