ESQL: Add ability to perform date math (#98870)

This adds the ability to perform date math: allow date types be updated
with a time span (a duration or a period).

Ex. : `row n = now() | eval then = n - 1 year + 2 minutes`

Also, time spans are now negateable.

docs/changelog/98870.yaml

@@ -0,0 +1,6 @@
+pr: 98870
+summary: "ESQL: Add ability to perform date math"
+area: ES|QL
+type: enhancement
+issues:
+ - 98402

x-pack/plugin/esql/qa/testFixtures/src/main/resources/date.csv-spec

@@ -453,3 +453,159 @@ emp_no:integer  |  birth_date:datetime       | birth_month:keyword
 10049           |  null                      | null
 10050           |  1958-05-21T00:00:00.000Z  | May
 ;
+
+datePlusPeriod
+row dt = to_dt("2100-01-01T01:01:01.000Z")
+| eval plus = dt + 4 years + 3 months + 2 weeks + 1 day;
+
+dt:datetime              |plus:datetime
+2100-01-01T01:01:01.000Z |2104-04-16T01:01:01.000Z
+;
+
+datePlusDuration
+row dt = to_dt("2100-01-01T00:00:00.000Z")
+| eval plus = dt + 1 hour + 1 minute + 1 second + 1 milliseconds;
+
+dt:datetime              |plus:datetime
+2100-01-01T00:00:00.000Z |2100-01-01T01:01:01.001Z
+;
+
+dateMinusPeriod
+row dt = to_dt("2104-04-16T01:01:01.000Z")
+| eval minus = dt - 4 years - 3 months - 2 weeks - 1 day;
+
+dt:datetime              |minus:datetime
+2104-04-16T01:01:01.000Z |2100-01-01T01:01:01.000Z
+;
+
+dateMinusDuration
+row dt = to_dt("2100-01-01T01:01:01.001Z")
+| eval minus = dt - 1 hour - 1 minute - 1 second - 1 milliseconds;
+
+dt:datetime              |minus:datetime
+2100-01-01T01:01:01.001Z |2100-01-01T00:00:00.000Z
+;
+
+datePlusPeriodAndDuration
+row dt = to_dt("2100-01-01T00:00:00.000Z")
+| eval plus = dt + 4 years + 3 months + 2 weeks + 1 day + 1 hour + 1 minute + 1 second + 1 milliseconds;
+
+dt:datetime              |plus:datetime
+2100-01-01T00:00:00.000Z |2104-04-16T01:01:01.001Z
+;
+
+dateMinusPeriodAndDuration
+row dt = to_dt("2104-04-16T01:01:01.001Z")
+| eval minus = dt - 4 years - 3 months - 2 weeks - 1 day - 1 hour - 1 minute - 1 second - 1 milliseconds;
+
+dt:datetime              |minus:datetime
+2104-04-16T01:01:01.001Z |2100-01-01T00:00:00.000Z
+;
+
+datePlusPeriodMinusDuration
+row dt = to_dt("2100-01-01T01:01:01.001Z")
+| eval plus = dt + 4 years + 3 months + 2 weeks + 1 day - 1 hour - 1 minute - 1 second - 1 milliseconds;
+
+dt:datetime              |plus:datetime
+2100-01-01T01:01:01.001Z |2104-04-16T00:00:00.000Z
+;
+
+datePlusDurationMinusPeriod
+row dt = to_dt("2104-04-16T00:00:00.000Z")
+| eval plus = dt - 4 years - 3 months - 2 weeks - 1 day + 1 hour + 1 minute + 1 second + 1 milliseconds;
+
+dt:datetime              |plus:datetime
+2104-04-16T00:00:00.000Z |2100-01-01T01:01:01.001Z
+;
+
+dateMathArithmeticOverflow
+row dt = to_dt(9223372036854775807)
+| eval plus = dt + 1 day
+| keep plus;
+
+warning:Line 2:15: evaluation of [dt + 1 day] failed, treating result as null. Only first 20 failures recorded.
+warning:java.lang.ArithmeticException: long overflow
+
+plus:datetime
+null
+;
+
+dateMathDateException
+row dt = to_dt(0)
+| eval plus = dt + 2147483647 years
+| keep plus;
+
+warning:Line 2:15: evaluation of [dt + 2147483647 years] failed, treating result as null. Only first 20 failures recorded.
+warning:java.time.DateTimeException: Invalid value for Year (valid values -999999999 - 999999999): 2147485617
+
+plus:datetime
+null
+;
+
+dateMathNegatedPeriod
+row dt = to_dt(0)
+| eval plus = -(-1 year) + dt
+| keep plus;
+
+plus:datetime
+1971-01-01T00:00:00.000Z
+;
+
+dateMathNegatedDuration
+row dt = to_dt(0)
+| eval plus = -(-1 second) + dt
+| keep plus;
+
+plus:datetime
+1970-01-01T00:00:01.000Z
+;
+
+
+fieldDateMathSimple
+from employees
+| eval bd = 1 year + birth_date - 1 millisecond
+| keep birth_date, bd
+| limit 5;
+
+birth_date:datetime      |bd:datetime
+1953-09-02T00:00:00.000Z |1954-09-01T23:59:59.999Z
+1964-06-02T00:00:00.000Z |1965-06-01T23:59:59.999Z
+1959-12-03T00:00:00.000Z |1960-12-02T23:59:59.999Z
+1954-05-01T00:00:00.000Z |1955-04-30T23:59:59.999Z
+1955-01-21T00:00:00.000Z |1956-01-20T23:59:59.999Z
+;
+
+fieldDateMath
+from employees
+| eval bd = -1 millisecond + birth_date + 1 year
+| eval bd = date_trunc(1 day, bd)
+| eval bd = bd + 1 day - 1 year
+| where birth_date != bd
+| stats c = count(bd);
+
+c:long
+0
+;
+
+filteringWithDateMath
+from employees
+| where birth_date < to_dt("2023-08-25T11:25:41.052Z") - 70 years
+| keep birth_date;
+
+birth_date:datetime
+1953-04-20T00:00:00.000Z
+1952-04-19T00:00:00.000Z
+1953-01-23T00:00:00.000Z
+1952-12-24T00:00:00.000Z
+1952-07-08T00:00:00.000Z
+1953-04-03T00:00:00.000Z
+1953-02-08T00:00:00.000Z
+1953-07-28T00:00:00.000Z
+1952-08-06T00:00:00.000Z
+1952-11-13T00:00:00.000Z
+1953-01-07T00:00:00.000Z
+1952-05-15T00:00:00.000Z
+1952-06-13T00:00:00.000Z
+1952-02-27T00:00:00.000Z
+1953-04-21T00:00:00.000Z
+;

x-pack/plugin/esql/src/main/java/generated/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/AddDatetimesEvaluator.java

@@ -0,0 +1,86 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License
+// 2.0; you may not use this file except in compliance with the Elastic License
+// 2.0.
+package org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic;
+
+import java.lang.ArithmeticException;
+import java.lang.Override;
+import java.lang.String;
+import java.time.DateTimeException;
+import java.time.temporal.TemporalAmount;
+import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.LongVector;
+import org.elasticsearch.compute.data.Page;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.xpack.esql.expression.function.Warnings;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+/**
+ * {@link EvalOperator.ExpressionEvaluator} implementation for {@link Add}.
+ * This class is generated. Do not edit it.
+ */
+public final class AddDatetimesEvaluator implements EvalOperator.ExpressionEvaluator {
+  private final Warnings warnings;
+
+  private final EvalOperator.ExpressionEvaluator datetime;
+
+  private final TemporalAmount temporalAmount;
+
+  public AddDatetimesEvaluator(Source source, EvalOperator.ExpressionEvaluator datetime,
+      TemporalAmount temporalAmount) {
+    this.warnings = new Warnings(source);
+    this.datetime = datetime;
+    this.temporalAmount = temporalAmount;
+  }
+
+  @Override
+  public Block eval(Page page) {
+    Block datetimeUncastBlock = datetime.eval(page);
+    if (datetimeUncastBlock.areAllValuesNull()) {
+      return Block.constantNullBlock(page.getPositionCount());
+    }
+    LongBlock datetimeBlock = (LongBlock) datetimeUncastBlock;
+    LongVector datetimeVector = datetimeBlock.asVector();
+    if (datetimeVector == null) {
+      return eval(page.getPositionCount(), datetimeBlock);
+    }
+    return eval(page.getPositionCount(), datetimeVector);
+  }
+
+  public LongBlock eval(int positionCount, LongBlock datetimeBlock) {
+    LongBlock.Builder result = LongBlock.newBlockBuilder(positionCount);
+    position: for (int p = 0; p < positionCount; p++) {
+      if (datetimeBlock.isNull(p) || datetimeBlock.getValueCount(p) != 1) {
+        result.appendNull();
+        continue position;
+      }
+      try {
+        result.appendLong(Add.processDatetimes(datetimeBlock.getLong(datetimeBlock.getFirstValueIndex(p)), temporalAmount));
+      } catch (ArithmeticException | DateTimeException e) {
+        warnings.registerException(e);
+        result.appendNull();
+      }
+    }
+    return result.build();
+  }
+
+  public LongBlock eval(int positionCount, LongVector datetimeVector) {
+    LongBlock.Builder result = LongBlock.newBlockBuilder(positionCount);
+    position: for (int p = 0; p < positionCount; p++) {
+      try {
+        result.appendLong(Add.processDatetimes(datetimeVector.getLong(p), temporalAmount));
+      } catch (ArithmeticException | DateTimeException e) {
+        warnings.registerException(e);
+        result.appendNull();
+      }
+    }
+    return result.build();
+  }
+
+  @Override
+  public String toString() {
+    return "AddDatetimesEvaluator[" + "datetime=" + datetime + ", temporalAmount=" + temporalAmount + "]";
+  }
+}

x-pack/plugin/esql/src/main/java/generated/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/SubDatetimesEvaluator.java

@@ -0,0 +1,86 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License
+// 2.0; you may not use this file except in compliance with the Elastic License
+// 2.0.
+package org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic;
+
+import java.lang.ArithmeticException;
+import java.lang.Override;
+import java.lang.String;
+import java.time.DateTimeException;
+import java.time.temporal.TemporalAmount;
+import org.elasticsearch.compute.data.Block;
+import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.LongVector;
+import org.elasticsearch.compute.data.Page;
+import org.elasticsearch.compute.operator.EvalOperator;
+import org.elasticsearch.xpack.esql.expression.function.Warnings;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+/**
+ * {@link EvalOperator.ExpressionEvaluator} implementation for {@link Sub}.
+ * This class is generated. Do not edit it.
+ */
+public final class SubDatetimesEvaluator implements EvalOperator.ExpressionEvaluator {
+  private final Warnings warnings;
+
+  private final EvalOperator.ExpressionEvaluator datetime;
+
+  private final TemporalAmount temporalAmount;
+
+  public SubDatetimesEvaluator(Source source, EvalOperator.ExpressionEvaluator datetime,
+      TemporalAmount temporalAmount) {
+    this.warnings = new Warnings(source);
+    this.datetime = datetime;
+    this.temporalAmount = temporalAmount;
+  }
+
+  @Override
+  public Block eval(Page page) {
+    Block datetimeUncastBlock = datetime.eval(page);
+    if (datetimeUncastBlock.areAllValuesNull()) {
+      return Block.constantNullBlock(page.getPositionCount());
+    }
+    LongBlock datetimeBlock = (LongBlock) datetimeUncastBlock;
+    LongVector datetimeVector = datetimeBlock.asVector();
+    if (datetimeVector == null) {
+      return eval(page.getPositionCount(), datetimeBlock);
+    }
+    return eval(page.getPositionCount(), datetimeVector);
+  }
+
+  public LongBlock eval(int positionCount, LongBlock datetimeBlock) {
+    LongBlock.Builder result = LongBlock.newBlockBuilder(positionCount);
+    position: for (int p = 0; p < positionCount; p++) {
+      if (datetimeBlock.isNull(p) || datetimeBlock.getValueCount(p) != 1) {
+        result.appendNull();
+        continue position;
+      }
+      try {
+        result.appendLong(Sub.processDatetimes(datetimeBlock.getLong(datetimeBlock.getFirstValueIndex(p)), temporalAmount));
+      } catch (ArithmeticException | DateTimeException e) {
+        warnings.registerException(e);
+        result.appendNull();
+      }
+    }
+    return result.build();
+  }
+
+  public LongBlock eval(int positionCount, LongVector datetimeVector) {
+    LongBlock.Builder result = LongBlock.newBlockBuilder(positionCount);
+    position: for (int p = 0; p < positionCount; p++) {
+      try {
+        result.appendLong(Sub.processDatetimes(datetimeVector.getLong(p), temporalAmount));
+      } catch (ArithmeticException | DateTimeException e) {
+        warnings.registerException(e);
+        result.appendNull();
+      }
+    }
+    return result.build();
+  }
+
+  @Override
+  public String toString() {
+    return "SubDatetimesEvaluator[" + "datetime=" + datetime + ", temporalAmount=" + temporalAmount + "]";
+  }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/date/DateTrunc.java

@@ -15,11 +15,9 @@ import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.xpack.esql.evaluator.mapper.EvaluatorMapper;
 import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 import org.elasticsearch.xpack.ql.expression.Expression;
-import org.elasticsearch.xpack.ql.expression.TypeResolutions;
 import org.elasticsearch.xpack.ql.expression.function.scalar.BinaryScalarFunction;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
-import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.DataTypes;
 
 import java.time.Duration;
@@ -30,6 +28,7 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.common.logging.LoggerMessageFormat.format;
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isTemporalAmount;
 import static org.elasticsearch.xpack.ql.expression.TypeResolutions.ParamOrdinal.FIRST;
 import static org.elasticsearch.xpack.ql.expression.TypeResolutions.ParamOrdinal.SECOND;
 import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isDate;
@@ -57,30 +56,17 @@ public class DateTrunc extends BinaryDateTimeFunction implements EvaluatorMapper
             return resolution;
         }
 
-        return isInterval(interval(), sourceText(), SECOND);
+        return isType(interval(), EsqlDataTypes::isTemporalAmount, sourceText(), SECOND, "dateperiod", "timeduration");
     }
 
     // TODO: drop check once 8.11 is released
     private TypeResolution argumentTypesAreSwapped() {
-        DataType leftType = left().dataType();
-        DataType rightType = right().dataType();
-        if (leftType == DataTypes.DATETIME && (rightType == EsqlDataTypes.DATE_PERIOD || rightType == EsqlDataTypes.TIME_DURATION)) {
+        if (DataTypes.isDateTime(left().dataType()) && isTemporalAmount(right().dataType())) {
             return new TypeResolution(format(null, "function definition has been updated, please swap arguments in [{}]", sourceText()));
         }
         return TypeResolution.TYPE_RESOLVED;
     }
 
-    private static TypeResolution isInterval(Expression e, String operationName, TypeResolutions.ParamOrdinal paramOrd) {
-        return isType(
-            e,
-            dt -> dt == EsqlDataTypes.DATE_PERIOD || dt == EsqlDataTypes.TIME_DURATION,
-            operationName,
-            paramOrd,
-            "dateperiod",
-            "timeduration"
-        );
-    }
-
     @Override
     public Object fold() {
         return EvaluatorMapper.super.fold();

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/Add.java

@@ -8,15 +8,21 @@
 package org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic;
 
 import org.elasticsearch.compute.ann.Evaluator;
+import org.elasticsearch.compute.ann.Fixed;
 import org.elasticsearch.xpack.ql.expression.Expression;
 import org.elasticsearch.xpack.ql.expression.predicate.operator.arithmetic.BinaryComparisonInversible;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 
+import java.time.DateTimeException;
+import java.time.temporal.TemporalAmount;
+
 import static org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic.EsqlArithmeticOperation.OperationSymbol.ADD;
+import static org.elasticsearch.xpack.ql.type.DateUtils.asDateTime;
+import static org.elasticsearch.xpack.ql.type.DateUtils.asMillis;
 import static org.elasticsearch.xpack.ql.util.NumericUtils.unsignedLongAddExact;
 
-public class Add extends EsqlArithmeticOperation implements BinaryComparisonInversible {
+public class Add extends DateTimeArithmeticOperation implements BinaryComparisonInversible {
 
     public Add(Source source, Expression left, Expression right) {
         super(
@@ -27,7 +33,8 @@ public class Add extends EsqlArithmeticOperation implements BinaryComparisonInve
             AddIntsEvaluator::new,
             AddLongsEvaluator::new,
             AddUnsignedLongsEvaluator::new,
-            (s, l, r) -> new AddDoublesEvaluator(l, r)
+            (s, l, r) -> new AddDoublesEvaluator(l, r),
+            AddDatetimesEvaluator::new
         );
     }
 
@@ -75,4 +82,10 @@ public class Add extends EsqlArithmeticOperation implements BinaryComparisonInve
     static double processDoubles(double lhs, double rhs) {
         return lhs + rhs;
     }
+
+    @Evaluator(extraName = "Datetimes", warnExceptions = { ArithmeticException.class, DateTimeException.class })
+    static long processDatetimes(long datetime, @Fixed TemporalAmount temporalAmount) {
+        // using a UTC conversion since `datetime` is always a UTC-Epoch timestamp, either read from ES or converted through a function
+        return asMillis(asDateTime(datetime).plus(temporalAmount));
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/DateTimeArithmeticOperation.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic;
+
+import org.elasticsearch.common.TriFunction;
+import org.elasticsearch.compute.operator.EvalOperator.ExpressionEvaluator;
+import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.tree.Source;
+import org.elasticsearch.xpack.ql.type.DataType;
+import org.elasticsearch.xpack.ql.type.DataTypes;
+
+import java.time.temporal.TemporalAmount;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+import static org.elasticsearch.common.logging.LoggerMessageFormat.format;
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isDateTimeOrTemporal;
+
+abstract class DateTimeArithmeticOperation extends EsqlArithmeticOperation {
+
+    interface DatetimeArithmeticEvaluator extends TriFunction<Source, ExpressionEvaluator, TemporalAmount, ExpressionEvaluator> {};
+
+    private final DatetimeArithmeticEvaluator datetimes;
+
+    DateTimeArithmeticOperation(
+        Source source,
+        Expression left,
+        Expression right,
+        OperationSymbol op,
+        ArithmeticEvaluator ints,
+        ArithmeticEvaluator longs,
+        ArithmeticEvaluator ulongs,
+        ArithmeticEvaluator doubles,
+        DatetimeArithmeticEvaluator datetimes
+    ) {
+        super(source, left, right, op, ints, longs, ulongs, doubles);
+        this.datetimes = datetimes;
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        DataType leftType = left().dataType();
+        DataType rightType = right().dataType();
+        // date math is only possible if one argument is a DATETIME and the other a (foldable) TemporalValue
+        if (isDateTimeOrTemporal(leftType) || isDateTimeOrTemporal(rightType)) {
+            if (argumentOfType(DataTypes::isDateTime) == null || argumentOfType(EsqlDataTypes::isTemporalAmount) == null) {
+                return new TypeResolution(
+                    format(null, "[{}] has arguments with incompatible types [{}] and [{}]", symbol(), leftType, rightType)
+                );
+            }
+            return TypeResolution.TYPE_RESOLVED;
+        }
+        return super.resolveType();
+    }
+
+    @Override
+    public Supplier<ExpressionEvaluator> toEvaluator(Function<Expression, Supplier<ExpressionEvaluator>> toEvaluator) {
+        return dataType() == DataTypes.DATETIME
+            ? () -> datetimes.apply(
+                source(),
+                toEvaluator.apply(argumentOfType(DataTypes::isDateTime)).get(),
+                (TemporalAmount) argumentOfType(EsqlDataTypes::isTemporalAmount).fold()
+            )
+            : super.toEvaluator(toEvaluator);
+    }
+
+    private Expression argumentOfType(Predicate<DataType> filter) {
+        return filter.test(left().dataType()) ? left() : filter.test(right().dataType()) ? right() : null;
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/EsqlArithmeticOperation.java

@@ -23,6 +23,7 @@ import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataType;
 
 import java.io.IOException;
+import java.util.function.Function;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.xpack.ql.type.DataTypes.DOUBLE;
@@ -110,9 +111,7 @@ abstract class EsqlArithmeticOperation extends ArithmeticOperation implements Ev
     }
 
     @Override
-    public final Supplier<ExpressionEvaluator> toEvaluator(
-        java.util.function.Function<Expression, Supplier<ExpressionEvaluator>> toEvaluator
-    ) {
+    public Supplier<ExpressionEvaluator> toEvaluator(Function<Expression, Supplier<ExpressionEvaluator>> toEvaluator) {
         var commonType = dataType();
         var leftType = left().dataType();
         if (leftType.isNumeric()) {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/Neg.java

@@ -11,24 +11,32 @@ import org.elasticsearch.compute.ann.Evaluator;
 import org.elasticsearch.compute.operator.EvalOperator.ExpressionEvaluator;
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.evaluator.mapper.EvaluatorMapper;
+import org.elasticsearch.xpack.esql.expression.function.Warnings;
 import org.elasticsearch.xpack.esql.expression.function.scalar.UnaryScalarFunction;
 import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.expression.Literal;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.DataTypes;
 
+import java.time.Duration;
+import java.time.Period;
 import java.util.List;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isTemporalAmount;
 import static org.elasticsearch.xpack.ql.expression.TypeResolutions.ParamOrdinal.DEFAULT;
-import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isNumeric;
+import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isType;
 
 public class Neg extends UnaryScalarFunction implements EvaluatorMapper {
 
+    private final Warnings warnings;
+
     public Neg(Source source, Expression field) {
         super(source, field);
+        warnings = new Warnings(source);
     }
 
     @Override
@@ -52,18 +60,50 @@ public class Neg extends UnaryScalarFunction implements EvaluatorMapper {
             if (supplier != null) {
                 return supplier;
             }
+        } else if (isTemporalAmount(type)) {
+            return toEvaluator.apply(field());
         }
         throw new EsqlIllegalArgumentException("arithmetic negation operator with unsupported data type [" + type + "]");
     }
 
     @Override
     public final Object fold() {
+        if (isTemporalAmount(field().dataType()) && field() instanceof Literal literal) {
+            return foldTemporalAmount(literal);
+        }
         return EvaluatorMapper.super.fold();
     }
 
+    private Object foldTemporalAmount(Literal literal) {
+        try {
+            Object value = literal.fold();
+            if (value instanceof Period period) {
+                return period.negated();
+            }
+            if (value instanceof Duration duration) {
+                return duration.negated();
+            }
+        } catch (ArithmeticException ae) {
+            warnings.registerException(ae);
+            return null;
+        }
+
+        throw new EsqlIllegalArgumentException(
+            "unexpected non-temporal amount literal [" + literal.sourceText() + "] of type [" + literal.dataType() + "]"
+        );
+    }
+
     @Override
     protected TypeResolution resolveType() {
-        return isNumeric(field(), sourceText(), DEFAULT);
+        return isType(
+            field(),
+            dt -> dt.isNumeric() || isTemporalAmount(dt),
+            sourceText(),
+            DEFAULT,
+            "numeric",
+            "date_period",
+            "time_duration"
+        );
     }
 
     @Override

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/Sub.java

@@ -8,15 +8,24 @@
 package org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic;
 
 import org.elasticsearch.compute.ann.Evaluator;
+import org.elasticsearch.compute.ann.Fixed;
+import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 import org.elasticsearch.xpack.ql.expression.Expression;
 import org.elasticsearch.xpack.ql.expression.predicate.operator.arithmetic.BinaryComparisonInversible;
 import org.elasticsearch.xpack.ql.tree.NodeInfo;
 import org.elasticsearch.xpack.ql.tree.Source;
+import org.elasticsearch.xpack.ql.type.DataTypes;
 
+import java.time.DateTimeException;
+import java.time.temporal.TemporalAmount;
+
+import static org.elasticsearch.common.logging.LoggerMessageFormat.format;
 import static org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic.EsqlArithmeticOperation.OperationSymbol.SUB;
+import static org.elasticsearch.xpack.ql.type.DateUtils.asDateTime;
+import static org.elasticsearch.xpack.ql.type.DateUtils.asMillis;
 import static org.elasticsearch.xpack.ql.util.NumericUtils.unsignedLongSubtractExact;
 
-public class Sub extends EsqlArithmeticOperation implements BinaryComparisonInversible {
+public class Sub extends DateTimeArithmeticOperation implements BinaryComparisonInversible {
 
     public Sub(Source source, Expression left, Expression right) {
         super(
@@ -27,10 +36,30 @@ public class Sub extends EsqlArithmeticOperation implements BinaryComparisonInve
             SubIntsEvaluator::new,
             SubLongsEvaluator::new,
             SubUnsignedLongsEvaluator::new,
-            (s, l, r) -> new SubDoublesEvaluator(l, r)
+            (s, l, r) -> new SubDoublesEvaluator(l, r),
+            SubDatetimesEvaluator::new
         );
     }
 
+    @Override
+    protected TypeResolution resolveType() {
+        TypeResolution resolution = super.resolveType();
+        if (resolution.resolved() && EsqlDataTypes.isDateTimeOrTemporal(dataType()) && DataTypes.isDateTime(left().dataType()) == false) {
+            return new TypeResolution(
+                format(
+                    null,
+                    "[{}] arguments are in unsupported order: cannot subtract a [{}] value [{}] from a [{}] amount [{}]",
+                    symbol(),
+                    right().dataType(),
+                    right().sourceText(),
+                    left().dataType(),
+                    left().sourceText()
+                )
+            );
+        }
+        return resolution;
+    }
+
     @Override
     protected NodeInfo<Sub> info() {
         return NodeInfo.create(this, Sub::new, left(), right());
@@ -65,4 +94,10 @@ public class Sub extends EsqlArithmeticOperation implements BinaryComparisonInve
     static double processDoubles(double lhs, double rhs) {
         return lhs - rhs;
     }
+
+    @Evaluator(extraName = "Datetimes", warnExceptions = { ArithmeticException.class, DateTimeException.class })
+    static long processDatetimes(long datetime, @Fixed TemporalAmount temporalAmount) {
+        // using a UTC conversion since `datetime` is always a UTC-Epoch timestamp, either read from ES or converted through a function
+        return asMillis(asDateTime(datetime).minus(temporalAmount));
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/type/EsqlDataTypeRegistry.java

@@ -10,9 +10,13 @@ package org.elasticsearch.xpack.esql.type;
 import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.DataTypeConverter;
 import org.elasticsearch.xpack.ql.type.DataTypeRegistry;
+import org.elasticsearch.xpack.ql.type.DataTypes;
 
 import java.util.Collection;
 
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isTemporalAmount;
+import static org.elasticsearch.xpack.ql.type.DataTypes.isDateTime;
+
 public class EsqlDataTypeRegistry implements DataTypeRegistry {
 
     public static final DataTypeRegistry INSTANCE = new EsqlDataTypeRegistry();
@@ -51,6 +55,9 @@ public class EsqlDataTypeRegistry implements DataTypeRegistry {
 
     @Override
     public DataType commonType(DataType left, DataType right) {
+        if (isDateTime(left) && isTemporalAmount(right) || isTemporalAmount(left) && isDateTime(right)) {
+            return DataTypes.DATETIME;
+        }
         return DataTypeConverter.commonType(left, right);
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/type/EsqlDataTypes.java

@@ -137,6 +137,14 @@ public final class EsqlDataTypes {
         return t != OBJECT && t != NESTED;
     }
 
+    public static boolean isDateTimeOrTemporal(DataType t) {
+        return DataTypes.isDateTime(t) || isTemporalAmount(t);
+    }
+
+    public static boolean isTemporalAmount(DataType t) {
+        return t == DATE_PERIOD || t == TIME_DURATION;
+    }
+
     /**
      * Supported types that can be contained in a block.
      */

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/analysis/VerifierTests.java

@@ -250,6 +250,25 @@ public class VerifierTests extends ESTestCase {
         }
     }
 
+    public void testSubtractDateTimeFromTemporal() {
+        for (var unit : List.of("millisecond", "second", "minute", "hour")) {
+            assertEquals(
+                "1:5: [-] arguments are in unsupported order: cannot subtract a [DATETIME] value [now()] from a [TIME_DURATION] amount [1 "
+                    + unit
+                    + "]",
+                error("row 1 " + unit + " - now() ")
+            );
+        }
+        for (var unit : List.of("day", "week", "month", "year")) {
+            assertEquals(
+                "1:5: [-] arguments are in unsupported order: cannot subtract a [DATETIME] value [now()] from a [DATE_PERIOD] amount [1 "
+                    + unit
+                    + "]",
+                error("row 1 " + unit + " - now() ")
+            );
+        }
+    }
+
     private String error(String query) {
         return error(query, defaultAnalyzer);
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/AbstractFunctionTestCase.java

@@ -103,7 +103,7 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
         private List<TypedData> data;
 
         /**
-         * The expected toString output for the evaluator this fuction invocation should generate
+         * The expected toString output for the evaluator this function invocation should generate
          */
         String evaluatorToString;
         /**
@@ -122,6 +122,8 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
 
         private final String expectedTypeError;
 
+        private final boolean allTypesAreRepresentable;
+
         public TestCase(List<TypedData> data, String evaluatorToString, DataType expectedType, Matcher<Object> matcher) {
             this(data, evaluatorToString, expectedType, matcher, null, null);
         }
@@ -145,6 +147,7 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
             this.matcher = matcher;
             this.expectedWarnings = expectedWarnings;
             this.expectedTypeError = expectedTypeError;
+            this.allTypesAreRepresentable = data.stream().allMatch(d -> EsqlDataTypes.isRepresentable(d.type));
         }
 
         public Source getSource() {
@@ -167,6 +170,10 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
             return data.stream().map(t -> t.data()).collect(Collectors.toList());
         }
 
+        public boolean allTypesAreRepresentable() {
+            return allTypesAreRepresentable;
+        }
+
         public Matcher<Object> getMatcher() {
             return matcher;
         }
@@ -531,6 +538,7 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
     }
 
     public final void testEvaluate() {
+        assumeTrue("All test data types must be representable in order to build fields", testCase.allTypesAreRepresentable);
         Expression expression = buildFieldExpression(testCase);
         if (testCase.expectedTypeError != null) {
             assertTrue("expected unresolved", expression.typeResolved().unresolved());
@@ -550,6 +558,7 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
 
     public final void testSimpleWithNulls() { // TODO replace this with nulls inserted into the test case like anyNullIsNull
         assumeTrue("nothing to do if a type error", testCase.expectedTypeError == null);
+        assumeTrue("All test data types must be representable in order to build fields", testCase.allTypesAreRepresentable);
         List<Object> simpleData = testCase.getDataValues();
         EvalOperator.ExpressionEvaluator eval = evaluator(buildFieldExpression(testCase)).get();
         Block[] orig = BlockUtils.fromListRow(simpleData);
@@ -576,6 +585,7 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
 
     public final void testEvaluateInManyThreads() throws ExecutionException, InterruptedException {
         assumeTrue("nothing to do if a type error", testCase.expectedTypeError == null);
+        assumeTrue("All test data types must be representable in order to build fields", testCase.allTypesAreRepresentable);
         int count = 10_000;
         int threads = 5;
         Supplier<EvalOperator.ExpressionEvaluator> evalSupplier = evaluator(buildFieldExpression(testCase));
@@ -603,6 +613,7 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
 
     public final void testEvaluatorToString() {
         assumeTrue("nothing to do if a type error", testCase.expectedTypeError == null);
+        assumeTrue("All test data types must be representable in order to build fields", testCase.allTypesAreRepresentable);
         var supplier = evaluator(buildFieldExpression(testCase));
         var ev = supplier.get();
         assertThat(ev.toString(), equalTo(testCase.evaluatorToString));
@@ -626,6 +637,7 @@ public abstract class AbstractFunctionTestCase extends ESTestCase {
     }
 
     public void testSerializationOfSimple() {
+        assumeTrue("All test data types must be representable in order to build fields", testCase.allTypesAreRepresentable);
         assertSerialization(buildFieldExpression(testCase));
     }
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/AbstractScalarFunctionTestCase.java

@@ -25,6 +25,7 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
+import java.util.stream.Stream;
 
 import static org.hamcrest.Matchers.equalTo;
 
@@ -67,6 +68,14 @@ public abstract class AbstractScalarFunctionTestCase extends AbstractFunctionTes
         return realValidTypes;
     }
 
+    public Set<DataType> sortedTypesSet(DataType[] validTypes, DataType... additionalTypes) {
+        Set<DataType> mergedSet = new LinkedHashSet<>();
+        Stream.concat(Stream.of(validTypes), Stream.of(additionalTypes))
+            .sorted(Comparator.comparing(DataType::name))
+            .forEach(mergedSet::add);
+        return mergedSet;
+    }
+
     /**
      * All string types (keyword, text, match_only_text, etc). For passing to {@link #required} or {@link #optional}.
      */
@@ -179,6 +188,12 @@ public abstract class AbstractScalarFunctionTestCase extends AbstractFunctionTes
         if (withoutNull.equals(List.of(DataTypes.DATETIME))) {
             return "datetime";
         }
+        List<DataType> negations = Stream.concat(Stream.of(numerics()), Stream.of(EsqlDataTypes.DATE_PERIOD, EsqlDataTypes.TIME_DURATION))
+            .sorted(Comparator.comparing(DataType::name))
+            .toList();
+        if (withoutNull.equals(negations)) {
+            return "numeric, date_period or time_duration";
+        }
         if (validTypes.equals(Set.copyOf(Arrays.asList(representable())))) {
             return "representable";
         }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/predicate/operator/AbstractBinaryOperatorTestCase.java

@@ -24,7 +24,9 @@ import java.util.List;
 import java.util.Locale;
 
 import static org.elasticsearch.compute.data.BlockUtils.toJavaObject;
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isRepresentable;
 import static org.elasticsearch.xpack.ql.type.DataTypeConverter.commonType;
+import static org.elasticsearch.xpack.ql.type.DataTypes.isNull;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
@@ -52,37 +54,48 @@ public abstract class AbstractBinaryOperatorTestCase extends AbstractFunctionTes
 
     protected abstract BinaryOperator<?, ?, ?, ?> build(Source source, Expression lhs, Expression rhs);
 
+    /**
+     * What type is acceptable for any of the function parameters.
+     * @param type The type to probe.
+     * @return True if the type is supported by the respective function.
+     */
     protected abstract boolean supportsType(DataType type);
 
+    /**
+     * What combination of parameter types are acceptable by the function.
+     * @param lhsType Left argument type.
+     * @param rhsType Right argument type.
+     * @return True if the type combination is supported by the respective function.
+     */
+    protected boolean supportsTypes(DataType lhsType, DataType rhsType) {
+        if (isNull(lhsType) || isNull(rhsType)) {
+            return false;
+        }
+        if ((lhsType == DataTypes.UNSIGNED_LONG || rhsType == DataTypes.UNSIGNED_LONG) && lhsType != rhsType) {
+            // UL can only be operated on together with another UL, so skip non-UL&UL combinations
+            return false;
+        }
+        return supportsType(lhsType) && supportsType(rhsType);
+    }
+
     public final void testApplyToAllTypes() {
         for (DataType lhsType : EsqlDataTypes.types()) {
-            if (EsqlDataTypes.isRepresentable(lhsType) == false || lhsType == DataTypes.NULL) {
-                continue;
-            }
-            if (supportsType(lhsType) == false) {
-                continue;
-            }
-            Literal lhs = randomLiteral(lhsType);
             for (DataType rhsType : EsqlDataTypes.types()) {
-                if (EsqlDataTypes.isRepresentable(rhsType) == false || rhsType == DataTypes.NULL) {
-                    continue;
-                }
-                if (supportsType(rhsType) == false) {
-                    continue;
-                }
-                if (false == (lhsType == rhsType || lhsType.isNumeric() && rhsType.isNumeric())) {
-                    continue;
-                }
-                if (lhsType != rhsType && (lhsType == DataTypes.UNSIGNED_LONG || rhsType == DataTypes.UNSIGNED_LONG)) {
+                if (supportsTypes(lhsType, rhsType) == false) {
                     continue;
                 }
+                Literal lhs = randomLiteral(lhsType);
                 Literal rhs = randomValueOtherThanMany(l -> rhsOk(l.value()) == false, () -> randomLiteral(rhsType));
-                BinaryOperator<?, ?, ?, ?> op = build(
-                    new Source(Location.EMPTY, lhsType.typeName() + " " + rhsType.typeName()),
-                    field("lhs", lhsType),
-                    field("rhs", rhsType)
-                );
-                Object result = toJavaObject(evaluator(op).get().eval(row(List.of(lhs.value(), rhs.value()))), 0);
+                Object result;
+                BinaryOperator<?, ?, ?, ?> op;
+                Source src = new Source(Location.EMPTY, lhsType.typeName() + " " + rhsType.typeName());
+                if (isRepresentable(lhsType) && isRepresentable(rhsType)) {
+                    op = build(src, field("lhs", lhsType), field("rhs", rhsType));
+                    result = toJavaObject(evaluator(op).get().eval(row(List.of(lhs.value(), rhs.value()))), 0);
+                } else {
+                    op = build(src, lhs, rhs);
+                    result = op.fold();
+                }
                 if (result == null) {
                     assertCriticalWarnings(
                         "Line -1:-1: evaluation of [" + op + "] failed, treating result as null. Only first 20 failures recorded.",
@@ -100,12 +113,12 @@ public abstract class AbstractBinaryOperatorTestCase extends AbstractFunctionTes
 
     public final void testResolveType() {
         for (DataType lhsType : EsqlDataTypes.types()) {
-            if (EsqlDataTypes.isRepresentable(lhsType) == false) {
+            if (isRepresentable(lhsType) == false) {
                 continue;
             }
             Literal lhs = randomLiteral(lhsType);
             for (DataType rhsType : EsqlDataTypes.types()) {
-                if (EsqlDataTypes.isRepresentable(rhsType) == false) {
+                if (isRepresentable(rhsType) == false) {
                     continue;
                 }
                 Literal rhs = randomLiteral(rhsType);

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/AbstractArithmeticTestCase.java

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic;
 
 import org.elasticsearch.xpack.esql.expression.predicate.operator.AbstractBinaryOperatorTestCase;
+import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 import org.elasticsearch.xpack.ql.expression.predicate.BinaryOperator;
 import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.DataTypes;
@@ -21,7 +22,7 @@ import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 
 public abstract class AbstractArithmeticTestCase extends AbstractBinaryOperatorTestCase {
-    protected final Matcher<Object> resultMatcher(List<Object> data, DataType dataType) {
+    protected Matcher<Object> resultMatcher(List<Object> data, DataType dataType) {
         Number lhs = (Number) data.get(0);
         Number rhs = (Number) data.get(1);
         if (lhs instanceof Double || rhs instanceof Double) {
@@ -67,12 +68,12 @@ public abstract class AbstractArithmeticTestCase extends AbstractBinaryOperatorT
     protected abstract long expectedUnsignedLongValue(long lhs, long rhs);
 
     @Override
-    protected final boolean supportsType(DataType type) {
-        return type.isNumeric();
+    protected boolean supportsType(DataType type) {
+        return type.isNumeric() && EsqlDataTypes.isRepresentable(type);
     }
 
     @Override
-    protected final void validateType(BinaryOperator<?, ?, ?, ?> op, DataType lhsType, DataType rhsType) {
+    protected void validateType(BinaryOperator<?, ?, ?, ?> op, DataType lhsType, DataType rhsType) {
         if (DataTypes.isNullOrNumeric(lhsType) && DataTypes.isNullOrNumeric(rhsType)) {
             assertTrue(op.toString(), op.typeResolved().resolved());
             assertThat(op.toString(), op.dataType(), equalTo(expectedType(lhsType, rhsType)));
@@ -97,7 +98,7 @@ public abstract class AbstractArithmeticTestCase extends AbstractBinaryOperatorT
         );
     }
 
-    private DataType expectedType(DataType lhsType, DataType rhsType) {
+    protected DataType expectedType(DataType lhsType, DataType rhsType) {
         if (lhsType == DataTypes.DOUBLE || rhsType == DataTypes.DOUBLE) {
             return DataTypes.DOUBLE;
         }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/AbstractDateTimeArithmeticTestCase.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic;
+
+import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
+import org.elasticsearch.xpack.ql.expression.predicate.BinaryOperator;
+import org.elasticsearch.xpack.ql.type.DataType;
+import org.elasticsearch.xpack.ql.type.DataTypes;
+import org.hamcrest.Matcher;
+
+import java.time.temporal.TemporalAmount;
+import java.util.List;
+import java.util.Locale;
+
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isDateTimeOrTemporal;
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isTemporalAmount;
+import static org.elasticsearch.xpack.ql.type.DataTypes.isDateTime;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.oneOf;
+
+public abstract class AbstractDateTimeArithmeticTestCase extends AbstractArithmeticTestCase {
+
+    @Override
+    protected Matcher<Object> resultMatcher(List<Object> data, DataType dataType) {
+        Object lhs = data.get(0);
+        Object rhs = data.get(1);
+        if (lhs instanceof TemporalAmount || rhs instanceof TemporalAmount) {
+            TemporalAmount temporal = lhs instanceof TemporalAmount leftTemporal ? leftTemporal : (TemporalAmount) rhs;
+            long datetime = temporal == lhs ? (Long) rhs : (Long) lhs;
+            return equalTo(expectedValue(datetime, temporal));
+        }
+        return super.resultMatcher(data, dataType);
+    }
+
+    protected abstract long expectedValue(long datetime, TemporalAmount temporalAmount);
+
+    @Override
+    protected final boolean supportsType(DataType type) {
+        return EsqlDataTypes.isDateTimeOrTemporal(type) || super.supportsType(type);
+    }
+
+    @Override
+    protected void validateType(BinaryOperator<?, ?, ?, ?> op, DataType lhsType, DataType rhsType) {
+        if (isDateTime(lhsType) && isTemporalAmount(rhsType) || isTemporalAmount(lhsType) && isDateTime(rhsType)) {
+            assertTrue(op.toString(), op.typeResolved().resolved());
+            assertTrue(op.toString(), isTemporalAmount(lhsType) || isTemporalAmount(rhsType));
+            assertFalse(op.toString(), isTemporalAmount(lhsType) && isTemporalAmount(rhsType));
+            assertThat(op.toString(), op.dataType(), equalTo(expectedType(lhsType, rhsType)));
+            assertThat(op.toString(), op.getClass(), oneOf(Add.class, Sub.class));
+        } else if (isDateTimeOrTemporal(lhsType) || isDateTimeOrTemporal(rhsType)) {
+            assertFalse(op.toString(), op.typeResolved().resolved());
+            assertThat(
+                op.toString(),
+                op.typeResolved().message(),
+                equalTo(
+                    String.format(Locale.ROOT, "[%s] has arguments with incompatible types [%s] and [%s]", op.symbol(), lhsType, rhsType)
+                )
+            );
+        } else {
+            super.validateType(op, lhsType, rhsType);
+        }
+    }
+
+    @Override
+    protected DataType expectedType(DataType lhsType, DataType rhsType) {
+        return isDateTimeOrTemporal(lhsType) ? DataTypes.DATETIME : super.expectedType(lhsType, rhsType);
+    }
+}

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/AddTests.java

@@ -10,19 +10,29 @@ package org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic;
 import com.carrotsearch.randomizedtesting.annotations.Name;
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
+import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 import org.elasticsearch.xpack.ql.expression.Expression;
 import org.elasticsearch.xpack.ql.tree.Source;
+import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.DataTypes;
 
 import java.math.BigInteger;
+import java.time.Duration;
+import java.time.Period;
+import java.time.temporal.TemporalAmount;
 import java.util.List;
 import java.util.function.Supplier;
 
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isDateTimeOrTemporal;
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isTemporalAmount;
+import static org.elasticsearch.xpack.ql.type.DataTypes.isDateTime;
+import static org.elasticsearch.xpack.ql.type.DateUtils.asDateTime;
+import static org.elasticsearch.xpack.ql.type.DateUtils.asMillis;
 import static org.elasticsearch.xpack.ql.util.NumericUtils.asLongUnsigned;
 import static org.elasticsearch.xpack.ql.util.NumericUtils.unsignedLongAsBigInteger;
 import static org.hamcrest.Matchers.equalTo;
 
-public class AddTests extends AbstractArithmeticTestCase {
+public class AddTests extends AbstractDateTimeArithmeticTestCase {
     public AddTests(@Name("TestCase") Supplier<TestCase> testCaseSupplier) {
         this.testCase = testCaseSupplier.get();
     }
@@ -71,9 +81,51 @@ public class AddTests extends AbstractArithmeticTestCase {
                 "AddUnsignedLongsEvaluator[lhs=Attribute[channel=0], rhs=Attribute[channel=1]]",
                 equalTo(asLongUnsigned(lhsBI.add(rhsBI).longValue()))
             );
-          })
-          */
-        ));
+          }) */, new TestCaseSupplier("Datetime + Period", () -> {
+            long lhs = (Long) randomLiteral(DataTypes.DATETIME).value();
+            Period rhs = (Period) randomLiteral(EsqlDataTypes.DATE_PERIOD).value();
+            return new TestCase(
+                List.of(new TypedData(lhs, DataTypes.DATETIME, "lhs"), new TypedData(rhs, EsqlDataTypes.DATE_PERIOD, "rhs")),
+                "AddDatetimesEvaluator[lhs=Attribute[channel=0], rhs=Attribute[channel=1]]",
+                DataTypes.DATETIME,
+                equalTo(asMillis(asDateTime(lhs).plus(rhs)))
+            );
+        }), new TestCaseSupplier("Period + Datetime", () -> {
+            Period lhs = (Period) randomLiteral(EsqlDataTypes.DATE_PERIOD).value();
+            long rhs = (Long) randomLiteral(DataTypes.DATETIME).value();
+            return new TestCase(
+                List.of(new TypedData(lhs, EsqlDataTypes.DATE_PERIOD, "lhs"), new TypedData(rhs, DataTypes.DATETIME, "rhs")),
+                "AddDatetimesEvaluator[lhs=Attribute[channel=0], rhs=Attribute[channel=1]]",
+                DataTypes.DATETIME,
+                equalTo(asMillis(asDateTime(rhs).plus(lhs)))
+            );
+        }), new TestCaseSupplier("Datetime + Duration", () -> {
+            long lhs = (Long) randomLiteral(DataTypes.DATETIME).value();
+            Duration rhs = (Duration) randomLiteral(EsqlDataTypes.TIME_DURATION).value();
+            return new TestCase(
+                List.of(new TypedData(lhs, DataTypes.DATETIME, "lhs"), new TypedData(rhs, EsqlDataTypes.TIME_DURATION, "rhs")),
+                "AddDatetimesEvaluator[lhs=Attribute[channel=0], rhs=Attribute[channel=1]]",
+                DataTypes.DATETIME,
+                equalTo(asMillis(asDateTime(lhs).plus(rhs)))
+            );
+        }), new TestCaseSupplier("Duration + Datetime", () -> {
+            long lhs = (Long) randomLiteral(DataTypes.DATETIME).value();
+            Duration rhs = (Duration) randomLiteral(EsqlDataTypes.TIME_DURATION).value();
+            return new TestCase(
+                List.of(new TypedData(lhs, DataTypes.DATETIME, "lhs"), new TypedData(rhs, EsqlDataTypes.TIME_DURATION, "rhs")),
+                "AddDatetimesEvaluator[lhs=Attribute[channel=0], rhs=Attribute[channel=1]]",
+                DataTypes.DATETIME,
+                equalTo(asMillis(asDateTime(lhs).plus(rhs)))
+            );
+        })));
+    }
+
+    @Override
+    protected boolean supportsTypes(DataType lhsType, DataType rhsType) {
+        if (isDateTimeOrTemporal(lhsType) || isDateTimeOrTemporal(rhsType)) {
+            return isDateTime(lhsType) && isTemporalAmount(rhsType) || isTemporalAmount(lhsType) && isDateTime(rhsType);
+        }
+        return super.supportsTypes(lhsType, rhsType);
     }
 
     @Override
@@ -102,4 +154,9 @@ public class AddTests extends AbstractArithmeticTestCase {
         BigInteger rhsBI = unsignedLongAsBigInteger(rhs);
         return asLongUnsigned(lhsBI.add(rhsBI).longValue());
     }
+
+    @Override
+    protected long expectedValue(long datetime, TemporalAmount temporalAmount) {
+        return asMillis(asDateTime(datetime).plus(temporalAmount));
+    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/NegTests.java

@@ -11,11 +11,17 @@ import com.carrotsearch.randomizedtesting.annotations.Name;
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
 import org.elasticsearch.xpack.esql.expression.function.scalar.AbstractScalarFunctionTestCase;
+import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.expression.Literal;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.DataTypes;
 
+import java.time.Duration;
+import java.time.Period;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.function.Supplier;
 
@@ -56,6 +62,22 @@ public class NegTests extends AbstractScalarFunctionTestCase {
                 DataTypes.DOUBLE,
                 equalTo(-arg)
             );
+        }), new TestCaseSupplier("Duration", () -> {
+            Duration arg = (Duration) randomLiteral(EsqlDataTypes.TIME_DURATION).value();
+            return new TestCase(
+                List.of(new TypedData(arg, EsqlDataTypes.TIME_DURATION, "arg")),
+                "NegDurationEvaluator[v=Attribute[channel=0]]",
+                EsqlDataTypes.TIME_DURATION,
+                equalTo(arg.negated())
+            );
+        }), new TestCaseSupplier("Period", () -> {
+            Period arg = (Period) randomLiteral(EsqlDataTypes.DATE_PERIOD).value();
+            return new TestCase(
+                List.of(new TypedData(arg, EsqlDataTypes.DATE_PERIOD, "arg")),
+                "NegPeriodEvaluator[v=Attribute[channel=0]]",
+                EsqlDataTypes.DATE_PERIOD,
+                equalTo(arg.negated())
+            );
         })));
     }
 
@@ -67,7 +89,10 @@ public class NegTests extends AbstractScalarFunctionTestCase {
     @Override
     protected List<ArgumentSpec> argSpec() {
         // More precisely: numerics without unsigned longs; however, `Neg::resolveType` uses `numeric`.
-        return List.of(required(numerics()));
+        List<DataType> types = new ArrayList<>(Arrays.asList(numerics()));
+        types.add(EsqlDataTypes.DATE_PERIOD);
+        types.add(EsqlDataTypes.TIME_DURATION);
+        return List.of(required(types.toArray(DataType[]::new)));
     }
 
     @Override
@@ -112,14 +137,47 @@ public class NegTests extends AbstractScalarFunctionTestCase {
 
             return;
         }
+        if (testCaseType == EsqlDataTypes.DATE_PERIOD) {
+            Period minPeriod = Period.of(Integer.MIN_VALUE, Integer.MIN_VALUE, Integer.MIN_VALUE);
+            assertNull(process(minPeriod));
+            assertCriticalWarnings(
+                "Line -1:-1: evaluation of [] failed, treating result as null. Only first 20 failures recorded.",
+                "java.lang.ArithmeticException: integer overflow"
+            );
+
+            Period maxPeriod = Period.of(Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE);
+            Period negatedMaxPeriod = Period.of(-Integer.MAX_VALUE, -Integer.MAX_VALUE, -Integer.MAX_VALUE);
+            assertEquals(negatedMaxPeriod, process(maxPeriod));
+            return;
+        }
+        if (testCaseType == EsqlDataTypes.TIME_DURATION) {
+            Duration minDuration = Duration.ofSeconds(Long.MIN_VALUE, 0);
+            assertNull(process(minDuration));
+            assertCriticalWarnings(
+                "Line -1:-1: evaluation of [] failed, treating result as null. Only first 20 failures recorded.",
+                "java.lang.ArithmeticException: Exceeds capacity of Duration: 9223372036854775808000000000"
+            );
+
+            Duration maxDuration = Duration.ofSeconds(Long.MAX_VALUE, 0);
+            Duration negatedMaxDuration = Duration.ofSeconds(-Long.MAX_VALUE, 0);
+            assertEquals(negatedMaxDuration, process(maxDuration));
+
+            return;
+        }
         throw new AssertionError("Edge cases not tested for negation with type [" + testCaseType.typeName() + "]");
     }
 
-    private Object process(Number val) {
-        return toJavaObject(evaluator(new Neg(Source.EMPTY, field("val", typeOf(val)))).get().eval(row(List.of(val))), 0);
+    private Object process(Object val) {
+        if (testCase.allTypesAreRepresentable()) {
+            Neg neg = new Neg(Source.EMPTY, field("val", typeOf(val)));
+            return toJavaObject(evaluator(neg).get().eval(row(List.of(val))), 0);
+        } else { // just fold if type is not representable
+            Neg neg = new Neg(Source.EMPTY, new Literal(Source.EMPTY, val, typeOf(val)));
+            return neg.fold();
+        }
     }
 
-    private DataType typeOf(Number val) {
+    private static DataType typeOf(Object val) {
         if (val instanceof Integer) {
             return DataTypes.INTEGER;
         }
@@ -129,6 +187,12 @@ public class NegTests extends AbstractScalarFunctionTestCase {
         if (val instanceof Double) {
             return DataTypes.DOUBLE;
         }
+        if (val instanceof Duration) {
+            return EsqlDataTypes.TIME_DURATION;
+        }
+        if (val instanceof Period) {
+            return EsqlDataTypes.DATE_PERIOD;
+        }
         throw new UnsupportedOperationException("unsupported type [" + val.getClass() + "]");
     }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/predicate/operator/arithmetic/SubTests.java

@@ -10,19 +10,29 @@ package org.elasticsearch.xpack.esql.expression.predicate.operator.arithmetic;
 import com.carrotsearch.randomizedtesting.annotations.Name;
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
+import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 import org.elasticsearch.xpack.ql.expression.Expression;
 import org.elasticsearch.xpack.ql.tree.Source;
+import org.elasticsearch.xpack.ql.type.DataType;
 import org.elasticsearch.xpack.ql.type.DataTypes;
 
 import java.math.BigInteger;
+import java.time.Duration;
+import java.time.Period;
+import java.time.temporal.TemporalAmount;
 import java.util.List;
 import java.util.function.Supplier;
 
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isDateTimeOrTemporal;
+import static org.elasticsearch.xpack.esql.type.EsqlDataTypes.isTemporalAmount;
+import static org.elasticsearch.xpack.ql.type.DataTypes.isDateTime;
+import static org.elasticsearch.xpack.ql.type.DateUtils.asDateTime;
+import static org.elasticsearch.xpack.ql.type.DateUtils.asMillis;
 import static org.elasticsearch.xpack.ql.util.NumericUtils.asLongUnsigned;
 import static org.elasticsearch.xpack.ql.util.NumericUtils.unsignedLongAsBigInteger;
 import static org.hamcrest.Matchers.equalTo;
 
-public class SubTests extends AbstractArithmeticTestCase {
+public class SubTests extends AbstractDateTimeArithmeticTestCase {
     public SubTests(@Name("TestCase") Supplier<TestCase> testCaseSupplier) {
         this.testCase = testCaseSupplier.get();
     }
@@ -71,9 +81,32 @@ public class SubTests extends AbstractArithmeticTestCase {
                 "SubUnsignedLongsEvaluator[lhs=Attribute[channel=0], rhs=Attribute[channel=1]]",
                 equalTo(asLongUnsigned(lhsBI.subtract(rhsBI).longValue()))
             );
-          })
-          */
-        ));
+          }) */, new TestCaseSupplier("Datetime - Period", () -> {
+            long lhs = (Long) randomLiteral(DataTypes.DATETIME).value();
+            Period rhs = (Period) randomLiteral(EsqlDataTypes.DATE_PERIOD).value();
+            return new TestCase(
+                List.of(new TypedData(lhs, DataTypes.DATETIME, "lhs"), new TypedData(rhs, EsqlDataTypes.DATE_PERIOD, "rhs")),
+                "SubDatetimesEvaluator[lhs=Attribute[channel=0], rhs=Attribute[channel=1]]",
+                DataTypes.DATETIME,
+                equalTo(asMillis(asDateTime(lhs).minus(rhs)))
+            );
+        }), new TestCaseSupplier("Datetime - Duration", () -> {
+            long lhs = (Long) randomLiteral(DataTypes.DATETIME).value();
+            Duration rhs = (Duration) randomLiteral(EsqlDataTypes.TIME_DURATION).value();
+            return new TestCase(
+                List.of(new TypedData(lhs, DataTypes.DATETIME, "lhs"), new TypedData(rhs, EsqlDataTypes.TIME_DURATION, "rhs")),
+                "SubDatetimesEvaluator[lhs=Attribute[channel=0], rhs=Attribute[channel=1]]",
+                DataTypes.DATETIME,
+                equalTo(asMillis(asDateTime(lhs).minus(rhs)))
+            );
+        })));
+    }
+
+    @Override
+    protected boolean supportsTypes(DataType lhsType, DataType rhsType) {
+        return isDateTimeOrTemporal(lhsType) || isDateTimeOrTemporal(rhsType)
+            ? isDateTime(lhsType) && isTemporalAmount(rhsType)
+            : super.supportsTypes(lhsType, rhsType);
     }
 
     @Override
@@ -102,4 +135,9 @@ public class SubTests extends AbstractArithmeticTestCase {
         BigInteger rhsBI = unsignedLongAsBigInteger(rhs);
         return asLongUnsigned(lhsBI.subtract(rhsBI).longValue());
     }
+
+    @Override
+    protected long expectedValue(long datetime, TemporalAmount temporalAmount) {
+        return asMillis(asDateTime(datetime).minus(temporalAmount));
+    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/predicate/operator/comparison/AbstractBinaryComparisonTestCase.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.esql.expression.predicate.operator.comparison;
 
 import org.elasticsearch.xpack.esql.analysis.Verifier;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.AbstractBinaryOperatorTestCase;
+import org.elasticsearch.xpack.esql.type.EsqlDataTypes;
 import org.elasticsearch.xpack.ql.common.Failure;
 import org.elasticsearch.xpack.ql.expression.predicate.BinaryOperator;
 import org.elasticsearch.xpack.ql.expression.predicate.operator.comparison.BinaryComparison;
@@ -69,7 +70,12 @@ public abstract class AbstractBinaryComparisonTestCase extends AbstractBinaryOpe
         if (type == DataTypes.BOOLEAN) {
             return isEquality();
         }
-        return true;
+        return EsqlDataTypes.isRepresentable(type);
+    }
+
+    @Override
+    protected boolean supportsTypes(DataType lhsType, DataType rhsType) {
+        return super.supportsTypes(lhsType, rhsType) && (lhsType == rhsType || lhsType.isNumeric() && rhsType.isNumeric());
     }
 
     @Override

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/type/DateUtils.java

@@ -52,6 +52,10 @@ public final class DateUtils {
         return ZonedDateTime.ofInstant(Instant.ofEpochMilli(millis), UTC);
     }
 
+    public static long asMillis(ZonedDateTime zonedDateTime) {
+        return zonedDateTime.toInstant().toEpochMilli();
+    }
+
     /**
      * Parses the given string into a DateTime using UTC as a default timezone.
      */