fix(sandbox): cluster D — datetime stringify uses space, time has no epoch prefix

- ArrowValues: format Date/Time/Timestamp scalars and list-of-temporal elements
  with space separator; emit Time as HH:mm:ss[.fraction] without 1970-01-01
  prefix; preserve nanosecond precision when present.
- ArrowValues: post-process VarChar cells matching ISO timestamp pattern to
  swap T → ' ' so DataFusion's native CAST(timestamp AS VARCHAR) results
  (chart column-split, list_merge intermediates) render with a space.
- CastToVarcharRewriter (new): rewrite CAST(boolean AS VARCHAR) to CASE WHEN v
  THEN 'TRUE' WHEN NOT v THEN 'FALSE' END so the engine emits upper-case
  literals matching tostring(boolean).
- ListAggregateMultiTypeIT: update boolean / date / date_nanos assertions to
  match the new space-separator and upper-case TRUE output.

Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>

Author: vinaykpud

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

@@ -0,0 +1,96 @@
+/*
+ * 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.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.SqlKind;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+import java.util.List;
+
+/**
+ * Pre-isthmus pass that rewrites {@code CAST(boolean AS VARCHAR)} to
+ * {@code CASE WHEN val THEN 'TRUE' WHEN NOT val THEN 'FALSE' END} so the engine emits the
+ * upper-case literals PPL clients expect (matching {@code tostring(boolean)}).
+ *
+ * <p>Temporal casts to VARCHAR are intentionally NOT rewritten here — that would force a fixed
+ * format pattern (lossy for nanosecond timestamps). The space-separator + nano-preservation fix
+ * for temporals lives in {@code ArrowValues}, which post-processes the rendered string.
+ *
+ * @opensearch.internal
+ */
+final class CastToVarcharRewriter {
+
+    private CastToVarcharRewriter() {}
+
+    static RelNode rewrite(RelNode root) {
+        return root.accept(new RelHomogeneousShuttle() {
+            @Override
+            public RelNode visit(RelNode other) {
+                RelNode visited = super.visit(other);
+                return visited.accept(new CastShuttle(visited.getCluster().getRexBuilder()));
+            }
+        });
+    }
+
+    /** Package-visible for unit tests; production code goes through {@link #rewrite}. */
+    static RexShuttle newShuttle(RexBuilder rexBuilder) {
+        return new CastShuttle(rexBuilder);
+    }
+
+    private static final class CastShuttle extends RexShuttle {
+
+        private final RexBuilder rexBuilder;
+
+        CastShuttle(RexBuilder rexBuilder) {
+            this.rexBuilder = rexBuilder;
+        }
+
+        @Override
+        public RexNode visitCall(RexCall call) {
+            RexCall recursed = (RexCall) super.visitCall(call);
+            if (!isCastBooleanToVarchar(recursed)) {
+                return recursed;
+            }
+            return rewriteBooleanCast(recursed, recursed.getOperands().get(0));
+        }
+
+        /** True iff the call is CAST(boolean AS VARCHAR/CHAR). */
+        private static boolean isCastBooleanToVarchar(RexCall call) {
+            if (call.getKind() != SqlKind.CAST && call.getKind() != SqlKind.SAFE_CAST) {
+                return false;
+            }
+            if (call.getOperands().size() != 1) {
+                return false;
+            }
+            SqlTypeName target = call.getType().getSqlTypeName();
+            if (target != SqlTypeName.VARCHAR && target != SqlTypeName.CHAR) {
+                return false;
+            }
+            return call.getOperands().get(0).getType().getSqlTypeName() == SqlTypeName.BOOLEAN;
+        }
+
+        /** boolean → CASE WHEN v THEN 'TRUE' WHEN NOT v THEN 'FALSE' END (matches tostring(boolean)). */
+        private RexNode rewriteBooleanCast(RexCall original, RexNode value) {
+            RelDataType varcharType = original.getType();
+            RexNode trueLit = rexBuilder.makeLiteral("TRUE");
+            RexNode falseLit = rexBuilder.makeLiteral("FALSE");
+            RexNode notValue = rexBuilder.makeCall(SqlStdOperatorTable.NOT, value);
+            RexNode nullLit = rexBuilder.makeNullLiteral(varcharType);
+            return rexBuilder.makeCall(varcharType, SqlStdOperatorTable.CASE, List.of(value, trueLit, notValue, falseLit, nullLit));
+        }
+    }
+}

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

@@ -458,6 +458,7 @@ private byte[] convertToSubstrait(RelNode fragment) {
         preprocessed = PplAggregateCallRewriter.rewrite(preprocessed);
         preprocessed = PplWindowCallRewriter.rewrite(preprocessed);
         preprocessed = ItemTypeRebuilder.rewrite(preprocessed);
+        preprocessed = CastToVarcharRewriter.rewrite(preprocessed);
         RelRoot root = RelRoot.of(preprocessed, SqlKind.SELECT);
         SubstraitRelVisitor visitor = createVisitor(preprocessed);
         Rel substraitRel;
@@ -488,6 +489,7 @@ private Rel convertStandalone(RelNode operator) {
         preprocessed = PplAggregateCallRewriter.rewrite(preprocessed);
         preprocessed = PplWindowCallRewriter.rewrite(preprocessed);
         preprocessed = ItemTypeRebuilder.rewrite(preprocessed);
+        preprocessed = CastToVarcharRewriter.rewrite(preprocessed);
         SubstraitRelVisitor visitor = createVisitor(preprocessed);
         return visitor.apply(preprocessed);
     }

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

@@ -0,0 +1,96 @@
+/*
+ * 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.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.opensearch.test.OpenSearchTestCase;
+
+public class CastToVarcharRewriterTests extends OpenSearchTestCase {
+
+    /** CAST(boolean AS VARCHAR) → CASE WHEN val THEN 'TRUE' WHEN NOT val THEN 'FALSE' END. */
+    public void testBooleanCastRewritesToCaseUpper() {
+        Cluster c = newCluster();
+        RexNode operand = c.inputRef(0, SqlTypeName.BOOLEAN);
+        RexNode cast = c.castTo(operand, SqlTypeName.VARCHAR);
+
+        RexCall out = (RexCall) cast.accept(CastToVarcharRewriter.newShuttle(c.rexBuilder));
+
+        assertEquals(SqlKind.CASE, out.getKind());
+        assertEquals("TRUE", ((RexLiteral) out.getOperands().get(1)).getValueAs(String.class));
+        assertEquals("FALSE", ((RexLiteral) out.getOperands().get(3)).getValueAs(String.class));
+    }
+
+    /** Temporal casts pass through — handled by ArrowValues post-processing instead. */
+    public void testTimestampCastPassesThrough() {
+        Cluster c = newCluster();
+        RexNode operand = c.inputRef(0, SqlTypeName.TIMESTAMP);
+        RexNode cast = c.castTo(operand, SqlTypeName.VARCHAR);
+
+        RexNode out = cast.accept(CastToVarcharRewriter.newShuttle(c.rexBuilder));
+
+        assertEquals(SqlKind.CAST, out.getKind());
+    }
+
+    /** Unrelated casts (INT → VARCHAR) pass through. */
+    public void testIntegerCastPassesThrough() {
+        Cluster c = newCluster();
+        RexNode operand = c.inputRef(0, SqlTypeName.INTEGER);
+        RexNode cast = c.castTo(operand, SqlTypeName.VARCHAR);
+
+        RexNode out = cast.accept(CastToVarcharRewriter.newShuttle(c.rexBuilder));
+
+        assertEquals(SqlKind.CAST, out.getKind());
+    }
+
+    /** Non-cast calls pass through. */
+    public void testNonCastCallPassesThrough() {
+        Cluster c = newCluster();
+        RexNode left = c.inputRef(0, SqlTypeName.INTEGER);
+        RexNode right = c.inputRef(1, SqlTypeName.INTEGER);
+        RexNode plus = c.rexBuilder.makeCall(SqlStdOperatorTable.PLUS, left, right);
+
+        RexNode out = plus.accept(CastToVarcharRewriter.newShuttle(c.rexBuilder));
+
+        assertSame(plus, out);
+    }
+
+    private static Cluster newCluster() {
+        RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl();
+        RexBuilder rexBuilder = new RexBuilder(typeFactory);
+        return new Cluster(typeFactory, rexBuilder);
+    }
+
+    private static final class Cluster {
+        final RelDataTypeFactory typeFactory;
+        final RexBuilder rexBuilder;
+
+        Cluster(RelDataTypeFactory typeFactory, RexBuilder rexBuilder) {
+            this.typeFactory = typeFactory;
+            this.rexBuilder = rexBuilder;
+        }
+
+        RexNode inputRef(int index, SqlTypeName type) {
+            return rexBuilder.makeInputRef(typeFactory.createTypeWithNullability(typeFactory.createSqlType(type), true), index);
+        }
+
+        RexNode castTo(RexNode operand, SqlTypeName target) {
+            RelDataType targetType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(target), true);
+            return rexBuilder.makeCast(targetType, operand);
+        }
+    }
+}