Remove DatetimeOutputCastRule on the analytics-engine route (#5420)

On the analytics-engine route, every datetime root column is wrapped in
`CAST(<DATE/TIME/TIMESTAMP> AS VARCHAR)` by `DatetimeOutputCastRule`.
A matching `DatetimeOutputCastRewriter` on the engine side then translates
that cast into DataFusion's `to_char` extension. Whenever the rewriter's
format string and the PPL formatter disagree (e.g. trailing `Z`, `T`
separator), users see wire-format divergence — issue #5420.

Drop the cast rule and let the engine return real datetime cells.
The PPL response pipeline already handles datetime → string conversion
natively at the formatter layer:
- `AnalyticsExecutionEngine.convertRows` feeds engine cells
  (`LocalDateTime` / `LocalDate` / `LocalTime`) through
  `ExprValueUtils.fromObjectValue`, which produces
  `ExprTimestampValue` / `ExprDateValue` / `ExprTimeValue`.
- Their `value()` methods render the documented PPL space-separated
  format (`yyyy-MM-dd HH:mm:ss[.SSSSSSSSS]` etc.).

The companion change in `opensearch-project/OpenSearch` removes the
matching `DatetimeOutputCastRewriter` and its callsites.

- Delete `DatetimeOutputCastRule`.
- Drop the unused `UdtMapping.isDatetimeType` helper.
- Rewrite the four cast-shape assertions in `DatetimeExtensionTest`
  to assert the post-removal RelNode (no `CAST(... VARCHAR)` wrapper,
  schema reports `DATE` / `TIME` / `TIMESTAMP`).

Signed-off-by: Eric Wei <mengwei.eric@gmail.com>

Author: mengweieric

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

@@ -17,13 +17,13 @@
 import org.opensearch.sql.calcite.type.AbstractExprRelDataType;
 import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT;
 
-/** Datetime language extension that normalizes UDT types and casts output for wire-format. */
+/** Datetime language extension that normalizes datetime UDT types to standard Calcite types. */
 public class DatetimeExtension implements LanguageExtension {
 
   @Override
   public List<RelShuttle> postAnalysisRules() {
-    // Fresh instances per plan() because RelHomogeneousShuttle inherits a stateful stack.
-    return List.of(new DatetimeUdtNormalizeRule(), new DatetimeOutputCastRule());
+    // Fresh instance per plan() because RelHomogeneousShuttle inherits a stateful stack.
+    return List.of(new DatetimeUdtNormalizeRule());
   }
 
   /** Maps datetime UDT types to their standard Calcite equivalents. */
@@ -45,10 +45,5 @@ static Optional<UdtMapping> fromUdtType(RelDataType type) {
       ExprUDT udt = e.getUdt();
       return Arrays.stream(values()).filter(u -> u.udtType == udt).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/DatetimeOutputCastRule.java

@@ -20,8 +20,8 @@
 import org.opensearch.sql.executor.QueryType;
 
 /**
- * Tests that DatetimeExtension post-analysis rules (UDT normalization and output VARCHAR cast)
- * apply correctly to the SQL V2 parser path through CalciteRelNodeVisitor.
+ * Tests that the DatetimeExtension UDT-normalization post-analysis rule applies correctly to the
+ * SQL V2 parser path through CalciteRelNodeVisitor.
  */
 public class DatetimeExtensionSqlTest extends UnifiedQueryTestBase {
 
@@ -57,12 +57,11 @@ private Table createEventsTable() {
   }
 
   @Test
-  public void testAllStandardDatetimeTypesCastToVarchar() {
+  public void testStandardDatetimeTypesNotWrapped() {
     givenQuery("SELECT * FROM catalog.events")
         .assertPlan(
             """
-            LogicalProject(id=[$0], name=[$1], hire_date=[CAST($2):VARCHAR NOT NULL], start_time=[CAST($3):VARCHAR NOT NULL], created_at=[CAST($4):VARCHAR NOT NULL])
-              LogicalTableScan(table=[[catalog, events]])
+            LogicalTableScan(table=[[catalog, events]])
             """);
   }
 
@@ -71,9 +70,8 @@ public void testFilterWithTimestampLiteral() {
     givenQuery("SELECT * FROM catalog.events WHERE created_at > '2024-01-01T00:00:00Z'")
         .assertPlan(
             """
-            LogicalProject(id=[$0], name=[$1], hire_date=[CAST($2):VARCHAR NOT NULL], start_time=[CAST($3):VARCHAR NOT NULL], created_at=[CAST($4):VARCHAR NOT NULL])
-              LogicalFilter(condition=[>($4, TIMESTAMP('2024-01-01T00:00:00Z':VARCHAR))])
-                LogicalTableScan(table=[[catalog, events]])
+            LogicalFilter(condition=[>($4, TIMESTAMP('2024-01-01T00:00:00Z':VARCHAR))])
+              LogicalTableScan(table=[[catalog, events]])
             """)
         .assertReturnType("TIMESTAMP", TIMESTAMP, 9);
   }
@@ -83,9 +81,8 @@ public void testComparisonWithDatetimeUdf() {
     givenQuery("SELECT * FROM catalog.events WHERE created_at < DATE(name)")
         .assertPlan(
             """
-            LogicalProject(id=[$0], name=[$1], hire_date=[CAST($2):VARCHAR NOT NULL], start_time=[CAST($3):VARCHAR NOT NULL], created_at=[CAST($4):VARCHAR NOT NULL])
-              LogicalFilter(condition=[<($4, TIMESTAMP(DATE($1)))])
-                LogicalTableScan(table=[[catalog, events]])
+            LogicalFilter(condition=[<($4, TIMESTAMP(DATE($1)))])
+              LogicalTableScan(table=[[catalog, events]])
             """)
         .assertReturnType("DATE", DATE)
         .assertReturnType("TIMESTAMP", TIMESTAMP, 9);

api/src/test/java/org/opensearch/sql/api/spec/datetime/DatetimeExtensionSqlTest.java

@@ -63,7 +63,7 @@ private Table createEventsTable() {
   }
 
   @Test
-  public void testUdfResultNormalizedAndCastToVarchar() {
+  public void testUdfResultNormalized() {
     givenQuery(
             """
             source = catalog.events \
@@ -72,9 +72,8 @@ public void testUdfResultNormalizedAndCastToVarchar() {
             """)
         .assertPlan(
             """
-            LogicalProject(d=[CAST($0):VARCHAR], t=[CAST($1):VARCHAR], ts=[CAST($2):VARCHAR])
-              LogicalProject(d=[DATE($1)], t=[TIME($1)], ts=[TIMESTAMP($1)])
-                LogicalTableScan(table=[[catalog, events]])
+            LogicalProject(d=[DATE($1)], t=[TIME($1)], ts=[TIMESTAMP($1)])
+              LogicalTableScan(table=[[catalog, events]])
             """)
         .assertReturnType("DATE", DATE)
         .assertReturnType("TIME", TIME, 9)
@@ -94,13 +93,12 @@ public void testNestedUdfCallsNormalized() {
   }
 
   @Test
-  public void testDateLiteralCastToVarchar() {
+  public void testDateLiteralNormalized() {
     givenQuery("source = catalog.events | eval d = DATE('2024-01-01') | fields d")
         .assertPlan(
             """
-            LogicalProject(d=[CAST($0):VARCHAR])
-              LogicalProject(d=[DATE('2024-01-01':VARCHAR)])
-                LogicalTableScan(table=[[catalog, events]])
+            LogicalProject(d=[DATE('2024-01-01':VARCHAR)])
+              LogicalTableScan(table=[[catalog, events]])
             """)
         .assertReturnType("DATE", DATE);
   }
@@ -134,13 +132,12 @@ public void testComparisonWithDatetimeUdf() {
   }
 
   @Test
-  public void testAllStandardDatetimeTypesCastToVarchar() {
+  public void testStandardDatetimeFieldsNotWrapped() {
     givenQuery("source = catalog.events | fields hire_date, start_time, created_at")
         .assertPlan(
             """
-            LogicalProject(hire_date=[CAST($0):VARCHAR NOT NULL], start_time=[CAST($1):VARCHAR NOT NULL], created_at=[CAST($2):VARCHAR NOT NULL])
-              LogicalProject(hire_date=[$2], start_time=[$3], created_at=[$4])
-                LogicalTableScan(table=[[catalog, events]])
+            LogicalProject(hire_date=[$2], start_time=[$3], created_at=[$4])
+              LogicalTableScan(table=[[catalog, events]])
             """);
   }
 
@@ -176,19 +173,16 @@ public void testSequentialPlanCallsDoNotCorruptShuttleStack() {
   }
 
   @Test
-  public void testOutputCastCanCompileAndExecute() throws Exception {
+  public void testDatetimeFieldsPreserveStandardTypes() throws Exception {
     RelNode plan =
         planner.plan("source = catalog.events | fields hire_date, start_time, created_at");
     try (PreparedStatement statement = compiler.compile(plan)) {
       ResultSet resultSet = statement.executeQuery();
       verify(resultSet)
           .expectSchema(
-              col("hire_date", java.sql.Types.VARCHAR),
-              col("start_time", java.sql.Types.VARCHAR),
-              col("created_at", java.sql.Types.VARCHAR))
-          .expectData(
-              row("2024-01-16", "12:00:00", "2024-01-15 08:00:00"),
-              row("2024-06-20", "14:00:00", "2024-06-20 00:00:00"));
+              col("hire_date", java.sql.Types.DATE),
+              col("start_time", java.sql.Types.TIME),
+              col("created_at", java.sql.Types.TIMESTAMP));
     }
   }
 }

api/src/test/java/org/opensearch/sql/api/spec/datetime/DatetimeExtensionTest.java

@@ -32,9 +32,8 @@
  * <p>The test methods here fire many queries through a thread pool to exercise concurrent {@code
  * plan()} invocations. New planner-level concurrency / state-isolation regressions belong in this
  * class. The current cases cover {@code DatetimeExtension}'s {@code RelHomogeneousShuttle}
- * subclasses ({@code DatetimeUdtNormalizeRule}, {@code DatetimeOutputCastRule}) which were
- * previously returned as static {@code INSTANCE}s and caused the production failure that motivated
- * this suite.
+ * subclass {@code DatetimeUdtNormalizeRule}, which was previously returned as a static {@code
+ * INSTANCE} and caused the production failure that motivated this suite.
  *
  * <p>Run via:
  *