Add experimental support for first/last for double/float/long #10702

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

@@ -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

@@ -40,6 +40,36 @@
                     "type": "stringLast",
                     "name": "last_user",
                     "fieldName": "user"
+                },
+                {
+                    "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"
                 }
             ],
             "granularitySpec": {

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

@@ -37,6 +37,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"
                 }
             ],
             "granularitySpec": {

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

@@ -39,6 +39,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.SerializablePairLongDoubleComplexMetricSerde;
+import org.apache.druid.query.aggregation.SerializablePairLongFloatComplexMetricSerde;
+import org.apache.druid.query.aggregation.SerializablePairLongLongComplexMetricSerde;
 import org.apache.druid.query.aggregation.SerializablePairLongStringComplexMetricSerde;
 import org.apache.druid.query.aggregation.any.DoubleAnyAggregatorFactory;
 import org.apache.druid.query.aggregation.any.FloatAnyAggregatorFactory;
@@ -83,6 +86,10 @@ public AggregatorsModule()
     ComplexMetrics.registerSerde(PreComputedHyperUniquesSerde.TYPE_NAME, new PreComputedHyperUniquesSerde());
     ComplexMetrics.registerSerde(SerializablePairLongStringComplexMetricSerde.TYPE_NAME, new SerializablePairLongStringComplexMetricSerde());
 
+    ComplexMetrics.registerSerde(SerializablePairLongFloatComplexMetricSerde.TYPE_NAME, new SerializablePairLongFloatComplexMetricSerde());
+    ComplexMetrics.registerSerde(SerializablePairLongDoubleComplexMetricSerde.TYPE_NAME, new SerializablePairLongDoubleComplexMetricSerde());
+    ComplexMetrics.registerSerde(SerializablePairLongLongComplexMetricSerde.TYPE_NAME, new SerializablePairLongLongComplexMetricSerde());
+
     setMixInAnnotation(AggregatorFactory.class, AggregatorFactoryMixin.class);
     setMixInAnnotation(PostAggregator.class, PostAggregatorMixin.class);
 

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

@@ -0,0 +1,82 @@
+/*
+ * 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 javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+
+public abstract class AbstractSerializableLongObjectPairSerde<T extends SerializablePair<Long, ?>> extends
+    ComplexMetricSerde
+{
+  private final Class<T> pairClass;
+
+  public AbstractSerializableLongObjectPairSerde(Class<T> pairClass)
+  {
+    this.pairClass = pairClass;
+  }
+
+  @Override
+  public ComplexMetricExtractor<?> getExtractor()
+  {
+    return new ComplexMetricExtractor<Object>()
+    {
+      @Override
+      public Class<T> extractedClass()
+      {
+        return pairClass;
+      }
+
+      @Nullable
+      @Override
+      public Object extractValue(InputRow inputRow, String metricName)
+      {
+        return inputRow.getMetric(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,
+        getObjectStrategy()
+    );
+  }
+}

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

@@ -0,0 +1,35 @@
+/*
+ * 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);
+  }
+}