[analytics-engine] Rewrite engine-output CAST(TIMESTAMP AS VARCHAR) to to_char

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java

@@ -142,6 +142,10 @@ public class DataFusionFragmentConvertor implements FragmentConvertor {
         FunctionMappings.s(SqlLibraryOperators.CONCAT_WS, "concat_ws"),
         FunctionMappings.s(SqlLibraryOperators.ILIKE, "ilike"),
         FunctionMappings.s(SqlLibraryOperators.DATE_PART, "date_part"),
+        // Engine-output cast rewrite target — see DatetimeOutputCastRewriter (issue #5420).
+        // Routes Calcite's TO_CHAR call to DataFusion's native `to_char` so PPL's
+        // documented space-separator timestamp output is preserved on the AE path.
+        FunctionMappings.s(SqlLibraryOperators.TO_CHAR, "to_char"),
         FunctionMappings.s(ConvertTzAdapter.LOCAL_CONVERT_TZ_OP, "convert_tz"),
         FunctionMappings.s(UnixTimestampAdapter.LOCAL_TO_UNIXTIME_OP, "to_unixtime"),
         // Niladic ops from DateTimeAdapters — each maps 1:1 to a DF builtin.
@@ -323,6 +327,10 @@ private byte[] convertToSubstrait(RelNode fragment) {
         // with "Unable to convert the type NULL". The widening only changes literal type
         // tags; semantics and field names (used by Plan.Root.names) are unchanged.
         RelNode preprocessed = UntypedNullPreprocessor.rewrite(fragment);
+        // Rewrite DatetimeOutputCastRule's CAST(<TIMESTAMP> AS VARCHAR) to to_char(...) so
+        // DataFusion emits PPL's space-separator timestamp format instead of Arrow's ISO-T.
+        // See issue #5420.
+        preprocessed = DatetimeOutputCastRewriter.rewrite(preprocessed);
         RelRoot root = RelRoot.of(preprocessed, SqlKind.SELECT);
         SubstraitRelVisitor visitor = createVisitor(preprocessed);
         Rel substraitRel;
@@ -364,6 +372,8 @@ private Rel convertStandalone(RelNode operator) {
         // wrapper conversion is just as susceptible to a SqlTypeName.NULL literal lurking in
         // a CASE call attached on top of an inner plan.
         RelNode preprocessed = UntypedNullPreprocessor.rewrite(operator);
+        // Same rationale as convertToSubstrait — issue #5420.
+        preprocessed = DatetimeOutputCastRewriter.rewrite(preprocessed);
         SubstraitRelVisitor visitor = createVisitor(preprocessed);
         return visitor.apply(preprocessed);
     }

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatetimeOutputCastRewriter.java

@@ -0,0 +1,183 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+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.logical.LogicalProject;
+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.sql.SqlKind;
+import org.apache.calcite.sql.fun.SqlLibraryOperators;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Pre-isthmus pass that rewrites the engine-output cast emitted by
+ * {@code DatetimeOutputCastRule} from {@code CAST(<TIMESTAMP> AS VARCHAR)} to
+ * {@code to_char(<TIMESTAMP>, '%Y-%m-%d %H:%M:%S')} so the DataFusion runtime
+ * emits PPL's documented space-separator format instead of Arrow's ISO-8601
+ * {@code T}-separator.
+ *
+ * <p>Issue: <a href="https://github.com/opensearch-project/sql/issues/5420">opensearch-project/sql#5420</a>.
+ *
+ * <p>Background: {@code DatetimeOutputCastRule} (api/spec/datetime) wraps every
+ * datetime field at the OUTERMOST {@link LogicalProject} in
+ * {@code CAST(... AS VARCHAR)} so the unified planner never has to know which
+ * backend serializes datetimes. Calcite's reference planner emits ANSI
+ * {@code "2024-01-15 12:00:00"}; DataFusion's Arrow CAST kernel emits
+ * {@code "2024-01-15T12:00:00"}. The session config
+ * {@code datafusion.format.timestamp_format} only affects the CLI display
+ * pipeline, not the Arrow cast kernel — verified by the issue's reporter.
+ *
+ * <p>Scope is intentionally narrow:
+ * <ul>
+ *   <li>Only the output {@link Project} is inspected. {@code DatetimeOutputCastRule}
+ *       wraps the input in exactly one final {@link Project}; the unified planner
+ *       may then wrap that Project in a single {@link Sort} (system query-size
+ *       limit). The rewriter therefore inspects either the root {@link Project}
+ *       or, when the root is a {@link Sort}, the {@link Project} sitting
+ *       directly beneath it. Any deeper {@link Project} (whether user-authored
+ *       or optimizer-generated) carries expressions that must round-trip
+ *       verbatim.</li>
+ *   <li>Only direct project slots — {@code project.getProjects().get(i)} — are
+ *       inspected. Nested casts inside {@code CASE}/{@code COALESCE}/UDF args
+ *       were authored by the user query and must round-trip verbatim.</li>
+ *   <li>Only {@code CAST(... AS VARCHAR)} matches the rule's output shape;
+ *       {@code CAST(... AS CHAR(n))} is user-authored and has different
+ *       length/padding semantics that {@code to_char} does not preserve.</li>
+ *   <li>Only {@link SqlTypeName#TIMESTAMP} sources are rewritten. PPL's
+ *       {@code DATE} (no clock) and {@code TIME} (no calendar) cast cleanly
+ *       through Arrow already, and {@link SqlTypeName#TIMESTAMP_WITH_LOCAL_TIME_ZONE}
+ *       depends on the DataFusion session timezone — emitting a literal space
+ *       format there could silently lie about the instant. Defer until a
+ *       concrete failing case lands.</li>
+ *   <li>The rewriter assumes {@code DatetimeUdtNormalizeRule} (also a
+ *       postAnalysisRule, ordered before {@code DatetimeOutputCastRule}) has
+ *       already normalized {@code ExprUDT.EXPR_TIMESTAMP} → standard
+ *       {@code SqlTypeName.TIMESTAMP}, so we only need to match the standard
+ *       SqlTypeName here.</li>
+ * </ul>
+ *
+ * <p>The Substrait emit path is wired in {@code DataFusionFragmentConvertor}:
+ * {@code SqlLibraryOperators.TO_CHAR} is mapped to the Substrait extension
+ * name {@code to_char} declared in {@code opensearch_scalar_functions.yaml},
+ * which DataFusion resolves to its native {@code to_char} scalar function.
+ *
+ * @opensearch.internal
+ */
+final class DatetimeOutputCastRewriter {
+
+    /**
+     * PPL's documented timestamp output format (space separator). Mirrors the
+     * format used by Calcite's reference planner so the analytics-engine path
+     * matches per-row output exactly.
+     */
+    static final String PPL_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S";
+
+    private DatetimeOutputCastRewriter() {}
+
+    /**
+     * Rewrite the engine-output {@code CAST(<TIMESTAMP> AS VARCHAR)} slots in the
+     * output {@link Project}. Returns {@code root} unchanged when the output
+     * Project cannot be located (e.g. raw scan / aggregate fragment) or when no
+     * slot matches.
+     *
+     * <p>The output Project is located in one of two shapes:
+     * <ol>
+     *   <li>{@code root} is itself a {@link Project} — the rule's output sits at
+     *       the root.</li>
+     *   <li>{@code root} is a {@link Sort} (the unified planner's
+     *       {@code LogicalSystemLimit} system query-size cap) and its input is a
+     *       {@link Project} — rewrite that Project's slots and rebuild the Sort
+     *       on top of the rewritten Project.</li>
+     * </ol>
+     *
+     * <p>Matches any {@link Project} subclass — {@code DatetimeOutputCastRule}
+     * emits a {@link LogicalProject}, but engine-side optimizer rules
+     * (e.g. {@code OpenSearchProjectRule}) may have already converted the
+     * matched Project to a custom {@link Project} subclass (e.g.
+     * {@code OpenSearchProject}). {@link Project#copy} on the matched subclass
+     * round-trips back to the same subclass, so any subclass-specific state
+     * (viable backends, traits) is preserved.
+     *
+     * <p>The traversal is intentionally NOT recursive: only the output Project
+     * is rewritten — any deeper {@link Project} carries expressions that must
+     * round-trip verbatim.
+     */
+    static RelNode rewrite(RelNode root) {
+        if (root instanceof Project project) {
+            Project rewritten = rewriteOutputProject(project);
+            return rewritten == project ? root : rewritten;
+        }
+        if (root instanceof Sort sort && sort.getInput() instanceof Project project) {
+            Project rewritten = rewriteOutputProject(project);
+            if (rewritten == project) {
+                return root;
+            }
+            return sort.copy(sort.getTraitSet(), rewritten, sort.getCollation(), sort.offset, sort.fetch);
+        }
+        return root;
+    }
+
+    /**
+     * Returns a new {@link Project} (same subclass as {@code project}) with
+     * engine-output cast slots rewritten, or returns {@code project} unchanged
+     * when no slot matched.
+     */
+    private static Project rewriteOutputProject(Project project) {
+        List<RexNode> oldProjects = project.getProjects();
+        List<RexNode> newProjects = new ArrayList<>(oldProjects.size());
+        boolean changed = false;
+        RexBuilder rexBuilder = project.getCluster().getRexBuilder();
+        for (RexNode expr : oldProjects) {
+            RexNode rewritten = rewriteDirectOutputCast(expr, rexBuilder);
+            if (rewritten != expr) {
+                changed = true;
+            }
+            newProjects.add(rewritten);
+        }
+        if (!changed) {
+            return project;
+        }
+        return project.copy(project.getTraitSet(), project.getInput(), newProjects, project.getRowType());
+    }
+
+    /**
+     * Returns a {@code to_char(<expr>, format)} call when {@code expr} is the
+     * exact shape {@code CAST(<TIMESTAMP> AS VARCHAR)} produced by
+     * {@code DatetimeOutputCastRule}; otherwise returns {@code expr} unchanged.
+     *
+     * <p>Note: deliberately not recursive — see class-level scope notes.
+     */
+    private static RexNode rewriteDirectOutputCast(RexNode expr, RexBuilder rexBuilder) {
+        if (!(expr instanceof RexCall call) || call.getKind() != SqlKind.CAST) {
+            return expr;
+        }
+        RexNode source = call.getOperands().get(0);
+        SqlTypeName sourceType = source.getType().getSqlTypeName();
+        SqlTypeName targetType = call.getType().getSqlTypeName();
+        if (sourceType != SqlTypeName.TIMESTAMP) {
+            return expr;
+        }
+        // VARCHAR-only: DatetimeOutputCastRule emits CAST(... AS VARCHAR) (length-unspecified).
+        // CHAR(n) is user-authored and has length/padding semantics that to_char does not preserve.
+        if (targetType != SqlTypeName.VARCHAR) {
+            return expr;
+        }
+        RelDataType formatType = rexBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR);
+        RexNode formatLiteral = rexBuilder.makeLiteral(PPL_TIMESTAMP_FORMAT, formatType, true);
+        return rexBuilder.makeCall(call.getType(), SqlLibraryOperators.TO_CHAR, List.of(source, formatLiteral));
+    }
+}

sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml

@@ -525,6 +525,21 @@ scalar_functions:
         nullability: DECLARED_OUTPUT
         return: string
 
+  # to_char(value, format) — render a timestamp using a strftime-style format string.
+  # Bound here so that DatetimeOutputCastRewriter can rewrite the engine-output cast
+  # `CAST(<TIMESTAMP> AS VARCHAR)` to `to_char(<TIMESTAMP>, '%Y-%m-%d %H:%M:%S')` and
+  # have the call serialize through this extension instead of Calcite's stdlib URN.
+  # DataFusion exposes a native `to_char` scalar that consumes strftime tokens. See
+  # issue #5420 (engine-output format divergence between Calcite and DataFusion).
+  - name: "to_char"
+    description: "Render a timestamp using a strftime-style format string (engine-output cast target)."
+    impls:
+      - args:
+          - { name: value, value: "precision_timestamp<P>" }
+          - { name: format, value: string }
+        nullability: DECLARED_OUTPUT
+        return: string
+
   # tonumber(string, base) — parse `string` as a base-N integer
   - name: "tonumber"
     description: "Parse a string to a number in the given radix (2-36). Returns NULL on parse failure."

sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/DataFusionFragmentConvertorTests.java

@@ -75,7 +75,10 @@ public void setUp() throws Exception {
             t.setContextClassLoader(DataFusionFragmentConvertorTests.class.getClassLoader());
             SimpleExtension.ExtensionCollection delegationExtensions = SimpleExtension.load(List.of("/delegation_functions.yaml"));
             SimpleExtension.ExtensionCollection aggregateExtensions = SimpleExtension.load(List.of("/opensearch_aggregate_functions.yaml"));
-            extensions = DefaultExtensionCatalog.DEFAULT_COLLECTION.merge(delegationExtensions).merge(aggregateExtensions);
+            SimpleExtension.ExtensionCollection scalarExtensions = SimpleExtension.load(List.of("/opensearch_scalar_functions.yaml"));
+            extensions = DefaultExtensionCatalog.DEFAULT_COLLECTION.merge(delegationExtensions)
+                .merge(aggregateExtensions)
+                .merge(scalarExtensions);
         } finally {
             t.setContextClassLoader(prev);
         }
@@ -576,6 +579,47 @@ public void testApproxCountDistinctRenamed() throws Exception {
      * SUM aggregate is not affected by the rename map — its extension function
      * name remains unchanged.
      */
+    /**
+     * End-to-end: a {@code Project[CAST(ts AS VARCHAR)]} fragment must serialize
+     * with a {@code to_char} extension function, not a raw Substrait {@code cast},
+     * proving {@link DatetimeOutputCastRewriter} fires inside the convertor and
+     * the {@code to_char} declaration in {@code opensearch_scalar_functions.yaml}
+     * is reachable through {@code FunctionMappings}. See issue
+     * <a href="https://github.com/opensearch-project/sql/issues/5420">sql#5420</a>.
+     */
+    public void testProjectTimestampOutputCastEmitsToCharExtension() throws Exception {
+        RelDataTypeFactory.Builder b = typeFactory.builder();
+        b.add("ts", typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.TIMESTAMP, 6), true));
+        RelNode scan = new DataFusionFragmentConvertor.StageInputTableScan(cluster, cluster.traitSet(), "test_index", b.build());
+
+        RelDataType varcharType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.VARCHAR), true);
+        RexNode tsField = rexBuilder.makeInputRef(scan, 0);
+        RexNode castExpr = rexBuilder.makeCast(varcharType, tsField);
+        RelNode project = org.apache.calcite.rel.logical.LogicalProject.create(
+            scan,
+            List.of(),
+            List.of(castExpr),
+            List.of("ts_str"),
+            java.util.Set.of()
+        );
+
+        byte[] bytes = newConvertor().convertShardScanFragment("test_index", project);
+        Plan plan = decodeSubstrait(bytes);
+
+        boolean foundToChar = false;
+        for (SimpleExtensionDeclaration decl : plan.getExtensionsList()) {
+            if (decl.hasExtensionFunction()) {
+                String name = decl.getExtensionFunction().getName();
+                String baseName = name.contains(":") ? name.substring(0, name.indexOf(':')) : name;
+                if (baseName.equals("to_char")) {
+                    foundToChar = true;
+                    break;
+                }
+            }
+        }
+        assertTrue("CAST(<TIMESTAMP> AS VARCHAR) must serialize as the to_char extension, not a raw cast", foundToChar);
+    }
+
     public void testOtherFunctionsNotRenamed() throws Exception {
         RelNode scan = buildTableScan("test_index", "A");
         LogicalAggregate agg = buildSumAggregate(scan, 0);