ES|QL: Add unsigned long and mixed numeric types support to Decay function

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

@@ -623,6 +623,10 @@ public static boolean isSortable(DataType t) {
         return false == (t == SOURCE || isCounter(t) || isSpatialOrGrid(t) || t == AGGREGATE_METRIC_DOUBLE);
     }
 
+    public static boolean isUnsignedLong(DataType t) {
+        return t == UNSIGNED_LONG;
+    }
+
     public String nameUpper() {
         return name;
     }

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

@@ -24,6 +24,28 @@ decay_result:double
 0.75
 ;
 
+intLinearMixedNumericTypes1
+required_capability: decay_function
+
+ROW value = 5
+| EVAL decay_result = decay(value, 10::double, 10::long, {"offset": 0, "decay": 0.5, "type": "linear"}) 
+| KEEP decay_result;
+
+decay_result:double
+0.75
+;
+
+intLinearMixedNumericTypes2
+required_capability: decay_function
+
+ROW value = 5
+| EVAL decay_result = decay(value, 10::unsigned_long, 10::long, {"offset": 0, "decay": 0.5, "type": "linear"}) 
+| KEEP decay_result;
+
+decay_result:double
+0.75
+;
+
 intExp
 required_capability: decay_function
 
@@ -162,6 +184,28 @@ decay_result:double
 0.75
 ;
 
+doubleLinearMixedNumericTypes1
+required_capability: decay_function
+
+ROW value = 5.0
+| EVAL decay_result = decay(value, 10.0::int, 10.0::long, {"offset": 0.0, "decay": 0.5, "type": "linear"}) 
+| KEEP decay_result;
+
+decay_result:double
+0.75
+;
+
+doubleLinearMixedNumericTypes2
+required_capability: decay_function
+
+ROW value = 5.0
+| EVAL decay_result = decay(value, 10.0::int, 10.0::unsigned_long, {"offset": 0.0, "decay": 0.5, "type": "linear"}) 
+| KEEP decay_result;
+
+decay_result:double
+0.75
+;
+
 doubleExp
 required_capability: decay_function
 
@@ -195,6 +239,61 @@ decay_result:double
 1.0
 ;
 
+longLinearMixedNumericTypes1
+required_capability: decay_function
+
+ROW value = 15::long
+| EVAL decay_result = decay(value, 10::int, 10::unsigned_long, {"offset": 10000000000, "decay": 0.5, "type": "linear"}) 
+| KEEP decay_result;
+
+decay_result:double
+1.0
+;
+
+longLinearMixedNumericTypes2
+required_capability: decay_function
+
+ROW value = 15::long
+| EVAL decay_result = decay(value, 10::double, 10::int, {"offset": 10000000000, "decay": 0.5, "type": "linear"}) 
+| KEEP decay_result;
+
+decay_result:double
+1.0
+;
+
+unsignedLongLinear
+required_capability: decay_function
+
+ROW value = 15::unsigned_long
+| EVAL decay_result = decay(value, 10::unsigned_long, 10::unsigned_long, {"offset": 18446744073709551615, "decay": 0.5, "type": "linear"}) 
+| KEEP decay_result;
+
+decay_result:double
+1.0
+;
+
+unsignedLongLinearMixedNumericTypes1
+required_capability: decay_function
+
+ROW value = 15::unsigned_long
+| EVAL decay_result = decay(value, 10::long, 10::integer, {"offset": 18446744073709551615, "decay": 0.5, "type": "linear"}) 
+| KEEP decay_result;
+
+decay_result:double
+1.0
+;
+
+unsignedLongLinearMixedNumericTypes2
+required_capability: decay_function
+
+ROW value = 15::unsigned_long
+| EVAL decay_result = decay(value, 10::double, 10::integer, {"offset": 18446744073709551615, "decay": 0.5, "type": "linear"}) 
+| KEEP decay_result;
+
+decay_result:double
+1.0
+;
+
 cartesianPointLinear1
 required_capability: decay_function
 

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

@@ -27,6 +27,7 @@
 import org.elasticsearch.xpack.esql.core.tree.NodeInfo;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.core.util.NumericUtils;
 import org.elasticsearch.xpack.esql.core.util.SpatialCoordinateTypes;
 import org.elasticsearch.xpack.esql.expression.function.Example;
 import org.elasticsearch.xpack.esql.expression.function.FunctionAppliesTo;
@@ -38,8 +39,10 @@
 import org.elasticsearch.xpack.esql.expression.function.Param;
 import org.elasticsearch.xpack.esql.expression.function.scalar.EsqlScalarFunction;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
+import org.elasticsearch.xpack.esql.type.EsqlDataTypeConverter;
 
 import java.io.IOException;
+import java.math.BigInteger;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
@@ -63,6 +66,7 @@
 import static org.elasticsearch.xpack.esql.core.type.DataType.LONG;
 import static org.elasticsearch.xpack.esql.core.type.DataType.TEXT;
 import static org.elasticsearch.xpack.esql.core.type.DataType.TIME_DURATION;
+import static org.elasticsearch.xpack.esql.core.type.DataType.UNSIGNED_LONG;
 import static org.elasticsearch.xpack.esql.core.type.DataType.isDateNanos;
 import static org.elasticsearch.xpack.esql.core.type.DataType.isGeoPoint;
 import static org.elasticsearch.xpack.esql.core.type.DataType.isMillisOrNanos;
@@ -91,7 +95,7 @@ public class Decay extends EsqlScalarFunction implements OptionalArgument, PostO
 
     private static final Map<String, Collection<DataType>> ALLOWED_OPTIONS = Map.of(
         OFFSET,
-        Set.of(TIME_DURATION, INTEGER, LONG, DOUBLE, KEYWORD, TEXT),
+        Set.of(TIME_DURATION, INTEGER, LONG, UNSIGNED_LONG, DOUBLE, KEYWORD, TEXT),
         DECAY,
         Set.of(DOUBLE),
         TYPE,
@@ -140,25 +144,25 @@ public Decay(
         Source source,
         @Param(
             name = "value",
-            type = { "double", "integer", "long", "date", "date_nanos", "geo_point", "cartesian_point" },
+            type = { "double", "integer", "long", "unsigned_long", "date", "date_nanos", "geo_point", "cartesian_point" },
             description = "The input value to apply decay scoring to."
         ) Expression value,
         @Param(
             name = ORIGIN,
-            type = { "double", "integer", "long", "date", "date_nanos", "geo_point", "cartesian_point" },
+            type = { "double", "integer", "long", "unsigned_long", "date", "date_nanos", "geo_point", "cartesian_point" },
             description = "Central point from which the distances are calculated."
         ) Expression origin,
         @Param(
             name = SCALE,
-            type = { "double", "integer", "long", "time_duration", "keyword", "text" },
+            type = { "double", "integer", "long", "unsigned_long", "time_duration", "keyword", "text" },
             description = "Distance from the origin where the function returns the decay value."
         ) Expression scale,
         @MapParam(
             name = "options",
             params = {
                 @MapParam.MapParamEntry(
                     name = OFFSET,
-                    type = { "double", "integer", "long", "time_duration", "keyword", "text" },
+                    type = { "double", "integer", "long", "unsigned_long", "time_duration", "keyword", "text" },
                     description = "Distance from the origin where no decay occurs."
                 ),
                 @MapParam.MapParamEntry(
@@ -285,8 +289,8 @@ public EvalOperator.ExpressionEvaluator.Factory toEvaluator(ToEvaluator toEvalua
         FoldContext foldCtx = toEvaluator.foldCtx();
 
         // Constants
-        Object originFolded = origin.fold(foldCtx);
-        Object scaleFolded = getFoldedScale(foldCtx, valueDataType);
+        Object originFolded = convertToExpectedType(origin.fold(foldCtx), origin.dataType(), valueDataType);
+        Object scaleFolded = convertToExpectedType(getFoldedScale(foldCtx, valueDataType), scale.dataType(), valueDataType);
         Object offsetFolded = getOffset(foldCtx, valueDataType, offsetExpr);
         Double decayFolded = decayExpr != null ? (Double) decayExpr.fold(foldCtx) : DEFAULT_DECAY;
         DecayFunction decayFunction = DecayFunction.fromBytesRef(typeExpr != null ? (BytesRef) typeExpr.fold(foldCtx) : DEFAULT_FUNCTION);
@@ -319,6 +323,15 @@ public EvalOperator.ExpressionEvaluator.Factory toEvaluator(ToEvaluator toEvalua
                 decayFolded,
                 decayFunction
             );
+            case UNSIGNED_LONG -> new DecayUnsignedLongEvaluator.Factory(
+                source(),
+                valueFactory,
+                (Long) originFolded,
+                (Long) scaleFolded,
+                (Long) offsetFolded,
+                decayFolded,
+                decayFunction
+            );
             case GEO_POINT -> new DecayGeoPointEvaluator.Factory(
                 source(),
                 valueFactory,
@@ -403,7 +416,24 @@ static double process(
         @Fixed DecayFunction decayFunction
     ) {
         return decayFunction.numericDecay(value, origin, scale, offset, decay);
+    }
 
+    @Evaluator(extraName = "UnsignedLong")
+    static double processUnsignedLong(
+        long value,
+        @Fixed long origin,
+        @Fixed long scale,
+        @Fixed long offset,
+        @Fixed double decay,
+        @Fixed DecayFunction decayFunction
+    ) {
+        return decayFunction.numericDecay(
+            NumericUtils.unsignedLongToDouble(value),
+            NumericUtils.unsignedLongToDouble(origin),
+            NumericUtils.unsignedLongToDouble(scale),
+            NumericUtils.unsignedLongToDouble(offset),
+            decay
+        );
     }
 
     @Evaluator(extraName = "GeoPoint")
@@ -634,7 +664,7 @@ private Long getTemporalOffsetAsNanos(FoldContext foldCtx, Expression offset) {
     private Object getDefaultOffset(DataType valueDataType) {
         return switch (valueDataType) {
             case INTEGER -> DEFAULT_INTEGER_OFFSET;
-            case LONG -> DEFAULT_LONG_OFFSET;
+            case LONG, UNSIGNED_LONG -> DEFAULT_LONG_OFFSET;
             case DOUBLE -> DEFAULT_DOUBLE_OFFSET;
             case GEO_POINT -> DEFAULT_GEO_POINT_OFFSET;
             case CARTESIAN_POINT -> DEFAULT_CARTESIAN_POINT_OFFSET;
@@ -643,4 +673,25 @@ private Object getDefaultOffset(DataType valueDataType) {
         };
     }
 
+    private Object convertToExpectedType(Object value, DataType valueType, DataType targetType) {
+        // No conversion needed
+        if (targetType.isNumeric() == false) {
+            return value;
+        }
+
+        // Conversion needed as unsigned longs are represented as signed longs
+        if (valueType == UNSIGNED_LONG) {
+            value = NumericUtils.unsignedLongToDouble(((Number) value).longValue());
+        }
+
+        Object convertedValue = EsqlDataTypeConverter.convert(value, targetType);
+
+        // Unsigned long evaluator expects unsigned longs in signed long representation
+        if (convertedValue instanceof BigInteger valueAsBigInteger) {
+            return NumericUtils.asLongUnsigned(valueAsBigInteger);
+        }
+
+        return convertedValue;
+    }
+
 }