datetime stringify uses space, time has no epoch prefix

- ArrowValues: scalar + list temporal cells use space separator; time renders
  as HH:mm:ss[.frac] (no 1970-01-01 prefix); nanos preserved.
- ArrowValues: post-process ISO-T VarChar cells from native CAST -> space.
- CastToVarcharRewriter (new): boolean -> varchar emits TRUE/FALSE.
- ListAggregateMultiTypeIT: assertions updated for new format.

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,84 @@
+/*
+ * 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;
+
+/** Boolean → varchar emits upper-case TRUE/FALSE (PPL contract matches {@code tostring(boolean)}). */
+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. */
+    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));
+        }
+
+        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;
+        }
+
+        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);
+        }
+    }
+}

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/ArrowValues.java

@@ -13,6 +13,7 @@
 import org.apache.arrow.vector.complex.ListVector;
 import org.apache.arrow.vector.complex.MapVector;
 import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.util.Text;
 
 import java.nio.charset.StandardCharsets;
@@ -21,20 +22,31 @@
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
+import java.util.regex.Pattern;
 
 /** Reads Arrow vector cells as plain Java values, unwrapping Arrow {@link Text} recursively. */
 public final class ArrowValues {
 
+    // Space-separator output matches the SQL plugin's ExprTimestampValue.
+    private static final DateTimeFormatter TIMESTAMP_NO_NANO = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss", Locale.ROOT);
+    private static final DateTimeFormatter TIMESTAMP_WITH_NANO = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSSSSS", Locale.ROOT);
+    private static final DateTimeFormatter TIME_NO_NANO = DateTimeFormatter.ofPattern("HH:mm:ss", Locale.ROOT);
+    private static final DateTimeFormatter TIME_WITH_NANO = DateTimeFormatter.ofPattern("HH:mm:ss.SSSSSSSSS", Locale.ROOT);
+    // DataFusion CAST(temporal AS VARCHAR) — date and time joined by 'T', optional fraction.
+    private static final Pattern ISO_TIMESTAMP_T = Pattern.compile("^(\\d{4}-\\d{2}-\\d{2})T(\\d{2}:\\d{2}:\\d{2}(?:\\.\\d+)?)$");
+
     private ArrowValues() {}
 
     public static Object toJavaValue(FieldVector vector, int index) {
         if (vector.isNull(index)) return null;
         if (vector instanceof VarCharVector v) {
-            return new String(v.get(index), StandardCharsets.UTF_8);
+            return spaceSeparator(new String(v.get(index), StandardCharsets.UTF_8));
         }
         // MapVector extends ListVector — must come first.
         if (vector instanceof MapVector && vector.getObject(index) instanceof List<?> entries) {
@@ -48,74 +60,103 @@ public static Object toJavaValue(FieldVector vector, int index) {
             return map;
         }
         Object value = vector.getObject(index);
-        if (vector instanceof ListVector && value instanceof List<?> raw) {
-            return normalizeList(raw);
+        if (vector instanceof ListVector lv && value instanceof List<?> raw) {
+            // child Arrow type drives temporal element formatting
+            return normalizeList(raw, lv.getDataVector().getField());
         }
-        // Date/Time/Timestamp cells return java.time.* so ExprValueUtils wraps them as
-        // ExprDateValue / ExprTimeValue / ExprTimestampValue and formats them.
-        Object temporal = normalizeTemporal(vector.getField().getType(), value);
+        Object temporal = formatTemporal(vector.getField().getType(), value);
         if (temporal != null) {
             return temporal;
         }
         return normalize(value);
     }
 
-    /** Returns LocalDate / LocalTime / LocalDateTime, or null for non-temporal types. */
-    private static Object normalizeTemporal(ArrowType type, Object value) {
+    /** ISO-T temporal → space separator; other strings unchanged. */
+    private static String spaceSeparator(String s) {
+        if (s == null) return null;
+        var m = ISO_TIMESTAMP_T.matcher(s);
+        return m.matches() ? m.group(1) + " " + m.group(2) : s;
+    }
+
+    private static Object formatTemporal(ArrowType type, Object value) {
         if (value == null) return null;
         if (type instanceof ArrowType.Date date) {
-            return toLocalDate(date, value);
+            return formatDate(date, value);
         }
         if (type instanceof ArrowType.Time time) {
-            return toLocalTime(time, value);
+            return formatTime(time, value);
         }
         if (type instanceof ArrowType.Timestamp ts) {
-            return toLocalDateTime(ts, value);
+            return formatTimestamp(ts, value);
         }
         return null;
     }
 
-    private static LocalDate toLocalDate(ArrowType.Date type, Object value) {
-        if (value instanceof LocalDate ld) return ld;
-        if (value instanceof LocalDateTime ldt) return ldt.toLocalDate();
-        long raw = ((Number) value).longValue();
-        return switch (type.getUnit()) {
-            case DAY -> LocalDate.ofEpochDay(raw);
-            case MILLISECOND -> LocalDate.ofEpochDay(Math.floorDiv(raw, 86_400_000L));
-        };
+    private static String formatDate(ArrowType.Date type, Object value) {
+        LocalDate ld;
+        if (value instanceof LocalDate d) {
+            ld = d;
+        } else if (value instanceof LocalDateTime ldt) {
+            ld = ldt.toLocalDate();
+        } else {
+            long raw = ((Number) value).longValue();
+            ld = switch (type.getUnit()) {
+                case DAY -> LocalDate.ofEpochDay(raw);
+                case MILLISECOND -> LocalDate.ofEpochDay(Math.floorDiv(raw, 86_400_000L));
+            };
+        }
+        return ld.format(DateTimeFormatter.ISO_LOCAL_DATE);
     }
 
-    private static LocalTime toLocalTime(ArrowType.Time type, Object value) {
-        if (value instanceof LocalTime lt) return lt;
-        if (value instanceof LocalDateTime ldt) return ldt.toLocalTime();
-        long raw = ((Number) value).longValue();
-        long nanoOfDay = switch (type.getUnit()) {
-            case SECOND -> raw * 1_000_000_000L;
-            case MILLISECOND -> raw * 1_000_000L;
-            case MICROSECOND -> raw * 1_000L;
-            case NANOSECOND -> raw;
-        };
-        return LocalTime.ofNanoOfDay(nanoOfDay);
+    /** Time -> HH:mm:ss[.frac]; never prefixes with the 1970 epoch date. */
+    private static String formatTime(ArrowType.Time type, Object value) {
+        LocalTime lt;
+        if (value instanceof LocalTime t) {
+            lt = t;
+        } else if (value instanceof LocalDateTime ldt) {
+            lt = ldt.toLocalTime();
+        } else {
+            long raw = ((Number) value).longValue();
+            long nanoOfDay = switch (type.getUnit()) {
+                case SECOND -> raw * 1_000_000_000L;
+                case MILLISECOND -> raw * 1_000_000L;
+                case MICROSECOND -> raw * 1_000L;
+                case NANOSECOND -> raw;
+            };
+            lt = LocalTime.ofNanoOfDay(nanoOfDay);
+        }
+        return lt.getNano() == 0 ? lt.format(TIME_NO_NANO) : lt.format(TIME_WITH_NANO);
     }
 
-    private static LocalDateTime toLocalDateTime(ArrowType.Timestamp type, Object value) {
-        if (value instanceof LocalDateTime ldt) return ldt;
-        long raw = ((Number) value).longValue();
-        Instant instant = switch (type.getUnit()) {
-            case SECOND -> Instant.ofEpochSecond(raw);
-            case MILLISECOND -> Instant.ofEpochMilli(raw);
-            case MICROSECOND -> Instant.ofEpochSecond(Math.floorDiv(raw, 1_000_000L), Math.floorMod(raw, 1_000_000L) * 1_000L);
-            case NANOSECOND -> Instant.ofEpochSecond(Math.floorDiv(raw, 1_000_000_000L), Math.floorMod(raw, 1_000_000_000L));
-        };
-        return LocalDateTime.ofInstant(instant, ZoneOffset.UTC);
+    /** Timestamp -> yyyy-MM-dd HH:mm:ss[.frac]. */
+    private static String formatTimestamp(ArrowType.Timestamp type, Object value) {
+        LocalDateTime ldt;
+        if (value instanceof LocalDateTime t) {
+            ldt = t;
+        } else if (value instanceof LocalDate ld) {
+            ldt = ld.atStartOfDay();
+        } else {
+            long raw = ((Number) value).longValue();
+            Instant instant = switch (type.getUnit()) {
+                case SECOND -> Instant.ofEpochSecond(raw);
+                case MILLISECOND -> Instant.ofEpochMilli(raw);
+                case MICROSECOND -> Instant.ofEpochSecond(Math.floorDiv(raw, 1_000_000L), Math.floorMod(raw, 1_000_000L) * 1_000L);
+                case NANOSECOND -> Instant.ofEpochSecond(Math.floorDiv(raw, 1_000_000_000L), Math.floorMod(raw, 1_000_000_000L));
+            };
+            ldt = LocalDateTime.ofInstant(instant, ZoneOffset.UTC);
+        }
+        return ldt.getNano() == 0 ? ldt.format(TIMESTAMP_NO_NANO) : ldt.format(TIMESTAMP_WITH_NANO);
     }
 
     private static Object normalize(Object value) {
         if (value instanceof Text t) {
-            return t.toString();
+            return spaceSeparator(t.toString());
+        }
+        if (value instanceof String s) {
+            return spaceSeparator(s);
         }
         if (value instanceof List<?> list) {
-            return normalizeList(list);
+            return normalizeList(list, null);
         }
         if (value instanceof Map<?, ?> m) {
             LinkedHashMap<String, Object> out = new LinkedHashMap<>(m.size());
@@ -128,10 +169,12 @@ private static Object normalize(Object value) {
         return value;
     }
 
-    private static List<Object> normalizeList(List<?> raw) {
+    private static List<Object> normalizeList(List<?> raw, Field childField) {
+        ArrowType childType = childField == null ? null : childField.getType();
         List<Object> out = new ArrayList<>(raw.size());
         for (Object element : raw) {
-            out.add(normalize(element));
+            Object formatted = childType == null ? null : formatTemporal(childType, element);
+            out.add(formatted != null ? formatted : normalize(element));
         }
         return out;
     }