fix(sandbox): cluster C — invalid date literals reject with typed format-hint message

- C1: pre-isthmus CastTemporalLiteralValidator surfaces malformed string-literal
  CAST(<lit> AS DATE/TIME/TIMESTAMP) and TIMESTAMPADD/TIMESTAMPDIFF string args
  as IllegalArgumentException with the typed format-hint wording
  ("<type>:<value> in unsupported format, please use '<pattern>'"), preventing
  the Arrow Parser error / StreamException opaque surface.
- C2: TimestampFunctionAdapter, DateAdapter, TimeAdapter, DateFormat/TimeFormat
  adapters, and DATE_PART unit-aware operands now reject malformed inputs at
  plan time with the typed format-hint message.
- C3: HOUR / DAY-of-month and str_to_date with invalid month or day reject
  with format-hint or out-of-range exception instead of silent success.
- DATETIME(<invalid>) folds to a NULL TIMESTAMP at plan time (matches legacy
  SQL DATETIME-of-invalid semantics).

The error surface is HTTP 400 with body
{"error": {"reason":"Invalid Query", "details":"<type>:<value> in unsupported
format, please use '<pattern>'", "type":"IllegalArgumentException"}}.

Author: vinaykpud

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

@@ -0,0 +1,93 @@
+/*
+ * 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.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+/** Pre-isthmus pass that rejects malformed CAST(string-lit AS DATE/TIME/TIMESTAMP) and TIMESTAMPADD/DIFF string args. */
+final class CastTemporalLiteralValidator {
+
+    private CastTemporalLiteralValidator() {}
+
+    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 ValidatingShuttle());
+            }
+        });
+    }
+
+    /** Test-only entry point; production goes through {@link #rewrite}. */
+    static RexShuttle newShuttle() {
+        return new ValidatingShuttle();
+    }
+
+    private static final class ValidatingShuttle extends RexShuttle {
+
+        @Override
+        public RexNode visitCall(RexCall call) {
+            RexCall recursed = (RexCall) super.visitCall(call);
+            if (isCastFromString(recursed)) {
+                DatetimeLiteralValidator.Kind kind = kindForTarget(recursed.getType().getSqlTypeName());
+                if (kind != null) {
+                    DatetimeLiteralValidator.validate(recursed.getOperands().get(0), kind);
+                }
+                return recursed;
+            }
+            validateTimestampAddDiffOperands(recursed);
+            return recursed;
+        }
+
+        /** Substrait can't bind TIMESTAMPADD/DIFF over string args; pre-validate so the user sees the format-hint, not "Unable to convert call". */
+        private static void validateTimestampAddDiffOperands(RexCall call) {
+            String name = call.getOperator().getName();
+            if (!"TIMESTAMPADD".equalsIgnoreCase(name) && !"TIMESTAMPDIFF".equalsIgnoreCase(name)) {
+                return;
+            }
+            int firstArg = "TIMESTAMPADD".equalsIgnoreCase(name) ? 2 : 1;
+            for (int i = firstArg; i < call.getOperands().size(); i++) {
+                RexNode operand = call.getOperands().get(i);
+                if (operand instanceof RexLiteral lit && SqlTypeName.CHAR_TYPES.contains(lit.getType().getSqlTypeName())) {
+                    DatetimeLiteralValidator.validate(operand, DatetimeLiteralValidator.Kind.TIMESTAMP);
+                }
+            }
+        }
+
+        private static boolean isCastFromString(RexCall call) {
+            if (call.getKind() != SqlKind.CAST && call.getKind() != SqlKind.SAFE_CAST) {
+                return false;
+            }
+            if (call.getOperands().size() != 1) {
+                return false;
+            }
+            if (!(call.getOperands().get(0) instanceof RexLiteral literal)) {
+                return false;
+            }
+            return SqlTypeName.CHAR_TYPES.contains(literal.getType().getSqlTypeName());
+        }
+
+        private static DatetimeLiteralValidator.Kind kindForTarget(SqlTypeName target) {
+            return switch (target) {
+                case DATE -> DatetimeLiteralValidator.Kind.DATE;
+                case TIME -> DatetimeLiteralValidator.Kind.TIME;
+                case TIMESTAMP, TIMESTAMP_WITH_LOCAL_TIME_ZONE -> DatetimeLiteralValidator.Kind.TIMESTAMP;
+                default -> null;
+            };
+        }
+    }
+}

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

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

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

@@ -28,6 +28,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
+import java.util.Set;
 
 /**
  * Date-part extractor adapters — rewrite {@code FN(ts)} to {@code date_part('<unit>', ts)}.
@@ -64,7 +65,7 @@ public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelO
         List<RexNode> coerced = new ArrayList<>(original.getOperands().size());
         for (RexNode operand : original.getOperands()) {
             if (isCharacterOperand(operand)) {
-                validateDatetimeLiteral(operand);
+                validateDatetimeLiteralForUnit(operand, unit);
                 coerced.add(castToTimestamp(operand, cluster));
             } else {
                 coerced.add(operand);
@@ -105,23 +106,25 @@ static RexNode coerceCharacterOperandToTimestamp(RexNode operand, RelOptCluster
         return castToTimestamp(operand, cluster);
     }
 
-    /**
-     * Eagerly parses string {@link RexLiteral} operands so an invalid literal surfaces as a
-     * coordinator-side {@link IllegalArgumentException} during planning, before the value reaches
-     * DataFusion's CAST kernel. The native error message ({@code "Arrow error: Parser error: ..."})
-     * is dropped by Flight RPC serialization on the worker→coordinator hop, so without this check
-     * users see {@code "Failed to start streaming fragment on ..."} instead of the legacy
-     * {@code "timestamp:<v> in unsupported format"} wording.
-     *
-     * <p>Non-literal operands (column refs, expressions) and NULL literals pass through — column
-     * value validation is a separate concern (Arrow CAST per-row error handling, tracked
-     * separately).
-     *
-     * <p>Accept-set mirrors legacy {@code DateTimeParser.parse}: try {@link LocalDateTime} (date+time
-     * with optional nanos), {@link LocalDate} (bare date), {@link LocalTime} (bare time), throw on
-     * all-failed. Same try/catch-fall-through shape used by
-     * {@link TimestampFunctionAdapter#parseTimestamp}.
-     */
+    /** Time-only units reject bare-date literals (HOUR('2020-08-26') must throw, not return 0). */
+    private static final Set<String> TIME_ONLY_UNITS = Set.of("hour", "minute", "second", "microsecond");
+
+    /** Date-part units reject bare-time literals (DAY('12:00:00') must throw, not silent-fail). */
+    private static final Set<String> DATE_ONLY_UNITS = Set.of("year", "quarter", "month", "day", "week", "doy", "dow");
+
+    static void validateDatetimeLiteralForUnit(RexNode operand, String unit) {
+        if (TIME_ONLY_UNITS.contains(unit)) {
+            DatetimeLiteralValidator.validate(operand, DatetimeLiteralValidator.Kind.TIME);
+            return;
+        }
+        if (DATE_ONLY_UNITS.contains(unit)) {
+            DatetimeLiteralValidator.validate(operand, DatetimeLiteralValidator.Kind.DATE);
+            return;
+        }
+        validateDatetimeLiteral(operand);
+    }
+
+    /** Plan-time validation for string-literal operands; accepts datetime / date / time, rejects garbage. */
     static void validateDatetimeLiteral(RexNode operand) {
         if (!(operand instanceof RexLiteral literal)) {
             return;
@@ -144,9 +147,7 @@ static void validateDatetimeLiteral(RexNode operand) {
             LocalTime.parse(value);
             return;
         } catch (DateTimeParseException ignored) {
-            // SQL plugin's ErrorMessageFactory.unwrapCause walks to the deepest cause for the response
-            // type/details, so attaching DateTimeParseException as cause would surface its stock JDK
-            // message instead of this one. Mirrors legacy DateTimeParser.parse, which throws causeless.
+            // causeless — see DatetimeLiteralValidator#fail.
             throw new IllegalArgumentException(
                 String.format(Locale.ROOT, "timestamp:%s in unsupported format, please use 'yyyy-MM-dd HH:mm:ss[.SSSSSSSSS]'", value)
             );

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

@@ -11,6 +11,7 @@
 import org.apache.calcite.plan.RelOptCluster;
 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.SqlFunction;
 import org.apache.calcite.sql.SqlFunctionCategory;
@@ -24,6 +25,8 @@
 import org.opensearch.analytics.spi.FieldStorageInfo;
 import org.opensearch.analytics.spi.ScalarFunctionAdapter;
 
+import java.time.LocalDateTime;
+import java.time.format.DateTimeParseException;
 import java.util.List;
 
 /**
@@ -138,6 +141,7 @@ public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelO
             List<RexNode> operands = original.getOperands();
             if (operands.size() == 1 && SqlTypeName.CHAR_TYPES.contains(operands.get(0).getType().getSqlTypeName())) {
                 RexNode arg = operands.get(0);
+                DatetimeLiteralValidator.validate(arg, DatetimeLiteralValidator.Kind.TIME);
                 RexNode pattern = rexBuilder.makeLiteral(DATE_PREFIX_PATTERN);
                 RexNode replacement = rexBuilder.makeLiteral("");
                 RexNode stripped = rexBuilder.makeCall(
@@ -151,15 +155,23 @@ public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelO
         }
     }
 
-    static final class DateAdapter extends AbstractNameMappingAdapter {
-        DateAdapter() {
-            super(LOCAL_DATE_OP, List.of(), List.of());
+    static final class DateAdapter implements ScalarFunctionAdapter {
+        @Override
+        public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelOptCluster cluster) {
+            if (original.getOperands().size() == 1) {
+                RexNode operand = original.getOperands().get(0);
+                if (SqlTypeName.CHAR_TYPES.contains(operand.getType().getSqlTypeName())) {
+                    DatetimeLiteralValidator.validate(operand, DatetimeLiteralValidator.Kind.DATE);
+                }
+            }
+            return cluster.getRexBuilder().makeCall(original.getType(), LOCAL_DATE_OP, original.getOperands());
         }
     }
 
     /**
      * Single-arg {@code DATETIME(string)} strips the trailing offset (+HH:MM / -HHMM / Z) so the
-     * result is wall-clock, not UTC-converted. Two-arg {@code DATETIME(string, tz)} and
+     * result is wall-clock, not UTC-converted. Invalid string literals fold to NULL TIMESTAMP
+     * (legacy SQL DATETIME-of-invalid semantics). Two-arg {@code DATETIME(string, tz)} and
      * non-string operands route through {@code to_timestamp} unchanged.
      *
      * <p>This belongs in the PPL frontend ({@code PPLBuiltinOperators.DATETIME}) so every
@@ -175,6 +187,10 @@ public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelO
             List<RexNode> operands = original.getOperands();
             if (operands.size() == 1 && SqlTypeName.CHAR_TYPES.contains(operands.get(0).getType().getSqlTypeName())) {
                 RexNode arg = operands.get(0);
+                RexNode foldedNull = tryFoldInvalidLiteralToNull(arg, original, cluster);
+                if (foldedNull != null) {
+                    return foldedNull;
+                }
                 RexNode pattern = rexBuilder.makeLiteral(OFFSET_SUFFIX_PATTERN);
                 RexNode replacement = rexBuilder.makeLiteral("");
                 RexNode stripped = rexBuilder.makeCall(
@@ -186,5 +202,23 @@ public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelO
             }
             return rexBuilder.makeCall(original.getType(), LOCAL_TO_TIMESTAMP_OP, operands);
         }
+
+        /** Folds any non-full-datetime string literal to NULL TIMESTAMP after stripping the offset suffix. */
+        private static RexNode tryFoldInvalidLiteralToNull(RexNode operand, RexCall original, RelOptCluster cluster) {
+            if (!(operand instanceof RexLiteral literal)) {
+                return null;
+            }
+            String value = literal.getValueAs(String.class);
+            if (value == null) {
+                return null;
+            }
+            String stripped = value.replaceAll(OFFSET_SUFFIX_PATTERN, "");
+            try {
+                LocalDateTime.parse(stripped.replace(' ', 'T'));
+                return null;
+            } catch (DateTimeParseException ignored) {
+                return cluster.getRexBuilder().makeNullLiteral(original.getType());
+            }
+        }
     }
 }

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

@@ -0,0 +1,94 @@
+/*
+ * 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.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.format.DateTimeParseException;
+import java.util.Locale;
+
+/** Plan-time validation for string-literal datetime operands; rejects malformed input with the format-hint message. */
+final class DatetimeLiteralValidator {
+
+    private DatetimeLiteralValidator() {}
+
+    enum Kind {
+        DATE("date", "yyyy-MM-dd"),
+        TIME("time", "HH:mm:ss[.SSSSSSSSS]"),
+        TIMESTAMP("timestamp", "yyyy-MM-dd HH:mm:ss[.SSSSSSSSS]");
+
+        final String typeName;
+        final String pattern;
+
+        Kind(String typeName, String pattern) {
+            this.typeName = typeName;
+            this.pattern = pattern;
+        }
+    }
+
+    static void validate(RexNode operand, Kind kind) {
+        if (!(operand instanceof RexLiteral literal)) {
+            return;
+        }
+        String value = literal.getValueAs(String.class);
+        if (value == null) {
+            return;
+        }
+        validate(value, kind);
+    }
+
+    static void validate(String value, Kind kind) {
+        boolean ok = switch (kind) {
+            case DATE -> isParseableAsDate(value) || isParseableAsFullDatetime(value);
+            case TIME -> isParseableAsTime(value) || isParseableAsFullDatetime(value);
+            case TIMESTAMP -> isParseableAsFullDatetime(value) || isParseableAsDate(value);
+        };
+        if (!ok) {
+            throw fail(value, kind);
+        }
+    }
+
+    private static boolean isParseableAsDate(String value) {
+        try {
+            LocalDate.parse(value);
+            return true;
+        } catch (DateTimeParseException ignored) {
+            return false;
+        }
+    }
+
+    private static boolean isParseableAsTime(String value) {
+        try {
+            LocalTime.parse(value);
+            return true;
+        } catch (DateTimeParseException ignored) {
+            return false;
+        }
+    }
+
+    private static boolean isParseableAsFullDatetime(String value) {
+        try {
+            LocalDateTime.parse(value.replace(' ', 'T'));
+            return true;
+        } catch (DateTimeParseException ignored) {
+            return false;
+        }
+    }
+
+    /** Causeless: ErrorMessageFactory.unwrapCause would surface the JDK DateTimeParseException's stock message otherwise. */
+    private static IllegalArgumentException fail(String value, Kind kind) {
+        return new IllegalArgumentException(
+            String.format(Locale.ROOT, "%s:%s in unsupported format, please use '%s'", kind.typeName, value, kind.pattern)
+        );
+    }
+}