Fix TIMESTAMP/DATE arithmetic and DATE→TIMESTAMP cast

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

@@ -676,6 +676,7 @@ public Map<ScalarFunction, ScalarFunctionAdapter> scalarFunctionAdapters() {
                     Map.entry(ScalarFunction.MINSPAN_BUCKET, new MinspanBucketAdapter()),
                     Map.entry(ScalarFunction.MINUTE, minute),
                     Map.entry(ScalarFunction.MINUTE_OF_HOUR, minute),
+                    Map.entry(ScalarFunction.MINUS, new MinusAdapter()),
                     Map.entry(ScalarFunction.MOD, new StdOperatorRewriteAdapter("MOD", SqlStdOperatorTable.MOD)),
                     Map.entry(ScalarFunction.MONTH, month),
                     Map.entry(ScalarFunction.MONTH_OF_YEAR, month),

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

@@ -0,0 +1,67 @@
+/*
+ * 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.plan.RelOptCluster;
+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.RexOver;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.opensearch.analytics.spi.FieldStorageInfo;
+import org.opensearch.analytics.spi.ScalarFunctionAdapter;
+
+import java.util.List;
+
+/**
+ * Rewrites PPL {@code t1 - t2} on TIMESTAMP / DATE operands as
+ * {@code to_unixtime(t1) - to_unixtime(t2)} (seconds), since Substrait's default
+ * catalog has no {@code subtract(precision_timestamp, ...)} binding. Numeric operands
+ * pass through unchanged.
+ *
+ * @opensearch.internal
+ */
+class MinusAdapter implements ScalarFunctionAdapter {
+
+    @Override
+    public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelOptCluster cluster) {
+        if (original.getOperator() != SqlStdOperatorTable.MINUS || original.getOperands().size() != 2) {
+            return original;
+        }
+        RexNode left = original.getOperands().get(0);
+        RexNode right = original.getOperands().get(1);
+        if (!isDateOrTimestamp(left.getType()) || !isDateOrTimestamp(right.getType())) {
+            return original;
+        }
+        // Leave MINUS(MAX OVER (), MIN OVER ()) for WidthBucketAdapter — it pattern-matches
+        // that exact shape to lower `bin <ts> bins=N` into integer-seconds math.
+        if (left instanceof RexOver && right instanceof RexOver) {
+            return original;
+        }
+
+        RexBuilder rexBuilder = cluster.getRexBuilder();
+        RexNode leftSeconds = rexBuilder.makeCall(UnixTimestampAdapter.LOCAL_TO_UNIXTIME_OP, left);
+        RexNode rightSeconds = rexBuilder.makeCall(UnixTimestampAdapter.LOCAL_TO_UNIXTIME_OP, right);
+        RexNode diffSeconds = rexBuilder.makeCall(SqlStdOperatorTable.MINUS, leftSeconds, rightSeconds);
+
+        // PPL's MINUS(DATETIME, DATETIME) infers TIMESTAMP at the call site; lift the
+        // BIGINT seconds back through from_unixtime so Project.isValid type-matches.
+        RelDataType fp64 = rexBuilder.getTypeFactory().createSqlType(SqlTypeName.DOUBLE);
+        RexNode diffDouble = rexBuilder.makeCast(fp64, diffSeconds, true);
+        RexNode asTimestamp = rexBuilder.makeCall(RustUdfDateTimeAdapters.LOCAL_FROM_UNIXTIME_OP, diffDouble);
+        return rexBuilder.makeCast(original.getType(), asTimestamp, true);
+    }
+
+    private static boolean isDateOrTimestamp(RelDataType type) {
+        SqlTypeName name = type.getSqlTypeName();
+        return name == SqlTypeName.TIMESTAMP || name == SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE || name == SqlTypeName.DATE;
+    }
+}

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

@@ -95,9 +95,31 @@ public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelO
         if (folded != null) {
             return wrapWithCallType(folded, original, cluster);
         }
+        RexNode dateLifted = tryLiftDateOperand(original, cluster);
+        if (dateLifted != null) {
+            return dateLifted;
+        }
         return wrapWithCallType(DATETIME_ADAPTER.adapt(original, fieldStorage, cluster), original, cluster);
     }
 
+    /**
+     * 1-arg {@code TIMESTAMP(<date>)} on a DATE column: emit a native CAST instead of
+     * letting it lower to {@code to_timestamp(date_col)}, which DataFusion's
+     * {@code to_timestamp} UDF rejects (Date32 is not in its accepted-type list).
+     * The CAST maps to arrow's Date32 → Timestamp(Nanosecond) kernel — midnight UTC,
+     * matching Shape B's literal fold. Returns {@code null} for non-DATE operands.
+     */
+    private static RexNode tryLiftDateOperand(RexCall original, RelOptCluster cluster) {
+        if (original.getOperands().size() != 1) {
+            return null;
+        }
+        RexNode operand = stripOperatorAnnotation(original.getOperands().get(0));
+        if (operand.getType().getSqlTypeName() != SqlTypeName.DATE) {
+            return null;
+        }
+        return cluster.getRexBuilder().makeAbstractCast(original.getType(), operand);
+    }
+
     /**
      * Recognize the four nested-with-literal shapes and fold them at plan time
      * into typed TIMESTAMP literals, matching legacy semantics from the SQL

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

@@ -0,0 +1,130 @@
+/*
+ * 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 com.google.common.collect.ImmutableList;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.hep.HepPlanner;
+import org.apache.calcite.plan.hep.HepProgramBuilder;
+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.RexNode;
+import org.apache.calcite.rex.RexWindowBounds;
+import org.apache.calcite.rex.RexWindowExclusion;
+import org.apache.calcite.sql.SqlAggFunction;
+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;
+
+import java.util.List;
+
+public class MinusAdapterTests extends OpenSearchTestCase {
+
+    private RelDataTypeFactory typeFactory;
+    private RexBuilder rexBuilder;
+    private RelOptCluster cluster;
+    private MinusAdapter adapter;
+
+    @Override
+    public void setUp() throws Exception {
+        super.setUp();
+        typeFactory = new JavaTypeFactoryImpl();
+        rexBuilder = new RexBuilder(typeFactory);
+        cluster = RelOptCluster.create(new HepPlanner(new HepProgramBuilder().build()), rexBuilder);
+        adapter = new MinusAdapter();
+    }
+
+    public void testTimestampMinusTimestampRewritesToUnixSecondsDiff() {
+        RexCall call = minusOf(SqlTypeName.TIMESTAMP, SqlTypeName.TIMESTAMP);
+        RexNode adapted = adapter.adapt(call, List.of(), cluster);
+
+        // Expect: CAST(from_unixtime(CAST(MINUS(to_unixtime(t1), to_unixtime(t2)) AS DOUBLE)) AS TIMESTAMP).
+        RexCall fromUnixtime = (RexCall) unwrapCast(adapted);
+        assertSame(RustUdfDateTimeAdapters.LOCAL_FROM_UNIXTIME_OP, fromUnixtime.getOperator());
+
+        RexCall minus = (RexCall) unwrapCast(fromUnixtime.getOperands().get(0));
+        assertSame(SqlStdOperatorTable.MINUS, minus.getOperator());
+        assertSame(UnixTimestampAdapter.LOCAL_TO_UNIXTIME_OP, ((RexCall) minus.getOperands().get(0)).getOperator());
+        assertSame(UnixTimestampAdapter.LOCAL_TO_UNIXTIME_OP, ((RexCall) minus.getOperands().get(1)).getOperator());
+    }
+
+    public void testDateMinusDateRewritesToUnixSecondsDiff() {
+        RexCall call = minusOf(SqlTypeName.DATE, SqlTypeName.DATE);
+        RexNode adapted = adapter.adapt(call, List.of(), cluster);
+        // Same shape as TIMESTAMP-TIMESTAMP — DATE is the other gap-shape operand type.
+        assertSame(RustUdfDateTimeAdapters.LOCAL_FROM_UNIXTIME_OP, ((RexCall) unwrapCast(adapted)).getOperator());
+    }
+
+    public void testNumericMinusPassesThroughUnchanged() {
+        RexCall call = minusOf(SqlTypeName.INTEGER, SqlTypeName.INTEGER);
+        assertSame("numeric subtract must be returned untouched", call, adapter.adapt(call, List.of(), cluster));
+    }
+
+    public void testNonMinusOperatorPassesThrough() {
+        RelDataType ts = typeFactory.createSqlType(SqlTypeName.TIMESTAMP);
+        RexCall plus = (RexCall) rexBuilder.makeCall(
+            SqlStdOperatorTable.PLUS,
+            rexBuilder.makeInputRef(ts, 0),
+            rexBuilder.makeInputRef(ts, 1)
+        );
+        assertSame(plus, adapter.adapt(plus, List.of(), cluster));
+    }
+
+    public void testUnaryMinusPassesThrough() {
+        RelDataType i32 = typeFactory.createSqlType(SqlTypeName.INTEGER);
+        RexCall unary = (RexCall) rexBuilder.makeCall(SqlStdOperatorTable.UNARY_MINUS, rexBuilder.makeInputRef(i32, 0));
+        assertSame(unary, adapter.adapt(unary, List.of(), cluster));
+    }
+
+    public void testMinusOfWindowAggsPassesThrough() {
+        // WidthBucketAdapter pattern-matches MINUS(MAX OVER (), MIN OVER ()) for `bin <ts> bins=N`.
+        // MinusAdapter must leave it intact so the downstream adapter still sees SqlKind.MINUS.
+        RelDataType ts = typeFactory.createSqlType(SqlTypeName.TIMESTAMP);
+        RexNode tsCol = rexBuilder.makeInputRef(ts, 0);
+        RexNode maxOver = makeOverEmpty(SqlStdOperatorTable.MAX, tsCol, ts);
+        RexNode minOver = makeOverEmpty(SqlStdOperatorTable.MIN, tsCol, ts);
+        RexCall minus = (RexCall) rexBuilder.makeCall(SqlStdOperatorTable.MINUS, maxOver, minOver);
+        assertSame(minus, adapter.adapt(minus, List.of(), cluster));
+    }
+
+    private RexNode makeOverEmpty(SqlAggFunction agg, RexNode arg, RelDataType returnType) {
+        return rexBuilder.makeOver(
+            returnType,
+            agg,
+            List.of(arg),
+            List.of(),
+            ImmutableList.of(),
+            RexWindowBounds.UNBOUNDED_PRECEDING,
+            RexWindowBounds.UNBOUNDED_FOLLOWING,
+            RexWindowExclusion.EXCLUDE_NO_OTHER,
+            true,
+            true,
+            false,
+            false,
+            false
+        );
+    }
+
+    private RexCall minusOf(SqlTypeName left, SqlTypeName right) {
+        RexNode l = rexBuilder.makeInputRef(typeFactory.createSqlType(left), 0);
+        RexNode r = rexBuilder.makeInputRef(typeFactory.createSqlType(right), 1);
+        return (RexCall) rexBuilder.makeCall(SqlStdOperatorTable.MINUS, l, r);
+    }
+
+    private static RexNode unwrapCast(RexNode node) {
+        while (node instanceof RexCall call && call.getKind() == SqlKind.CAST) {
+            node = call.getOperands().get(0);
+        }
+        return node;
+    }
+}

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

@@ -273,18 +273,30 @@ public void testAdaptShapeETwoArgUnparseableTimestampThrowsIllegalArgument() {
 
     // ── adapt(): Shape F — column ref → DATETIME_ADAPTER ──────────────────
 
-    public void testAdaptShapeFColumnRefRoutesToDatetimeAdapter() {
-        RelDataType dateType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.DATE), true);
-        RexNode columnRef = rexBuilder.makeInputRef(dateType, 0);
+    public void testAdaptShapeFTimestampColumnRoutesToDatetimeAdapter() {
+        // TIMESTAMP-typed column falls through to DatetimeAdapter rename.
+        RelDataType tsType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.TIMESTAMP), true);
+        RexNode columnRef = rexBuilder.makeInputRef(tsType, 0);
         RexCall call = buildOneArgCall(columnRef);
         RexNode adapted = adapter.adapt(call, List.of(), cluster);
 
-        // Fold did NOT catch — adapter emitted a RexCall over LOCAL_TO_TIMESTAMP_OP
-        // (possibly wrapped in a CAST to align with the call's declared type).
         RexNode unwrapped = unwrapCast(adapted);
         assertTrue("expected RexCall, got " + unwrapped.getClass().getSimpleName(), unwrapped instanceof RexCall);
-        SqlOperator op = ((RexCall) unwrapped).getOperator();
-        assertSame("expected rename to LOCAL_TO_TIMESTAMP_OP", DateTimeAdapters.LOCAL_TO_TIMESTAMP_OP, op);
+        assertSame("expected rename to LOCAL_TO_TIMESTAMP_OP", DateTimeAdapters.LOCAL_TO_TIMESTAMP_OP, ((RexCall) unwrapped).getOperator());
+    }
+
+    public void testAdaptShapeFDateColumnLiftsViaCast() {
+        // DATE-typed column emits a native CAST(Date → TIMESTAMP) instead of to_timestamp(Date32).
+        RelDataType dateType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.DATE), true);
+        RexNode columnRef = rexBuilder.makeInputRef(dateType, 0);
+        RexCall call = buildOneArgCall(columnRef);
+        RexNode adapted = adapter.adapt(call, List.of(), cluster);
+
+        assertTrue("expected CAST, got " + adapted.getClass().getSimpleName(), adapted instanceof RexCall);
+        RexCall castCall = (RexCall) adapted;
+        assertEquals(SqlKind.CAST, castCall.getKind());
+        assertEquals(SqlTypeName.TIMESTAMP, castCall.getType().getSqlTypeName());
+        assertSame("CAST operand must be the original DATE column ref", columnRef, castCall.getOperands().get(0));
     }
 
     // ── helpers ───────────────────────────────────────────────────────────

sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/DateTimeScalarFunctionsIT.java

@@ -175,7 +175,27 @@ public void testStrToDate() throws IOException {
         );
     }
 
+    // ── TIMESTAMP / DATE subtraction → MinusAdapter ──────────────────────────────
 
+    public void testTimestampMinusTimestampLiterals() throws IOException {
+        assertFirstRowString(
+            oneRow("key00") + "| eval v = timestamp('1999-12-31 15:42:13') - timestamp('1961-04-12 09:07:00') | fields v",
+            "2008-09-20 06:35:13"
+        );
+    }
+
+    public void testTimestampMinusTimestampColumn() throws IOException {
+        // datetime0 at key00 == the literal → diff is epoch.
+        assertFirstRowString(
+            oneRow("key00") + "| eval v = timestamp(datetime0) - timestamp('2004-07-09 10:17:35') | fields v",
+            "1970-01-01 00:00:00"
+        );
+    }
+
+    public void testDateMinusDateLiterals() throws IOException {
+        // DATE-DATE returns integer day-count.
+        assertFirstRowLong(oneRow("key00") + "| eval v = date('2024-01-15') - date('2024-01-10') | fields v", 5L);
+    }
 
     private void assertFirstRowString(String ppl, String expected) throws IOException {
         Object cell = firstRowFirstCell(ppl);

sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/TimestampFunctionIT.java

@@ -214,6 +214,22 @@ public void testFoldedTimestampLiteralsRemainComparable() throws IOException {
         assertEquals("neq2 between adjacent-second TIMESTAMP literals must be true", Boolean.TRUE, cell);
     }
 
+    // ── Shape F-DATE: TIMESTAMP(<date column>) lifts via native CAST, not to_timestamp ──
+
+    public void testShapeFDateColumnLiftsToTimestamp() throws IOException {
+        // date0 at key00 → 2004-04-15; midnight UTC.
+        assertFirstRowString(
+            oneRow("key00") + "| eval v = date_format(timestamp(date0), '%Y-%m-%d %H:%i:%s') | fields v",
+            "2004-04-15 00:00:00"
+        );
+    }
+
+    public void testTimeEqualsDateDoesNotCrash() throws IOException {
+        // Pre-fix this lowered to to_timestamp(Date32) which DataFusion rejected at runtime.
+        Object cell = firstRowFirstCell(oneRow("key00") + "| eval v = time('00:00:00') = date('2004-07-09') | fields v");
+        assertTrue("Expected boolean result, got: " + cell, cell instanceof Boolean);
+    }
+
     // ── helpers ──────────────────────────────────────────────────────────────────
 
     private void assertFirstRowString(String ppl, String expected) throws IOException {