Add support to first/last aggregators for numeric types during ingestion

docs/querying/aggregations.md

@@ -134,8 +134,6 @@ Computes and returns arithmetic mean of a column values as 64 bit float value. T
 
 ### First / Last aggregator
 
-(Double/Float/Long) First and Last aggregator cannot be used in ingestion spec, and should only be specified as part of queries.
-
 Note that queries with first/last aggregators on a segment created with rollup enabled will return the rolled up value, and not the last value within the raw ingested data.
 
 #### `doubleFirst` aggregator

integration-tests/src/test/resources/indexer/wikipedia_merge_index_queries.json

@@ -1,6 +1,6 @@
 [
     {
-        "description": "groupby, stringFirst/stringLast rollup aggs, all",
+        "description": "groupby, stringFirst/stringLast/doubleFirst/doubleLast/longFirst/longLast/floatFirst/floatLast rollup aggs, all",
         "query":{
             "queryType" : "groupBy",
             "dataSource": "%%DATASOURCE%%",
@@ -26,6 +26,36 @@
                     "type":"stringLast",
                     "name":"latest_user",
                     "fieldName":"last_user"
+                },
+                {
+                    "type": "doubleFirst",
+                    "name": "double_first_delta",
+                    "fieldName": "double_first_delta"
+                },
+                {
+                    "type": "doubleLast",
+                    "name": "double_last_delta",
+                    "fieldName": "double_last_delta"
+                },
+                {
+                    "type": "longFirst",
+                    "name": "long_first_delta",
+                    "fieldName": "long_first_delta"
+                },
+                {
+                    "type": "longFirst",
+                    "name": "long_last_delta",
+                    "fieldName": "long_last_delta"
+                },
+                {
+                    "type": "floatFirst",
+                    "name": "float_first_delta",
+                    "fieldName": "float_first_delta"
+                },
+                {
+                    "type": "floatLast",
+                    "name": "float_last_delta",
+                    "fieldName": "float_last_delta"
                 }
             ]
         },
@@ -35,7 +65,13 @@
             "event" : {
                 "continent":"Asia",
                 "earliest_user":"masterYi",
-                "latest_user":"stringer"
+                "latest_user":"stringer",
+                "double_first_delta": 111.0,
+                "double_last_delta": -9.0,
+                "long_first_delta": 111,
+                "long_last_delta": -9,
+                "float_first_delta": 111.0,
+                "float_last_delta": -9.0
             }
         } ]
     }

integration-tests/src/test/resources/indexer/wikipedia_merge_index_task.json

@@ -23,6 +23,36 @@
                     "name": "delta",
                     "fieldName": "delta"
                 },
+                {
+                    "type": "doubleFirst",
+                    "name": "double_first_delta",
+                    "fieldName": "delta"
+                },
+                {
+                    "type": "doubleLast",
+                    "name": "double_last_delta",
+                    "fieldName": "delta"
+                },
+                {
+                    "type": "longFirst",
+                    "name": "long_first_delta",
+                    "fieldName": "delta"
+                },
+                {
+                    "type": "longLast",
+                    "name": "long_last_delta",
+                    "fieldName": "delta"
+                },
+                {
+                    "type": "floatFirst",
+                    "name": "float_first_delta",
+                    "fieldName": "delta"
+                },
+                {
+                    "type": "floatLast",
+                    "name": "float_last_delta",
+                    "fieldName": "delta"
+                },
                 {
                     "type": "stringFirst",
                     "name": "first_user",

integration-tests/src/test/resources/indexer/wikipedia_merge_reindex_druid_input_source_task.json

@@ -56,6 +56,36 @@
                     "type": "stringLast",
                     "name": "last_user",
                     "fieldName": "last_user"
+                },
+                {
+                    "type": "doubleFirst",
+                    "name": "double_first_delta",
+                    "fieldName": "double_first_delta"
+                },
+                {
+                    "type": "doubleLast",
+                    "name": "double_last_delta",
+                    "fieldName": "double_last_delta"
+                },
+                {
+                    "type": "longFirst",
+                    "name": "long_first_delta",
+                    "fieldName": "long_first_delta"
+                },
+                {
+                    "type": "longLast",
+                    "name": "long_last_delta",
+                    "fieldName": "long_last_delta"
+                },
+                {
+                    "type": "floatFirst",
+                    "name": "float_first_delta",
+                    "fieldName": "float_first_delta"
+                },
+                {
+                    "type": "floatLast",
+                    "name": "float_last_delta",
+                    "fieldName": "float_last_delta"
                 }
             ]
         }

integration-tests/src/test/resources/indexer/wikipedia_merge_reindex_task.json

@@ -19,6 +19,36 @@
                     "name": "delta",
                     "fieldName": "delta"
                 },
+                {
+                    "type": "doubleFirst",
+                    "name": "double_first_delta",
+                    "fieldName": "double_first_delta"
+                },
+                {
+                    "type": "doubleLast",
+                    "name": "double_last_delta",
+                    "fieldName": "double_last_delta"
+                },
+                {
+                    "type": "longFirst",
+                    "name": "long_first_delta",
+                    "fieldName": "long_first_delta"
+                },
+                {
+                    "type": "longLast",
+                    "name": "long_last_delta",
+                    "fieldName": "long_last_delta"
+                },
+                {
+                    "type": "floatFirst",
+                    "name": "float_first_delta",
+                    "fieldName": "float_first_delta"
+                },
+                {
+                    "type": "floatLast",
+                    "name": "float_last_delta",
+                    "fieldName": "float_last_delta"
+                },
                 {
                     "type": "stringFirst",
                     "name": "first_user",
@@ -62,4 +92,4 @@
             "type": "index"
         }
     }
-}
+}

processing/src/main/java/org/apache/druid/jackson/AggregatorsModule.java

@@ -38,6 +38,9 @@
 import org.apache.druid.query.aggregation.LongMinAggregatorFactory;
 import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
 import org.apache.druid.query.aggregation.PostAggregator;
+import org.apache.druid.query.aggregation.SerializablePairLongDoubleSerde;
+import org.apache.druid.query.aggregation.SerializablePairLongFloatSerde;
+import org.apache.druid.query.aggregation.SerializablePairLongLongSerde;
 import org.apache.druid.query.aggregation.SerializablePairLongStringSerde;
 import org.apache.druid.query.aggregation.any.DoubleAnyAggregatorFactory;
 import org.apache.druid.query.aggregation.any.FloatAnyAggregatorFactory;
@@ -80,7 +83,10 @@ public AggregatorsModule()
 
     ComplexMetrics.registerSerde("hyperUnique", new HyperUniquesSerde());
     ComplexMetrics.registerSerde("preComputedHyperUnique", new PreComputedHyperUniquesSerde());
-    ComplexMetrics.registerSerde("serializablePairLongString", new SerializablePairLongStringSerde());
+    ComplexMetrics.registerSerde(new SerializablePairLongStringSerde());
+    ComplexMetrics.registerSerde(new SerializablePairLongDoubleSerde());
+    ComplexMetrics.registerSerde(new SerializablePairLongFloatSerde());
+    ComplexMetrics.registerSerde(new SerializablePairLongLongSerde());
 
     setMixInAnnotation(AggregatorFactory.class, AggregatorFactoryMixin.class);
     setMixInAnnotation(PostAggregator.class, PostAggregatorMixin.class);

processing/src/main/java/org/apache/druid/query/aggregation/AbstractSerializableLongObjectPairSerde.java

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query.aggregation;
+
+import org.apache.druid.collections.SerializablePair;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.segment.GenericColumnSerializer;
+import org.apache.druid.segment.column.ColumnBuilder;
+import org.apache.druid.segment.data.GenericIndexed;
+import org.apache.druid.segment.serde.ComplexColumnPartSupplier;
+import org.apache.druid.segment.serde.ComplexMetricExtractor;
+import org.apache.druid.segment.serde.ComplexMetricSerde;
+import org.apache.druid.segment.serde.LargeColumnSupportedComplexColumnSerializer;
+import org.apache.druid.segment.writeout.SegmentWriteOutMedium;
+
+import java.nio.ByteBuffer;
+
+/**
+ * The class serializes/deserializes a Pair<Long, ?> object for double/float/longFirst and double/float/longLast aggregators
+ */
+public abstract class AbstractSerializableLongObjectPairSerde<T extends SerializablePair<Long, ?>>
+    extends ComplexMetricSerde
+{
+  private final Class<T> pairClassObject;
+
+  public AbstractSerializableLongObjectPairSerde(Class<T> pairClassObject)
+  {
+    this.pairClassObject = pairClassObject;
+  }
+
+  @Override
+  public ComplexMetricExtractor getExtractor()
+  {
+    return new ComplexMetricExtractor()
+    {
+      @Override
+      public Class<T> extractedClass()
+      {
+        return pairClassObject;
+      }
+
+      @Override
+      public Object extractValue(InputRow inputRow, String metricName)
+      {
+        return inputRow.getRaw(metricName);
+      }
+    };
+  }
+
+  @Override
+  public void deserializeColumn(ByteBuffer buffer, ColumnBuilder columnBuilder)
+  {
+    final GenericIndexed column = GenericIndexed.read(buffer, getObjectStrategy(), columnBuilder.getFileMapper());
+    columnBuilder.setComplexColumnSupplier(new ComplexColumnPartSupplier(getTypeName(), column));
+  }
+
+  @Override
+  public GenericColumnSerializer<T> getSerializer(SegmentWriteOutMedium segmentWriteOutMedium, String column)
+  {
+    return LargeColumnSupportedComplexColumnSerializer.create(segmentWriteOutMedium, column, this.getObjectStrategy());
+  }
+}

processing/src/main/java/org/apache/druid/query/aggregation/SerializablePairLongDouble.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query.aggregation;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.collections.SerializablePair;
+
+import javax.annotation.Nullable;
+
+public class SerializablePairLongDouble extends SerializablePair<Long, Double>
+{
+  @JsonCreator
+  public SerializablePairLongDouble(@JsonProperty("lhs") Long lhs, @JsonProperty("rhs") @Nullable Double rhs)
+  {
+    super(lhs, rhs);
+  }
+}
+
+