From eacd5b9edcd4204aab4a25ad33d436b7b6d366e1 Mon Sep 17 00:00:00 2001 From: Clint Wylie Date: Mon, 20 Jan 2020 11:51:54 -0800 Subject: [PATCH] first/last aggregators and nulls (#9161) * null handling for numeric first/last aggregators, refactor to not extend nullable numeric agg since they are complex typed aggs * initially null or not based on config * review stuff, make string first/last consistent with null handling of numeric columns, more tests * docs * handle nil selectors, revert to primitive first/last types so groupby v1 works... --- .../druid/collections/SerializablePair.java | 22 + docs/querying/aggregations.md | 12 +- docs/querying/sql.md | 4 +- .../first/DoubleFirstAggregator.java | 31 +- .../first/DoubleFirstAggregatorFactory.java | 107 +++-- .../first/DoubleFirstBufferAggregator.java | 46 +- .../first/FloatFirstAggregator.java | 33 +- .../first/FloatFirstAggregatorFactory.java | 103 +++-- .../first/FloatFirstBufferAggregator.java | 46 +- .../first/LongFirstAggregator.java | 31 +- .../first/LongFirstAggregatorFactory.java | 102 +++-- .../first/LongFirstBufferAggregator.java | 46 +- .../first/NumericFirstAggregator.java | 74 ++++ .../first/NumericFirstBufferAggregator.java | 112 +++++ .../first/StringFirstAggregator.java | 9 +- .../first/StringFirstAggregatorFactory.java | 62 ++- .../first/StringFirstBufferAggregator.java | 16 +- .../last/DoubleLastAggregator.java | 31 +- .../last/DoubleLastAggregatorFactory.java | 107 +++-- .../last/DoubleLastBufferAggregator.java | 46 +- .../aggregation/last/FloatLastAggregator.java | 33 +- .../last/FloatLastAggregatorFactory.java | 99 ++++- .../last/FloatLastBufferAggregator.java | 51 +-- .../aggregation/last/LongLastAggregator.java | 38 +- .../last/LongLastAggregatorFactory.java | 97 ++++- .../last/LongLastBufferAggregator.java | 46 +- .../last/NumericLastAggregator.java | 75 ++++ .../last/NumericLastBufferAggregator.java | 115 +++++ .../last/StringLastAggregator.java | 9 +- .../last/StringLastAggregatorFactory.java | 62 ++- .../last/StringLastBufferAggregator.java | 16 +- .../first/DoubleFirstAggregationTest.java | 29 +- .../first/FloatFirstAggregationTest.java | 24 +- .../first/LongFirstAggregationTest.java | 16 +- .../StringFirstBufferAggregatorTest.java | 4 +- .../first/StringFirstTimeseriesQueryTest.java | 3 +- .../last/DoubleLastAggregationTest.java | 16 +- .../last/FloatLastAggregationTest.java | 20 +- .../last/LongLastAggregationTest.java | 16 +- .../druid/sql/calcite/CalciteQueryTest.java | 399 +++++++++++++++++- 40 files changed, 1617 insertions(+), 591 deletions(-) create mode 100644 processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstAggregator.java create mode 100644 processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstBufferAggregator.java create mode 100644 processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastAggregator.java create mode 100644 processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastBufferAggregator.java diff --git a/core/src/main/java/org/apache/druid/collections/SerializablePair.java b/core/src/main/java/org/apache/druid/collections/SerializablePair.java index 46b399551173..22847cbac3a8 100644 --- a/core/src/main/java/org/apache/druid/collections/SerializablePair.java +++ b/core/src/main/java/org/apache/druid/collections/SerializablePair.java @@ -24,6 +24,7 @@ import org.apache.druid.java.util.common.Pair; import javax.annotation.Nullable; +import java.util.Comparator; public class SerializablePair extends Pair { @@ -45,4 +46,25 @@ public T2 getRhs() { return rhs; } + + public static Comparator> createNullHandlingComparator( + Comparator delegate, + boolean nullsFirst + ) + { + final int firstIsNull = nullsFirst ? -1 : 1; + final int secondIsNull = nullsFirst ? 1 : -1; + return (o1, o2) -> { + if (o1 == null || o1.rhs == null) { + if (o2 == null || o2.rhs == null) { + return 0; + } + return firstIsNull; + } + if (o2 == null || o2.rhs == null) { + return secondIsNull; + } + return delegate.compare(o1.rhs, o2.rhs); + }; + } } diff --git a/docs/querying/aggregations.md b/docs/querying/aggregations.md index c1d8d6537c91..688b9aee3813 100644 --- a/docs/querying/aggregations.md +++ b/docs/querying/aggregations.md @@ -136,7 +136,7 @@ Note that queries with first/last aggregators on a segment created with rollup e #### `doubleFirst` aggregator -`doubleFirst` computes the metric value with the minimum timestamp or 0 if no row exist +`doubleFirst` computes the metric value with the minimum timestamp or 0 in default mode or `null` in SQL compatible mode if no row exist ```json { @@ -148,7 +148,7 @@ Note that queries with first/last aggregators on a segment created with rollup e #### `doubleLast` aggregator -`doubleLast` computes the metric value with the maximum timestamp or 0 if no row exist +`doubleLast` computes the metric value with the maximum timestamp or 0 in default mode or `null` in SQL compatible mode if no row exist ```json { @@ -160,7 +160,7 @@ Note that queries with first/last aggregators on a segment created with rollup e #### `floatFirst` aggregator -`floatFirst` computes the metric value with the minimum timestamp or 0 if no row exist +`floatFirst` computes the metric value with the minimum timestamp or 0 in default mode or `null` in SQL compatible mode if no row exist ```json { @@ -172,7 +172,7 @@ Note that queries with first/last aggregators on a segment created with rollup e #### `floatLast` aggregator -`floatLast` computes the metric value with the maximum timestamp or 0 if no row exist +`floatLast` computes the metric value with the maximum timestamp or 0 in default mode or `null` in SQL compatible mode if no row exist ```json { @@ -184,7 +184,7 @@ Note that queries with first/last aggregators on a segment created with rollup e #### `longFirst` aggregator -`longFirst` computes the metric value with the minimum timestamp or 0 if no row exist +`longFirst` computes the metric value with the minimum timestamp or 0 in default mode or `null` in SQL compatible mode if no row exist ```json { @@ -196,7 +196,7 @@ Note that queries with first/last aggregators on a segment created with rollup e #### `longLast` aggregator -`longLast` computes the metric value with the maximum timestamp or 0 if no row exist +`longLast` computes the metric value with the maximum timestamp or 0 in default mode or `null` in SQL compatible mode if no row exist ```json { diff --git a/docs/querying/sql.md b/docs/querying/sql.md index 8aa1367d14cc..f29618d82513 100644 --- a/docs/querying/sql.md +++ b/docs/querying/sql.md @@ -199,9 +199,9 @@ Only the COUNT aggregation can accept DISTINCT. |`STDDEV_POP(expr)`|Computes standard deviation population of `expr`. See [stats extension](../development/extensions-core/stats.html) documentation for additional details.| |`STDDEV_SAMP(expr)`|Computes standard deviation sample of `expr`. See [stats extension](../development/extensions-core/stats.html) documentation for additional details.| |`STDDEV(expr)`|Computes standard deviation sample of `expr`. See [stats extension](../development/extensions-core/stats.html) documentation for additional details.| -|`EARLIEST(expr)`|Returns the earliest non-null value of `expr`, which must be numeric. If `expr` comes from a relation with a timestamp column (like a Druid datasource) then "earliest" is the value first encountered with the minimum overall timestamp of all values being aggregated. If `expr` does not come from a relation with a timestamp, then it is simply the first value encountered.| +|`EARLIEST(expr)`|Returns the earliest value of `expr`, which must be numeric. If `expr` comes from a relation with a timestamp column (like a Druid datasource) then "earliest" is the value first encountered with the minimum overall timestamp of all values being aggregated. If `expr` does not come from a relation with a timestamp, then it is simply the first value encountered.| |`EARLIEST(expr, maxBytesPerString)`|Like `EARLIEST(expr)`, but for strings. The `maxBytesPerString` parameter determines how much aggregation space to allocate per string. Strings longer than this limit will be truncated. This parameter should be set as low as possible, since high values will lead to wasted memory.| -|`LATEST(expr)`|Returns the latest non-null value of `expr`, which must be numeric. If `expr` comes from a relation with a timestamp column (like a Druid datasource) then "latest" is the value last encountered with the maximum overall timestamp of all values being aggregated. If `expr` does not come from a relation with a timestamp, then it is simply the last value encountered.| +|`LATEST(expr)`|Returns the latest value of `expr`, which must be numeric. If `expr` comes from a relation with a timestamp column (like a Druid datasource) then "latest" is the value last encountered with the maximum overall timestamp of all values being aggregated. If `expr` does not come from a relation with a timestamp, then it is simply the last value encountered.| |`LATEST(expr, maxBytesPerString)`|Like `LATEST(expr)`, but for strings. The `maxBytesPerString` parameter determines how much aggregation space to allocate per string. Strings longer than this limit will be truncated. This parameter should be set as low as possible, since high values will lead to wasted memory.| For advice on choosing approximate aggregation functions, check out our [approximate aggregations documentation](aggregations.html#approx). diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregator.java index 11a355830e79..8b4a89d0d72e 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregator.java @@ -20,42 +20,29 @@ package org.apache.druid.query.aggregation.first; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.Aggregator; import org.apache.druid.segment.BaseDoubleColumnValueSelector; import org.apache.druid.segment.BaseLongColumnValueSelector; -public class DoubleFirstAggregator implements Aggregator +public class DoubleFirstAggregator extends NumericFirstAggregator { - - private final BaseDoubleColumnValueSelector valueSelector; - private final BaseLongColumnValueSelector timeSelector; - - protected long firstTime; - protected double firstValue; + double firstValue; public DoubleFirstAggregator(BaseLongColumnValueSelector timeSelector, BaseDoubleColumnValueSelector valueSelector) { - this.valueSelector = valueSelector; - this.timeSelector = timeSelector; - - firstTime = Long.MAX_VALUE; + super(timeSelector, valueSelector); firstValue = 0; } @Override - public void aggregate() + void setCurrentValue() { - long time = timeSelector.getLong(); - if (time < firstTime) { - firstTime = time; - firstValue = valueSelector.getDouble(); - } + firstValue = valueSelector.getDouble(); } @Override public Object get() { - return new SerializablePair<>(firstTime, firstValue); + return new SerializablePair<>(firstTime, rhsNull ? null : firstValue); } @Override @@ -75,11 +62,5 @@ public long getLong() { return (long) firstValue; } - - @Override - public void close() - { - - } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregatorFactory.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregatorFactory.java index eb39e35e4efc..0920ee3e7bfc 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregatorFactory.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregatorFactory.java @@ -30,10 +30,11 @@ import org.apache.druid.query.aggregation.AggregatorFactory; import org.apache.druid.query.aggregation.AggregatorUtil; import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.aggregation.NullableNumericAggregatorFactory; import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; +import org.apache.druid.segment.BaseDoubleColumnValueSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.ColumnValueSelector; +import org.apache.druid.segment.NilColumnValueSelector; import org.apache.druid.segment.column.ColumnHolder; import javax.annotation.Nullable; @@ -45,10 +46,34 @@ import java.util.Map; import java.util.Objects; -public class DoubleFirstAggregatorFactory extends NullableNumericAggregatorFactory +public class DoubleFirstAggregatorFactory extends AggregatorFactory { + private static final Aggregator NIL_AGGREGATOR = new DoubleFirstAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate() + { + // no-op + } + }; + + private static final BufferAggregator NIL_BUFFER_AGGREGATOR = new DoubleFirstBufferAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate(ByteBuffer buf, int position) + { + // no-op + } + }; + public static final Comparator> VALUE_COMPARATOR = - Comparator.comparingDouble(o -> o.rhs); + SerializablePair.createNullHandlingComparator(Double::compare, true); private final String fieldName; private final String name; @@ -69,24 +94,31 @@ public DoubleFirstAggregatorFactory( } @Override - protected ColumnValueSelector selector(ColumnSelectorFactory metricFactory) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { - return metricFactory.makeColumnValueSelector(fieldName); - } - - @Override - protected Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) - { - return new DoubleFirstAggregator(metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), selector); + final BaseDoubleColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_AGGREGATOR; + } else { + return new DoubleFirstAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override - protected BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { - return new DoubleFirstBufferAggregator( - metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), - selector - ); + final BaseDoubleColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_BUFFER_AGGREGATOR; + } else { + return new DoubleFirstBufferAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override @@ -126,35 +158,54 @@ public AggregatorFactory getCombiningFactory() return new DoubleFirstAggregatorFactory(name, name) { @Override - public Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = + metricFactory.makeColumnValueSelector(name); return new DoubleFirstAggregator(null, null) { @Override public void aggregate() { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); if (pair.lhs < firstTime) { firstTime = pair.lhs; - firstValue = pair.rhs; + if (pair.rhs != null) { + firstValue = pair.rhs; + rhsNull = false; + } else { + rhsNull = true; + } } } }; } @Override - public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = + metricFactory.makeColumnValueSelector(name); return new DoubleFirstBufferAggregator(null, null) { + @Override + public void putValue(ByteBuffer buf, int position) + { + SerializablePair pair = selector.getObject(); + buf.putDouble(position, pair.rhs); + } + @Override public void aggregate(ByteBuffer buf, int position) { SerializablePair pair = (SerializablePair) selector.getObject(); long firstTime = buf.getLong(position); if (pair.lhs < firstTime) { - buf.putLong(position, pair.lhs); - buf.putDouble(position + Long.BYTES, pair.rhs); + if (pair.rhs != null) { + updateTimeWithValue(buf, position, pair.lhs); + } else { + updateTimeWithNull(buf, position, pair.lhs); + } } } @@ -178,6 +229,9 @@ public List getRequiredColumns() public Object deserialize(Object object) { Map map = (Map) object; + if (map.get("rhs") == null) { + return new SerializablePair<>(((Number) map.get("lhs")).longValue(), null); + } return new SerializablePair<>(((Number) map.get("lhs")).longValue(), ((Number) map.get("rhs")).doubleValue()); } @@ -221,16 +275,15 @@ public byte[] getCacheKey() @Override public String getTypeName() { - if (storeDoubleAsFloat) { - return "float"; - } - return "double"; + // if we don't pretend to be a primitive, group by v1 gets sad and doesn't work because no complex type serde + return storeDoubleAsFloat ? "float" : "double"; } @Override public int getMaxIntermediateSize() { - return Long.BYTES + Double.BYTES; + // timestamp, is null, value + return Long.BYTES + Byte.BYTES + Double.BYTES; } @Override diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstBufferAggregator.java index e917c680ef96..dabade475369 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstBufferAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstBufferAggregator.java @@ -20,79 +20,55 @@ package org.apache.druid.query.aggregation.first; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; import org.apache.druid.segment.BaseDoubleColumnValueSelector; import org.apache.druid.segment.BaseLongColumnValueSelector; import java.nio.ByteBuffer; -public class DoubleFirstBufferAggregator implements BufferAggregator +public class DoubleFirstBufferAggregator extends NumericFirstBufferAggregator { - private final BaseLongColumnValueSelector timeSelector; - private final BaseDoubleColumnValueSelector valueSelector; - public DoubleFirstBufferAggregator( BaseLongColumnValueSelector timeSelector, BaseDoubleColumnValueSelector valueSelector ) { - this.timeSelector = timeSelector; - this.valueSelector = valueSelector; + super(timeSelector, valueSelector); } @Override - public void init(ByteBuffer buf, int position) + void initValue(ByteBuffer buf, int position) { - buf.putLong(position, Long.MAX_VALUE); - buf.putDouble(position + Long.BYTES, 0); + buf.putDouble(position, 0); } @Override - public void aggregate(ByteBuffer buf, int position) + void putValue(ByteBuffer buf, int position) { - long time = timeSelector.getLong(); - long firstTime = buf.getLong(position); - if (time < firstTime) { - buf.putLong(position, time); - buf.putDouble(position + Long.BYTES, valueSelector.getDouble()); - } + buf.putDouble(position, valueSelector.getDouble()); } @Override public Object get(ByteBuffer buf, int position) { - return new SerializablePair<>(buf.getLong(position), buf.getDouble(position + Long.BYTES)); + final boolean rhsNull = isValueNull(buf, position); + return new SerializablePair<>(buf.getLong(position), rhsNull ? null : buf.getDouble(position + VALUE_OFFSET)); } @Override public float getFloat(ByteBuffer buf, int position) { - return (float) buf.getDouble(position + Long.BYTES); + return (float) buf.getDouble(position + VALUE_OFFSET); } @Override public long getLong(ByteBuffer buf, int position) { - return (long) buf.getDouble(position + Long.BYTES); + return (long) buf.getDouble(position + VALUE_OFFSET); } @Override public double getDouble(ByteBuffer buf, int position) { - return buf.getDouble(position + Long.BYTES); - } - - @Override - public void close() - { - // no resources to cleanup - } - - @Override - public void inspectRuntimeShape(RuntimeShapeInspector inspector) - { - inspector.visit("timeSelector", timeSelector); - inspector.visit("valueSelector", valueSelector); + return buf.getDouble(position + VALUE_OFFSET); } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstAggregator.java index af7167c4e1de..2c0f62934ffe 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstAggregator.java @@ -20,45 +20,32 @@ package org.apache.druid.query.aggregation.first; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.Aggregator; import org.apache.druid.segment.BaseFloatColumnValueSelector; import org.apache.druid.segment.BaseLongColumnValueSelector; -public class FloatFirstAggregator implements Aggregator +public class FloatFirstAggregator extends NumericFirstAggregator { - - private final BaseFloatColumnValueSelector valueSelector; - private final BaseLongColumnValueSelector timeSelector; - - protected long firstTime; - protected float firstValue; + float firstValue; public FloatFirstAggregator( BaseLongColumnValueSelector timeSelector, BaseFloatColumnValueSelector valueSelector ) { - this.valueSelector = valueSelector; - this.timeSelector = timeSelector; - - firstTime = Long.MAX_VALUE; + super(timeSelector, valueSelector); firstValue = 0; } @Override - public void aggregate() + void setCurrentValue() { - long time = timeSelector.getLong(); - if (time < firstTime) { - firstTime = time; - firstValue = valueSelector.getFloat(); - } + firstValue = valueSelector.getFloat(); } @Override public Object get() { - return new SerializablePair<>(firstTime, firstValue); + return new SerializablePair<>(firstTime, rhsNull ? null : firstValue); } @Override @@ -70,7 +57,7 @@ public float getFloat() @Override public double getDouble() { - return (double) firstValue; + return firstValue; } @Override @@ -78,11 +65,5 @@ public long getLong() { return (long) firstValue; } - - @Override - public void close() - { - - } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstAggregatorFactory.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstAggregatorFactory.java index 4a2d1c12abd0..74d10fa516d2 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstAggregatorFactory.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstAggregatorFactory.java @@ -30,10 +30,11 @@ import org.apache.druid.query.aggregation.AggregatorFactory; import org.apache.druid.query.aggregation.AggregatorUtil; import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.aggregation.NullableNumericAggregatorFactory; import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; +import org.apache.druid.segment.BaseFloatColumnValueSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.ColumnValueSelector; +import org.apache.druid.segment.NilColumnValueSelector; import org.apache.druid.segment.column.ColumnHolder; import javax.annotation.Nullable; @@ -45,10 +46,34 @@ import java.util.Map; import java.util.Objects; -public class FloatFirstAggregatorFactory extends NullableNumericAggregatorFactory +public class FloatFirstAggregatorFactory extends AggregatorFactory { + private static final Aggregator NIL_AGGREGATOR = new FloatFirstAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate() + { + // no-op + } + }; + + private static final BufferAggregator NIL_BUFFER_AGGREGATOR = new FloatFirstBufferAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate(ByteBuffer buf, int position) + { + // no-op + } + }; + public static final Comparator> VALUE_COMPARATOR = - Comparator.comparingDouble(o -> o.rhs); + SerializablePair.createNullHandlingComparator(Float::compare, true); private final String fieldName; private final String name; @@ -67,24 +92,31 @@ public FloatFirstAggregatorFactory( } @Override - protected ColumnValueSelector selector(ColumnSelectorFactory metricFactory) - { - return metricFactory.makeColumnValueSelector(fieldName); - } - - @Override - protected Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { - return new FloatFirstAggregator(metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), selector); + final BaseFloatColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_AGGREGATOR; + } else { + return new FloatFirstAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override - protected BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { - return new FloatFirstBufferAggregator( - metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), - selector - ); + final BaseFloatColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_BUFFER_AGGREGATOR; + } else { + return new FloatFirstBufferAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override @@ -121,38 +153,56 @@ public AggregateCombiner makeAggregateCombiner() @Override public AggregatorFactory getCombiningFactory() { + return new FloatFirstAggregatorFactory(name, name) { @Override - public Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = metricFactory.makeColumnValueSelector(name); return new FloatFirstAggregator(null, null) { @Override public void aggregate() { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); if (pair.lhs < firstTime) { firstTime = pair.lhs; - firstValue = pair.rhs; + if (pair.rhs != null) { + firstValue = pair.rhs; + rhsNull = false; + } else { + rhsNull = true; + } } } }; } @Override - public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = metricFactory.makeColumnValueSelector(name); return new FloatFirstBufferAggregator(null, null) { + @Override + public void putValue(ByteBuffer buf, int position) + { + SerializablePair pair = selector.getObject(); + buf.putFloat(position, pair.rhs); + } + @Override public void aggregate(ByteBuffer buf, int position) { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); long firstTime = buf.getLong(position); if (pair.lhs < firstTime) { - buf.putLong(position, pair.lhs); - buf.putFloat(position + Long.BYTES, pair.rhs); + if (pair.rhs != null) { + updateTimeWithValue(buf, position, pair.lhs); + } else { + updateTimeWithNull(buf, position, pair.lhs); + } } } @@ -176,6 +226,9 @@ public List getRequiredColumns() public Object deserialize(Object object) { Map map = (Map) object; + if (map.get("rhs") == null) { + return new SerializablePair<>(((Number) map.get("lhs")).longValue(), null); + } return new SerializablePair<>(((Number) map.get("lhs")).longValue(), ((Number) map.get("rhs")).floatValue()); } @@ -219,13 +272,15 @@ public byte[] getCacheKey() @Override public String getTypeName() { + // if we don't pretend to be a primitive, group by v1 gets sad and doesn't work because no complex type serde return "float"; } @Override public int getMaxIntermediateSize() { - return Long.BYTES + Float.BYTES; + // timestamp, is null, value + return Long.BYTES + Byte.BYTES + Float.BYTES; } @Override diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstBufferAggregator.java index d7797b52eb6c..cf7d272b0085 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstBufferAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstBufferAggregator.java @@ -20,79 +20,55 @@ package org.apache.druid.query.aggregation.first; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; import org.apache.druid.segment.BaseFloatColumnValueSelector; import org.apache.druid.segment.BaseLongColumnValueSelector; import java.nio.ByteBuffer; -public class FloatFirstBufferAggregator implements BufferAggregator +public class FloatFirstBufferAggregator extends NumericFirstBufferAggregator { - private final BaseLongColumnValueSelector timeSelector; - private final BaseFloatColumnValueSelector valueSelector; - public FloatFirstBufferAggregator( BaseLongColumnValueSelector timeSelector, BaseFloatColumnValueSelector valueSelector ) { - this.timeSelector = timeSelector; - this.valueSelector = valueSelector; + super(timeSelector, valueSelector); } @Override - public void init(ByteBuffer buf, int position) + void initValue(ByteBuffer buf, int position) { - buf.putLong(position, Long.MAX_VALUE); - buf.putFloat(position + Long.BYTES, 0); + buf.putFloat(position, 0); } @Override - public void aggregate(ByteBuffer buf, int position) + void putValue(ByteBuffer buf, int position) { - long time = timeSelector.getLong(); - long firstTime = buf.getLong(position); - if (time < firstTime) { - buf.putLong(position, time); - buf.putFloat(position + Long.BYTES, valueSelector.getFloat()); - } + buf.putFloat(position, valueSelector.getFloat()); } @Override public Object get(ByteBuffer buf, int position) { - return new SerializablePair<>(buf.getLong(position), buf.getFloat(position + Long.BYTES)); + final boolean rhsNull = isValueNull(buf, position); + return new SerializablePair<>(buf.getLong(position), rhsNull ? null : buf.getFloat(position + VALUE_OFFSET)); } @Override public float getFloat(ByteBuffer buf, int position) { - return buf.getFloat(position + Long.BYTES); + return buf.getFloat(position + VALUE_OFFSET); } @Override public long getLong(ByteBuffer buf, int position) { - return (long) buf.getFloat(position + Long.BYTES); + return (long) buf.getFloat(position + VALUE_OFFSET); } @Override public double getDouble(ByteBuffer buf, int position) { - return (double) buf.getFloat(position + Long.BYTES); - } - - @Override - public void close() - { - // no resources to cleanup - } - - @Override - public void inspectRuntimeShape(RuntimeShapeInspector inspector) - { - inspector.visit("timeSelector", timeSelector); - inspector.visit("valueSelector", valueSelector); + return buf.getFloat(position + VALUE_OFFSET); } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstAggregator.java index d1fd9aebd268..8cda544521ea 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstAggregator.java @@ -20,41 +20,28 @@ package org.apache.druid.query.aggregation.first; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.Aggregator; import org.apache.druid.segment.BaseLongColumnValueSelector; -public class LongFirstAggregator implements Aggregator +public class LongFirstAggregator extends NumericFirstAggregator { - - private final BaseLongColumnValueSelector valueSelector; - private final BaseLongColumnValueSelector timeSelector; - - protected long firstTime; - protected long firstValue; + long firstValue; public LongFirstAggregator(BaseLongColumnValueSelector timeSelector, BaseLongColumnValueSelector valueSelector) { - this.valueSelector = valueSelector; - this.timeSelector = timeSelector; - - firstTime = Long.MAX_VALUE; + super(timeSelector, valueSelector); firstValue = 0; } @Override - public void aggregate() + void setCurrentValue() { - long time = timeSelector.getLong(); - if (time < firstTime) { - firstTime = time; - firstValue = valueSelector.getLong(); - } + firstValue = valueSelector.getLong(); } @Override public Object get() { - return new SerializablePair<>(firstTime, firstValue); + return new SerializablePair<>(firstTime, rhsNull ? null : firstValue); } @Override @@ -74,10 +61,4 @@ public long getLong() { return firstValue; } - - @Override - public void close() - { - - } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstAggregatorFactory.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstAggregatorFactory.java index 399f1fd81617..5b7e16dec0bc 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstAggregatorFactory.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstAggregatorFactory.java @@ -30,10 +30,11 @@ import org.apache.druid.query.aggregation.AggregatorFactory; import org.apache.druid.query.aggregation.AggregatorUtil; import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.aggregation.NullableNumericAggregatorFactory; import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; +import org.apache.druid.segment.BaseLongColumnValueSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.ColumnValueSelector; +import org.apache.druid.segment.NilColumnValueSelector; import org.apache.druid.segment.column.ColumnHolder; import javax.annotation.Nullable; @@ -44,10 +45,34 @@ import java.util.List; import java.util.Map; -public class LongFirstAggregatorFactory extends NullableNumericAggregatorFactory +public class LongFirstAggregatorFactory extends AggregatorFactory { + private static final Aggregator NIL_AGGREGATOR = new LongFirstAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate() + { + // no-op + } + }; + + private static final BufferAggregator NIL_BUFFER_AGGREGATOR = new LongFirstBufferAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate(ByteBuffer buf, int position) + { + // no-op + } + }; + public static final Comparator> VALUE_COMPARATOR = - Comparator.comparingLong(o -> o.rhs); + SerializablePair.createNullHandlingComparator(Long::compare, true); private final String fieldName; private final String name; @@ -66,24 +91,31 @@ public LongFirstAggregatorFactory( } @Override - protected ColumnValueSelector selector(ColumnSelectorFactory metricFactory) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { - return metricFactory.makeColumnValueSelector(fieldName); - } - - @Override - protected Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) - { - return new LongFirstAggregator(metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), selector); + final BaseLongColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_AGGREGATOR; + } else { + return new LongFirstAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override - protected BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { - return new LongFirstBufferAggregator( - metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), - selector - ); + final BaseLongColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_BUFFER_AGGREGATOR; + } else { + return new LongFirstBufferAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override @@ -123,35 +155,52 @@ public AggregatorFactory getCombiningFactory() return new LongFirstAggregatorFactory(name, name) { @Override - public Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = metricFactory.makeColumnValueSelector(name); return new LongFirstAggregator(null, null) { @Override public void aggregate() { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); if (pair.lhs < firstTime) { firstTime = pair.lhs; - firstValue = pair.rhs; + if (pair.rhs != null) { + firstValue = pair.rhs; + rhsNull = false; + } else { + rhsNull = true; + } } } }; } @Override - public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = metricFactory.makeColumnValueSelector(name); return new LongFirstBufferAggregator(null, null) { + @Override + public void putValue(ByteBuffer buf, int position) + { + SerializablePair pair = selector.getObject(); + buf.putLong(position, pair.rhs); + } + @Override public void aggregate(ByteBuffer buf, int position) { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); long firstTime = buf.getLong(position); if (pair.lhs < firstTime) { - buf.putLong(position, pair.lhs); - buf.putLong(position + Long.BYTES, pair.rhs); + if (pair.rhs != null) { + updateTimeWithValue(buf, position, pair.lhs); + } else { + updateTimeWithNull(buf, position, pair.lhs); + } } } @@ -175,6 +224,9 @@ public List getRequiredColumns() public Object deserialize(Object object) { Map map = (Map) object; + if (map.get("rhs") == null) { + return new SerializablePair<>(((Number) map.get("lhs")).longValue(), null); + } return new SerializablePair<>(((Number) map.get("lhs")).longValue(), ((Number) map.get("rhs")).longValue()); } @@ -218,13 +270,15 @@ public byte[] getCacheKey() @Override public String getTypeName() { + // if we don't pretend to be a primitive, group by v1 gets sad and doesn't work because no complex type serde return "long"; } @Override public int getMaxIntermediateSize() { - return Long.BYTES * 2; + // timestamp, is null, value + return Long.BYTES + Byte.BYTES + Long.BYTES; } @Override diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstBufferAggregator.java index 2ef812af81d7..582cda160153 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstBufferAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstBufferAggregator.java @@ -20,75 +20,51 @@ package org.apache.druid.query.aggregation.first; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; import org.apache.druid.segment.BaseLongColumnValueSelector; import java.nio.ByteBuffer; -public class LongFirstBufferAggregator implements BufferAggregator +public class LongFirstBufferAggregator extends NumericFirstBufferAggregator { - private final BaseLongColumnValueSelector timeSelector; - private final BaseLongColumnValueSelector valueSelector; - public LongFirstBufferAggregator(BaseLongColumnValueSelector timeSelector, BaseLongColumnValueSelector valueSelector) { - this.timeSelector = timeSelector; - this.valueSelector = valueSelector; + super(timeSelector, valueSelector); } @Override - public void init(ByteBuffer buf, int position) + void initValue(ByteBuffer buf, int position) { - buf.putLong(position, Long.MAX_VALUE); - buf.putLong(position + Long.BYTES, 0); + buf.putLong(position, 0); } @Override - public void aggregate(ByteBuffer buf, int position) + void putValue(ByteBuffer buf, int position) { - long time = timeSelector.getLong(); - long firstTime = buf.getLong(position); - if (time < firstTime) { - buf.putLong(position, time); - buf.putLong(position + Long.BYTES, valueSelector.getLong()); - } + buf.putLong(position, valueSelector.getLong()); } @Override public Object get(ByteBuffer buf, int position) { - return new SerializablePair<>(buf.getLong(position), buf.getLong(position + Long.BYTES)); + final boolean rhsNull = isValueNull(buf, position); + return new SerializablePair<>(buf.getLong(position), rhsNull ? null : buf.getLong(position + VALUE_OFFSET)); } @Override public float getFloat(ByteBuffer buf, int position) { - return (float) buf.getLong(position + Long.BYTES); + return (float) buf.getLong(position + VALUE_OFFSET); } @Override public long getLong(ByteBuffer buf, int position) { - return buf.getLong(position + Long.BYTES); + return buf.getLong(position + VALUE_OFFSET); } @Override public double getDouble(ByteBuffer buf, int position) { - return (double) buf.getLong(position + Long.BYTES); - } - - @Override - public void close() - { - // no resources to cleanup - } - - @Override - public void inspectRuntimeShape(RuntimeShapeInspector inspector) - { - inspector.visit("timeSelector", timeSelector); - inspector.visit("valueSelector", valueSelector); + return (double) buf.getLong(position + VALUE_OFFSET); } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstAggregator.java new file mode 100644 index 000000000000..e7b60b9e5c81 --- /dev/null +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstAggregator.java @@ -0,0 +1,74 @@ +/* + * 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.first; + +import org.apache.druid.common.config.NullHandling; +import org.apache.druid.query.aggregation.Aggregator; +import org.apache.druid.segment.BaseLongColumnValueSelector; +import org.apache.druid.segment.BaseNullableColumnValueSelector; + +/** + * Base type for on heap 'first' aggregator for primitive numeric column selectors + */ +public abstract class NumericFirstAggregator implements Aggregator +{ + private final boolean useDefault = NullHandling.replaceWithDefault(); + private final BaseLongColumnValueSelector timeSelector; + + final TSelector valueSelector; + + long firstTime; + boolean rhsNull; + + public NumericFirstAggregator(BaseLongColumnValueSelector timeSelector, TSelector valueSelector) + { + this.timeSelector = timeSelector; + this.valueSelector = valueSelector; + + firstTime = Long.MAX_VALUE; + rhsNull = !useDefault; + } + + /** + * Store the current primitive typed 'first' value + */ + abstract void setCurrentValue(); + + @Override + public void aggregate() + { + long time = timeSelector.getLong(); + if (time < firstTime) { + firstTime = time; + if (useDefault || !valueSelector.isNull()) { + setCurrentValue(); + rhsNull = false; + } else { + rhsNull = true; + } + } + } + + @Override + public void close() + { + // nothing to close + } +} diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstBufferAggregator.java new file mode 100644 index 000000000000..ebb0a87169a0 --- /dev/null +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/NumericFirstBufferAggregator.java @@ -0,0 +1,112 @@ +/* + * 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.first; + +import org.apache.druid.common.config.NullHandling; +import org.apache.druid.query.aggregation.BufferAggregator; +import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; +import org.apache.druid.segment.BaseLongColumnValueSelector; +import org.apache.druid.segment.BaseNullableColumnValueSelector; + +import java.nio.ByteBuffer; + +/** + * Base type for buffer based 'first' aggregator for primitive numeric column selectors + */ +public abstract class NumericFirstBufferAggregator + implements BufferAggregator +{ + static final int NULL_OFFSET = Long.BYTES; + static final int VALUE_OFFSET = NULL_OFFSET + Byte.BYTES; + + private final boolean useDefault = NullHandling.replaceWithDefault(); + private final BaseLongColumnValueSelector timeSelector; + + final TSelector valueSelector; + + public NumericFirstBufferAggregator(BaseLongColumnValueSelector timeSelector, TSelector valueSelector) + { + this.timeSelector = timeSelector; + this.valueSelector = valueSelector; + } + + /** + * Initialize the buffer value at the position of {@link #VALUE_OFFSET} + */ + abstract void initValue(ByteBuffer buf, int position); + + /** + * Place the primitive value in the buffer at the position of {@link #VALUE_OFFSET} + */ + abstract void putValue(ByteBuffer buf, int position); + + void updateTimeWithValue(ByteBuffer buf, int position, long time) + { + buf.putLong(position, time); + buf.put(position + NULL_OFFSET, NullHandling.IS_NOT_NULL_BYTE); + putValue(buf, position + VALUE_OFFSET); + } + + void updateTimeWithNull(ByteBuffer buf, int position, long time) + { + buf.putLong(position, time); + buf.put(position + NULL_OFFSET, NullHandling.IS_NULL_BYTE); + } + + boolean isValueNull(ByteBuffer buf, int position) + { + return buf.get(position + NULL_OFFSET) == NullHandling.IS_NULL_BYTE; + } + + @Override + public void init(ByteBuffer buf, int position) + { + buf.putLong(position, Long.MAX_VALUE); + buf.put(position + NULL_OFFSET, useDefault ? NullHandling.IS_NOT_NULL_BYTE : NullHandling.IS_NULL_BYTE); + initValue(buf, position + VALUE_OFFSET); + } + + @Override + public void aggregate(ByteBuffer buf, int position) + { + long time = timeSelector.getLong(); + long firstTime = buf.getLong(position); + if (time < firstTime) { + if (useDefault || !valueSelector.isNull()) { + updateTimeWithValue(buf, position, time); + } else { + updateTimeWithNull(buf, position, time); + } + } + } + + @Override + public void close() + { + // no resources to cleanup + } + + @Override + public void inspectRuntimeShape(RuntimeShapeInspector inspector) + { + inspector.visit("timeSelector", timeSelector); + inspector.visit("valueSelector", valueSelector); + } +} diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstAggregator.java index 2d5ee990ed1f..5b581d5c1cbe 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstAggregator.java @@ -64,7 +64,7 @@ public void aggregate() valueSelector ); - if (inPair != null && inPair.rhs != null && inPair.lhs < firstTime) { + if (inPair != null && inPair.lhs < firstTime) { firstTime = inPair.lhs; firstValue = StringUtils.fastLooseChop(inPair.rhs, maxStringBytes); } @@ -73,11 +73,8 @@ public void aggregate() if (time < firstTime) { final String value = DimensionHandlerUtils.convertObjectToString(valueSelector.getObject()); - - if (value != null) { - firstTime = time; - firstValue = StringUtils.fastLooseChop(value, maxStringBytes); - } + firstTime = time; + firstValue = StringUtils.fastLooseChop(value, maxStringBytes); } } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstAggregatorFactory.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstAggregatorFactory.java index 6ad8558ef0bd..ada369829e7b 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstAggregatorFactory.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstAggregatorFactory.java @@ -34,9 +34,11 @@ import org.apache.druid.query.cache.CacheKeyBuilder; import org.apache.druid.segment.BaseObjectColumnValueSelector; import org.apache.druid.segment.ColumnSelectorFactory; +import org.apache.druid.segment.NilColumnValueSelector; import org.apache.druid.segment.column.ColumnHolder; import javax.annotation.Nullable; +import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; @@ -47,6 +49,34 @@ @JsonTypeName("stringFirst") public class StringFirstAggregatorFactory extends AggregatorFactory { + private static final Aggregator NIL_AGGREGATOR = new StringFirstAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance(), + 0, + false + ) + { + @Override + public void aggregate() + { + // no-op + } + }; + + private static final BufferAggregator NIL_BUFFER_AGGREGATOR = new StringFirstBufferAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance(), + 0, + false + ) + { + @Override + public void aggregate(ByteBuffer buf, int position) + { + // no-op + } + }; + public static final int DEFAULT_MAX_STRING_SIZE = 1024; public static final Comparator TIME_COMPARATOR = (o1, o2) -> Longs.compare( @@ -120,24 +150,32 @@ public StringFirstAggregatorFactory( public Aggregator factorize(ColumnSelectorFactory metricFactory) { final BaseObjectColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); - return new StringFirstAggregator( - metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), - valueSelector, - maxStringBytes, - StringFirstLastUtils.selectorNeedsFoldCheck(valueSelector, metricFactory.getColumnCapabilities(fieldName)) - ); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_AGGREGATOR; + } else { + return new StringFirstAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector, + maxStringBytes, + StringFirstLastUtils.selectorNeedsFoldCheck(valueSelector, metricFactory.getColumnCapabilities(fieldName)) + ); + } } @Override public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { final BaseObjectColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); - return new StringFirstBufferAggregator( - metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), - valueSelector, - maxStringBytes, - StringFirstLastUtils.selectorNeedsFoldCheck(valueSelector, metricFactory.getColumnCapabilities(fieldName)) - ); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_BUFFER_AGGREGATOR; + } else { + return new StringFirstBufferAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector, + maxStringBytes, + StringFirstLastUtils.selectorNeedsFoldCheck(valueSelector, metricFactory.getColumnCapabilities(fieldName)) + ); + } } @Override diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstBufferAggregator.java index b7d5ac89b06a..d84793e997a9 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstBufferAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstBufferAggregator.java @@ -71,7 +71,7 @@ public void aggregate(ByteBuffer buf, int position) valueSelector ); - if (inPair != null && inPair.rhs != null) { + if (inPair != null) { final long firstTime = buf.getLong(position); if (inPair.lhs < firstTime) { StringFirstLastUtils.writePair( @@ -89,14 +89,12 @@ public void aggregate(ByteBuffer buf, int position) if (time < firstTime) { final String value = DimensionHandlerUtils.convertObjectToString(valueSelector.getObject()); - if (value != null) { - StringFirstLastUtils.writePair( - buf, - position, - new SerializablePairLongString(time, value), - maxStringBytes - ); - } + StringFirstLastUtils.writePair( + buf, + position, + new SerializablePairLongString(time, value), + maxStringBytes + ); } } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastAggregator.java index d6678d7f2ea6..3f6a1506bad6 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastAggregator.java @@ -20,42 +20,29 @@ package org.apache.druid.query.aggregation.last; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.Aggregator; import org.apache.druid.segment.BaseDoubleColumnValueSelector; import org.apache.druid.segment.BaseLongColumnValueSelector; -public class DoubleLastAggregator implements Aggregator +public class DoubleLastAggregator extends NumericLastAggregator { - - private final BaseDoubleColumnValueSelector valueSelector; - private final BaseLongColumnValueSelector timeSelector; - - protected long lastTime; - protected double lastValue; + double lastValue; public DoubleLastAggregator(BaseLongColumnValueSelector timeSelector, BaseDoubleColumnValueSelector valueSelector) { - this.valueSelector = valueSelector; - this.timeSelector = timeSelector; - - lastTime = Long.MIN_VALUE; + super(timeSelector, valueSelector); lastValue = 0; } @Override - public void aggregate() + void setCurrentValue() { - long time = timeSelector.getLong(); - if (time >= lastTime) { - lastTime = time; - lastValue = valueSelector.getDouble(); - } + lastValue = valueSelector.getDouble(); } @Override public Object get() { - return new SerializablePair<>(lastTime, lastValue); + return new SerializablePair<>(lastTime, rhsNull ? null : lastValue); } @Override @@ -75,10 +62,4 @@ public double getDouble() { return lastValue; } - - @Override - public void close() - { - - } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastAggregatorFactory.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastAggregatorFactory.java index ad90857cc82f..dd2503ed6e93 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastAggregatorFactory.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastAggregatorFactory.java @@ -30,12 +30,13 @@ import org.apache.druid.query.aggregation.AggregatorFactory; import org.apache.druid.query.aggregation.AggregatorUtil; import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.aggregation.NullableNumericAggregatorFactory; import org.apache.druid.query.aggregation.first.DoubleFirstAggregatorFactory; import org.apache.druid.query.aggregation.first.LongFirstAggregatorFactory; import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; +import org.apache.druid.segment.BaseDoubleColumnValueSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.ColumnValueSelector; +import org.apache.druid.segment.NilColumnValueSelector; import org.apache.druid.segment.column.ColumnHolder; import javax.annotation.Nullable; @@ -47,8 +48,31 @@ import java.util.Map; import java.util.Objects; -public class DoubleLastAggregatorFactory extends NullableNumericAggregatorFactory +public class DoubleLastAggregatorFactory extends AggregatorFactory { + private static final Aggregator NIL_AGGREGATOR = new DoubleLastAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate() + { + // no-op + } + }; + + private static final BufferAggregator NIL_BUFFER_AGGREGATOR = new DoubleLastBufferAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate(ByteBuffer buf, int position) + { + // no-op + } + }; private final String fieldName; private final String name; @@ -68,24 +92,31 @@ public DoubleLastAggregatorFactory( } @Override - protected ColumnValueSelector selector(ColumnSelectorFactory metricFactory) - { - return metricFactory.makeColumnValueSelector(fieldName); - } - - @Override - protected Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { - return new DoubleLastAggregator(metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), selector); + final BaseDoubleColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_AGGREGATOR; + } else { + return new DoubleLastAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override - protected BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { - return new DoubleLastBufferAggregator( - metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), - selector - ); + final BaseDoubleColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_BUFFER_AGGREGATOR; + } else { + return new DoubleLastBufferAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override @@ -125,35 +156,54 @@ public AggregatorFactory getCombiningFactory() return new DoubleLastAggregatorFactory(name, name) { @Override - public Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = + metricFactory.makeColumnValueSelector(name); return new DoubleLastAggregator(null, null) { @Override public void aggregate() { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); if (pair.lhs >= lastTime) { lastTime = pair.lhs; - lastValue = pair.rhs; + if (pair.rhs != null) { + lastValue = pair.rhs; + rhsNull = false; + } else { + rhsNull = true; + } } } }; } @Override - public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = + metricFactory.makeColumnValueSelector(name); return new DoubleLastBufferAggregator(null, null) { + @Override + public void putValue(ByteBuffer buf, int position) + { + SerializablePair pair = selector.getObject(); + buf.putDouble(position, pair.rhs); + } + @Override public void aggregate(ByteBuffer buf, int position) { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); long lastTime = buf.getLong(position); if (pair.lhs >= lastTime) { - buf.putLong(position, pair.lhs); - buf.putDouble(position + Long.BYTES, pair.rhs); + if (pair.rhs != null) { + updateTimeWithValue(buf, position, pair.lhs); + } else { + updateTimeWithNull(buf, position, pair.lhs); + } } } @@ -177,6 +227,9 @@ public List getRequiredColumns() public Object deserialize(Object object) { Map map = (Map) object; + if (map.get("rhs") == null) { + return new SerializablePair<>(((Number) map.get("lhs")).longValue(), null); + } return new SerializablePair<>(((Number) map.get("lhs")).longValue(), ((Number) map.get("rhs")).doubleValue()); } @@ -220,17 +273,15 @@ public byte[] getCacheKey() @Override public String getTypeName() { - - if (storeDoubleAsFloat) { - return "float"; - } - return "double"; + // if we don't pretend to be a primitive, group by v1 gets sad and doesn't work because no complex type serde + return storeDoubleAsFloat ? "float" : "double"; } @Override public int getMaxIntermediateSize() { - return Long.BYTES + Double.BYTES; + // timestamp, is null, value + return Long.BYTES + Byte.BYTES + Double.BYTES; } @Override diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastBufferAggregator.java index b389b5f9af85..8acddce53a83 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastBufferAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/DoubleLastBufferAggregator.java @@ -20,79 +20,55 @@ package org.apache.druid.query.aggregation.last; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; import org.apache.druid.segment.BaseDoubleColumnValueSelector; import org.apache.druid.segment.BaseLongColumnValueSelector; import java.nio.ByteBuffer; -public class DoubleLastBufferAggregator implements BufferAggregator +public class DoubleLastBufferAggregator extends NumericLastBufferAggregator { - private final BaseLongColumnValueSelector timeSelector; - private final BaseDoubleColumnValueSelector valueSelector; - public DoubleLastBufferAggregator( BaseLongColumnValueSelector timeSelector, BaseDoubleColumnValueSelector valueSelector ) { - this.timeSelector = timeSelector; - this.valueSelector = valueSelector; + super(timeSelector, valueSelector); } @Override - public void init(ByteBuffer buf, int position) + void initValue(ByteBuffer buf, int position) { - buf.putLong(position, Long.MIN_VALUE); - buf.putDouble(position + Long.BYTES, 0); + buf.putDouble(position, 0); } @Override - public void aggregate(ByteBuffer buf, int position) + void putValue(ByteBuffer buf, int position) { - long time = timeSelector.getLong(); - long lastTime = buf.getLong(position); - if (time >= lastTime) { - buf.putLong(position, time); - buf.putDouble(position + Long.BYTES, valueSelector.getDouble()); - } + buf.putDouble(position, valueSelector.getDouble()); } @Override public Object get(ByteBuffer buf, int position) { - return new SerializablePair<>(buf.getLong(position), buf.getDouble(position + Long.BYTES)); + final boolean rhsNull = isValueNull(buf, position); + return new SerializablePair<>(buf.getLong(position), rhsNull ? null : buf.getDouble(position + VALUE_OFFSET)); } @Override public float getFloat(ByteBuffer buf, int position) { - return (float) buf.getDouble(position + Long.BYTES); + return (float) buf.getDouble(position + VALUE_OFFSET); } @Override public long getLong(ByteBuffer buf, int position) { - return (long) buf.getDouble(position + Long.BYTES); + return (long) buf.getDouble(position + VALUE_OFFSET); } @Override public double getDouble(ByteBuffer buf, int position) { - return buf.getDouble(position + Long.BYTES); - } - - @Override - public void close() - { - // no resources to cleanup - } - - @Override - public void inspectRuntimeShape(RuntimeShapeInspector inspector) - { - inspector.visit("timeSelector", timeSelector); - inspector.visit("valueSelector", valueSelector); + return buf.getDouble(position + VALUE_OFFSET); } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastAggregator.java index 77e3455c489b..1381ccb18b7a 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastAggregator.java @@ -20,42 +20,29 @@ package org.apache.druid.query.aggregation.last; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.Aggregator; import org.apache.druid.segment.BaseFloatColumnValueSelector; import org.apache.druid.segment.BaseLongColumnValueSelector; -public class FloatLastAggregator implements Aggregator +public class FloatLastAggregator extends NumericLastAggregator { - - private final BaseFloatColumnValueSelector valueSelector; - private final BaseLongColumnValueSelector timeSelector; - - protected long lastTime; - protected float lastValue; + float lastValue; public FloatLastAggregator(BaseLongColumnValueSelector timeSelector, BaseFloatColumnValueSelector valueSelector) { - this.valueSelector = valueSelector; - this.timeSelector = timeSelector; - - lastTime = Long.MIN_VALUE; + super(timeSelector, valueSelector); lastValue = 0; } @Override - public void aggregate() + void setCurrentValue() { - long time = timeSelector.getLong(); - if (time >= lastTime) { - lastTime = time; - lastValue = valueSelector.getFloat(); - } + lastValue = valueSelector.getFloat(); } @Override public Object get() { - return new SerializablePair<>(lastTime, lastValue); + return new SerializablePair<>(lastTime, rhsNull ? null : lastValue); } @Override @@ -73,12 +60,6 @@ public long getLong() @Override public double getDouble() { - return (double) lastValue; - } - - @Override - public void close() - { - + return lastValue; } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastAggregatorFactory.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastAggregatorFactory.java index 0e42f9fded7d..78024f9c618f 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastAggregatorFactory.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastAggregatorFactory.java @@ -30,12 +30,13 @@ import org.apache.druid.query.aggregation.AggregatorFactory; import org.apache.druid.query.aggregation.AggregatorUtil; import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.aggregation.NullableNumericAggregatorFactory; import org.apache.druid.query.aggregation.first.FloatFirstAggregatorFactory; import org.apache.druid.query.aggregation.first.LongFirstAggregatorFactory; import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; +import org.apache.druid.segment.BaseFloatColumnValueSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.ColumnValueSelector; +import org.apache.druid.segment.NilColumnValueSelector; import org.apache.druid.segment.column.ColumnHolder; import javax.annotation.Nullable; @@ -47,8 +48,31 @@ import java.util.Map; import java.util.Objects; -public class FloatLastAggregatorFactory extends NullableNumericAggregatorFactory +public class FloatLastAggregatorFactory extends AggregatorFactory { + private static final Aggregator NIL_AGGREGATOR = new FloatLastAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate() + { + // no-op + } + }; + + private static final BufferAggregator NIL_BUFFER_AGGREGATOR = new FloatLastBufferAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate(ByteBuffer buf, int position) + { + // no-op + } + }; private final String fieldName; private final String name; @@ -66,24 +90,31 @@ public FloatLastAggregatorFactory( } @Override - protected ColumnValueSelector selector(ColumnSelectorFactory metricFactory) - { - return metricFactory.makeColumnValueSelector(fieldName); - } - - @Override - protected Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { - return new FloatLastAggregator(metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), selector); + final BaseFloatColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_AGGREGATOR; + } else { + return new FloatLastAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override - protected BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { - return new FloatLastBufferAggregator( - metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), - selector - ); + final BaseFloatColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_BUFFER_AGGREGATOR; + } else { + return new FloatLastBufferAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override @@ -123,35 +154,52 @@ public AggregatorFactory getCombiningFactory() return new FloatLastAggregatorFactory(name, name) { @Override - public Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { + ColumnValueSelector> selector = metricFactory.makeColumnValueSelector(name); return new FloatLastAggregator(null, null) { @Override public void aggregate() { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); if (pair.lhs >= lastTime) { lastTime = pair.lhs; - lastValue = pair.rhs; + if (pair.rhs != null) { + lastValue = pair.rhs; + rhsNull = false; + } else { + rhsNull = true; + } } } }; } @Override - public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { + ColumnValueSelector> selector = metricFactory.makeColumnValueSelector(name); return new FloatLastBufferAggregator(null, null) { + @Override + public void putValue(ByteBuffer buf, int position) + { + SerializablePair pair = selector.getObject(); + buf.putFloat(position, pair.rhs); + } + @Override public void aggregate(ByteBuffer buf, int position) { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); long lastTime = buf.getLong(position); if (pair.lhs >= lastTime) { - buf.putLong(position, pair.lhs); - buf.putFloat(position + Long.BYTES, pair.rhs); + if (pair.rhs != null) { + updateTimeWithValue(buf, position, pair.lhs); + } else { + updateTimeWithNull(buf, position, pair.lhs); + } } } @@ -175,6 +223,9 @@ public List getRequiredColumns() public Object deserialize(Object object) { Map map = (Map) object; + if (map.get("rhs") == null) { + return new SerializablePair<>(((Number) map.get("lhs")).longValue(), null); + } return new SerializablePair<>(((Number) map.get("lhs")).longValue(), ((Number) map.get("rhs")).floatValue()); } @@ -219,13 +270,15 @@ public byte[] getCacheKey() @Override public String getTypeName() { + // if we don't pretend to be a primitive, group by v1 gets sad and doesn't work because no complex type serde return "float"; } @Override public int getMaxIntermediateSize() { - return Long.BYTES + Float.BYTES; + // timestamp, is null, value + return Long.BYTES + Byte.BYTES + Float.BYTES; } @Override diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastBufferAggregator.java index 59d8c0bc484c..95ad6fe5c5ee 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastBufferAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/FloatLastBufferAggregator.java @@ -20,76 +20,55 @@ package org.apache.druid.query.aggregation.last; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; import org.apache.druid.segment.BaseFloatColumnValueSelector; import org.apache.druid.segment.BaseLongColumnValueSelector; import java.nio.ByteBuffer; -public class FloatLastBufferAggregator implements BufferAggregator +public class FloatLastBufferAggregator extends NumericLastBufferAggregator { - private final BaseLongColumnValueSelector timeSelector; - private final BaseFloatColumnValueSelector valueSelector; - - public FloatLastBufferAggregator(BaseLongColumnValueSelector timeSelector, BaseFloatColumnValueSelector valueSelector) + public FloatLastBufferAggregator( + BaseLongColumnValueSelector timeSelector, + BaseFloatColumnValueSelector valueSelector + ) { - this.timeSelector = timeSelector; - this.valueSelector = valueSelector; + super(timeSelector, valueSelector); } @Override - public void init(ByteBuffer buf, int position) + void initValue(ByteBuffer buf, int position) { - buf.putLong(position, Long.MIN_VALUE); - buf.putFloat(position + Long.BYTES, 0); + buf.putFloat(position, 0); } @Override - public void aggregate(ByteBuffer buf, int position) + void putValue(ByteBuffer buf, int position) { - long time = timeSelector.getLong(); - long lastTime = buf.getLong(position); - if (time >= lastTime) { - buf.putLong(position, time); - buf.putFloat(position + Long.BYTES, valueSelector.getFloat()); - } + buf.putFloat(position, valueSelector.getFloat()); } @Override public Object get(ByteBuffer buf, int position) { - return new SerializablePair<>(buf.getLong(position), buf.getFloat(position + Long.BYTES)); + final boolean rhsNull = isValueNull(buf, position); + return new SerializablePair<>(buf.getLong(position), rhsNull ? null : buf.getFloat(position + VALUE_OFFSET)); } @Override public float getFloat(ByteBuffer buf, int position) { - return buf.getFloat(position + Long.BYTES); + return buf.getFloat(position + VALUE_OFFSET); } @Override public long getLong(ByteBuffer buf, int position) { - return (long) buf.getFloat(position + Long.BYTES); + return (long) buf.getFloat(position + VALUE_OFFSET); } @Override public double getDouble(ByteBuffer buf, int position) { - return (double) buf.getFloat(position + Long.BYTES); - } - - @Override - public void close() - { - // no resources to cleanup - } - - @Override - public void inspectRuntimeShape(RuntimeShapeInspector inspector) - { - inspector.visit("timeSelector", timeSelector); - inspector.visit("valueSelector", valueSelector); + return buf.getFloat(position + VALUE_OFFSET); } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastAggregator.java index b44371f03aaf..59a159d2d875 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastAggregator.java @@ -20,46 +20,28 @@ package org.apache.druid.query.aggregation.last; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.Aggregator; import org.apache.druid.segment.BaseLongColumnValueSelector; -public class LongLastAggregator implements Aggregator +public class LongLastAggregator extends NumericLastAggregator { - private final BaseLongColumnValueSelector valueSelector; - private final BaseLongColumnValueSelector timeSelector; - - protected long lastTime; - protected long lastValue; + long lastValue; public LongLastAggregator(BaseLongColumnValueSelector timeSelector, BaseLongColumnValueSelector valueSelector) { - this.valueSelector = valueSelector; - this.timeSelector = timeSelector; - - lastTime = Long.MIN_VALUE; + super(timeSelector, valueSelector); lastValue = 0; } @Override - public void aggregate() - { - long time = timeSelector.getLong(); - if (time >= lastTime) { - lastTime = time; - lastValue = valueSelector.getLong(); - } - } - - @Override - public double getDouble() + void setCurrentValue() { - return (double) lastValue; + lastValue = valueSelector.getLong(); } @Override public Object get() { - return new SerializablePair<>(lastTime, lastValue); + return new SerializablePair<>(lastTime, rhsNull ? null : lastValue); } @Override @@ -69,14 +51,14 @@ public float getFloat() } @Override - public long getLong() + public double getDouble() { - return lastValue; + return (double) lastValue; } @Override - public void close() + public long getLong() { - + return lastValue; } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastAggregatorFactory.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastAggregatorFactory.java index 4a00a63dce7d..3cc333f9e86e 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastAggregatorFactory.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastAggregatorFactory.java @@ -30,11 +30,12 @@ import org.apache.druid.query.aggregation.AggregatorFactory; import org.apache.druid.query.aggregation.AggregatorUtil; import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.aggregation.NullableNumericAggregatorFactory; import org.apache.druid.query.aggregation.first.LongFirstAggregatorFactory; import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; +import org.apache.druid.segment.BaseLongColumnValueSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.ColumnValueSelector; +import org.apache.druid.segment.NilColumnValueSelector; import org.apache.druid.segment.column.ColumnHolder; import javax.annotation.Nullable; @@ -46,8 +47,32 @@ import java.util.Map; import java.util.Objects; -public class LongLastAggregatorFactory extends NullableNumericAggregatorFactory +public class LongLastAggregatorFactory extends AggregatorFactory { + private static final Aggregator NIL_AGGREGATOR = new LongLastAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate() + { + // no-op + } + }; + + private static final BufferAggregator NIL_BUFFER_AGGREGATOR = new LongLastBufferAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance() + ) + { + @Override + public void aggregate(ByteBuffer buf, int position) + { + // no-op + } + }; + private final String fieldName; private final String name; @@ -64,21 +89,31 @@ public LongLastAggregatorFactory( } @Override - protected ColumnValueSelector selector(ColumnSelectorFactory metricFactory) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { - return metricFactory.makeColumnValueSelector(fieldName); - } - - @Override - protected Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) - { - return new LongLastAggregator(metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), selector); + final BaseLongColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_AGGREGATOR; + } else { + return new LongLastAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override - protected BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { - return new LongLastBufferAggregator(metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), selector); + final BaseLongColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_BUFFER_AGGREGATOR; + } else { + return new LongLastBufferAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector + ); + } } @Override @@ -118,35 +153,52 @@ public AggregatorFactory getCombiningFactory() return new LongLastAggregatorFactory(name, name) { @Override - public Aggregator factorize(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public Aggregator factorize(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = metricFactory.makeColumnValueSelector(name); return new LongLastAggregator(null, null) { @Override public void aggregate() { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); if (pair.lhs >= lastTime) { lastTime = pair.lhs; - lastValue = pair.rhs; + if (pair.rhs != null) { + lastValue = pair.rhs; + rhsNull = false; + } else { + rhsNull = true; + } } } }; } @Override - public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, ColumnValueSelector selector) + public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { + final ColumnValueSelector> selector = metricFactory.makeColumnValueSelector(name); return new LongLastBufferAggregator(null, null) { + @Override + public void putValue(ByteBuffer buf, int position) + { + SerializablePair pair = selector.getObject(); + buf.putLong(position, pair.rhs); + } + @Override public void aggregate(ByteBuffer buf, int position) { - SerializablePair pair = (SerializablePair) selector.getObject(); + SerializablePair pair = selector.getObject(); long lastTime = buf.getLong(position); if (pair.lhs >= lastTime) { - buf.putLong(position, pair.lhs); - buf.putLong(position + Long.BYTES, pair.rhs); + if (pair.rhs != null) { + updateTimeWithValue(buf, position, pair.lhs); + } else { + updateTimeWithNull(buf, position, pair.lhs); + } } } @@ -170,6 +222,9 @@ public List getRequiredColumns() public Object deserialize(Object object) { Map map = (Map) object; + if (map.get("rhs") == null) { + return new SerializablePair<>(((Number) map.get("lhs")).longValue(), null); + } return new SerializablePair<>(((Number) map.get("lhs")).longValue(), ((Number) map.get("rhs")).longValue()); } @@ -213,13 +268,15 @@ public byte[] getCacheKey() @Override public String getTypeName() { + // if we don't pretend to be a primitive, group by v1 gets sad and doesn't work because no complex type serde return "long"; } @Override public int getMaxIntermediateSize() { - return Long.BYTES * 2; + // timestamp, is null, value + return Long.BYTES + Byte.BYTES + Long.BYTES; } @Override diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastBufferAggregator.java index 490c1e4c441d..981ba3e2f665 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastBufferAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/LongLastBufferAggregator.java @@ -20,75 +20,51 @@ package org.apache.druid.query.aggregation.last; import org.apache.druid.collections.SerializablePair; -import org.apache.druid.query.aggregation.BufferAggregator; -import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; import org.apache.druid.segment.BaseLongColumnValueSelector; import java.nio.ByteBuffer; -public class LongLastBufferAggregator implements BufferAggregator +public class LongLastBufferAggregator extends NumericLastBufferAggregator { - private final BaseLongColumnValueSelector timeSelector; - private final BaseLongColumnValueSelector valueSelector; - public LongLastBufferAggregator(BaseLongColumnValueSelector timeSelector, BaseLongColumnValueSelector valueSelector) { - this.timeSelector = timeSelector; - this.valueSelector = valueSelector; + super(timeSelector, valueSelector); } @Override - public void init(ByteBuffer buf, int position) + void initValue(ByteBuffer buf, int position) { - buf.putLong(position, Long.MIN_VALUE); - buf.putLong(position + Long.BYTES, 0); + buf.putLong(position, 0); } @Override - public void aggregate(ByteBuffer buf, int position) + void putValue(ByteBuffer buf, int position) { - long time = timeSelector.getLong(); - long lastTime = buf.getLong(position); - if (time >= lastTime) { - buf.putLong(position, time); - buf.putLong(position + Long.BYTES, valueSelector.getLong()); - } + buf.putLong(position, valueSelector.getLong()); } @Override public Object get(ByteBuffer buf, int position) { - return new SerializablePair<>(buf.getLong(position), buf.getLong(position + Long.BYTES)); + boolean rhsNull = isValueNull(buf, position); + return new SerializablePair<>(buf.getLong(position), rhsNull ? null : buf.getLong(position + VALUE_OFFSET)); } @Override public float getFloat(ByteBuffer buf, int position) { - return (float) buf.getLong(position + Long.BYTES); + return (float) buf.getLong(position + VALUE_OFFSET); } @Override public double getDouble(ByteBuffer buf, int position) { - return buf.getLong(position + Long.BYTES); + return buf.getLong(position + VALUE_OFFSET); } @Override public long getLong(ByteBuffer buf, int position) { - return buf.getLong(position + Long.BYTES); - } - - @Override - public void close() - { - // no resources to cleanup - } - - @Override - public void inspectRuntimeShape(RuntimeShapeInspector inspector) - { - inspector.visit("timeSelector", timeSelector); - inspector.visit("valueSelector", valueSelector); + return buf.getLong(position + VALUE_OFFSET); } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastAggregator.java new file mode 100644 index 000000000000..6506f976aeff --- /dev/null +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastAggregator.java @@ -0,0 +1,75 @@ +/* + * 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.last; + +import org.apache.druid.common.config.NullHandling; +import org.apache.druid.query.aggregation.Aggregator; +import org.apache.druid.segment.BaseLongColumnValueSelector; +import org.apache.druid.segment.BaseNullableColumnValueSelector; + +/** + * Base type for on heap 'last' aggregator for primitive numeric column selectors.. + * + * This could probably share a base class with {@link org.apache.druid.query.aggregation.first.NumericFirstAggregator} + */ +public abstract class NumericLastAggregator implements Aggregator +{ + private final boolean useDefault = NullHandling.replaceWithDefault(); + private final BaseLongColumnValueSelector timeSelector; + + final TSelector valueSelector; + long lastTime; + boolean rhsNull; + + public NumericLastAggregator(BaseLongColumnValueSelector timeSelector, TSelector valueSelector) + { + this.timeSelector = timeSelector; + this.valueSelector = valueSelector; + + lastTime = Long.MIN_VALUE; + rhsNull = !useDefault; + } + + @Override + public void aggregate() + { + long time = timeSelector.getLong(); + if (time >= lastTime) { + lastTime = time; + if (useDefault || !valueSelector.isNull()) { + setCurrentValue(); + rhsNull = false; + } else { + rhsNull = true; + } + } + } + + @Override + public void close() + { + // nothing to close + } + + /** + * Store the current primitive typed 'first' value + */ + abstract void setCurrentValue(); +} diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastBufferAggregator.java new file mode 100644 index 000000000000..7c90aadafb5f --- /dev/null +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/NumericLastBufferAggregator.java @@ -0,0 +1,115 @@ +/* + * 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.last; + +import org.apache.druid.common.config.NullHandling; +import org.apache.druid.query.aggregation.BufferAggregator; +import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector; +import org.apache.druid.segment.BaseLongColumnValueSelector; +import org.apache.druid.segment.BaseNullableColumnValueSelector; + +import java.nio.ByteBuffer; + +/** + * Base type for buffer based 'last' aggregator for primitive numeric column selectors + * + * This could probably share a base type with + * {@link org.apache.druid.query.aggregation.first.NumericFirstBufferAggregator} ... + */ +public abstract class NumericLastBufferAggregator + implements BufferAggregator +{ + static final int NULL_OFFSET = Long.BYTES; + static final int VALUE_OFFSET = NULL_OFFSET + Byte.BYTES; + + private final boolean useDefault = NullHandling.replaceWithDefault(); + private final BaseLongColumnValueSelector timeSelector; + + final TSelector valueSelector; + + public NumericLastBufferAggregator(BaseLongColumnValueSelector timeSelector, TSelector valueSelector) + { + this.timeSelector = timeSelector; + this.valueSelector = valueSelector; + } + + /** + * Initialize the buffer value at the position of {@link #VALUE_OFFSET} + */ + abstract void initValue(ByteBuffer buf, int position); + + /** + * Place the primitive value in the buffer at the position of {@link #VALUE_OFFSET} + */ + abstract void putValue(ByteBuffer buf, int position); + + boolean isValueNull(ByteBuffer buf, int position) + { + return buf.get(position + NULL_OFFSET) == NullHandling.IS_NULL_BYTE; + } + + void updateTimeWithValue(ByteBuffer buf, int position, long time) + { + buf.putLong(position, time); + buf.put(position + NULL_OFFSET, NullHandling.IS_NOT_NULL_BYTE); + putValue(buf, position + VALUE_OFFSET); + } + + void updateTimeWithNull(ByteBuffer buf, int position, long time) + { + buf.putLong(position, time); + buf.put(position + NULL_OFFSET, NullHandling.IS_NULL_BYTE); + } + + @Override + public void init(ByteBuffer buf, int position) + { + buf.putLong(position, Long.MIN_VALUE); + buf.put(position + NULL_OFFSET, useDefault ? NullHandling.IS_NOT_NULL_BYTE : NullHandling.IS_NULL_BYTE); + initValue(buf, position + VALUE_OFFSET); + } + + @Override + public void aggregate(ByteBuffer buf, int position) + { + long time = timeSelector.getLong(); + long lastTime = buf.getLong(position); + if (time >= lastTime) { + if (useDefault || !valueSelector.isNull()) { + updateTimeWithValue(buf, position, time); + } else { + updateTimeWithNull(buf, position, time); + } + } + } + + @Override + public void close() + { + // no resources to cleanup + } + + @Override + public void inspectRuntimeShape(RuntimeShapeInspector inspector) + { + inspector.visit("timeSelector", timeSelector); + inspector.visit("valueSelector", valueSelector); + } +} diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastAggregator.java index ea37ff420408..0c5fe331e9c7 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastAggregator.java @@ -65,7 +65,7 @@ public void aggregate() valueSelector ); - if (inPair != null && inPair.rhs != null && inPair.lhs >= lastTime) { + if (inPair != null && inPair.lhs >= lastTime) { lastTime = inPair.lhs; lastValue = StringUtils.fastLooseChop(inPair.rhs, maxStringBytes); } @@ -74,11 +74,8 @@ public void aggregate() if (time >= lastTime) { final String value = DimensionHandlerUtils.convertObjectToString(valueSelector.getObject()); - - if (value != null) { - lastTime = time; - lastValue = StringUtils.fastLooseChop(value, maxStringBytes); - } + lastTime = time; + lastValue = StringUtils.fastLooseChop(value, maxStringBytes); } } } diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastAggregatorFactory.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastAggregatorFactory.java index 9a3264f1fbae..e97f8b509781 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastAggregatorFactory.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastAggregatorFactory.java @@ -36,9 +36,11 @@ import org.apache.druid.query.cache.CacheKeyBuilder; import org.apache.druid.segment.BaseObjectColumnValueSelector; import org.apache.druid.segment.ColumnSelectorFactory; +import org.apache.druid.segment.NilColumnValueSelector; import org.apache.druid.segment.column.ColumnHolder; import javax.annotation.Nullable; +import java.nio.ByteBuffer; import java.util.Collections; import java.util.Comparator; import java.util.List; @@ -48,6 +50,34 @@ @JsonTypeName("stringLast") public class StringLastAggregatorFactory extends AggregatorFactory { + private static final Aggregator NIL_AGGREGATOR = new StringLastAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance(), + 0, + false + ) + { + @Override + public void aggregate() + { + // no-op + } + }; + + private static final BufferAggregator NIL_BUFFER_AGGREGATOR = new StringLastBufferAggregator( + NilColumnValueSelector.instance(), + NilColumnValueSelector.instance(), + 0, + false + ) + { + @Override + public void aggregate(ByteBuffer buf, int position) + { + // no-op + } + }; + private final String fieldName; private final String name; protected final int maxStringBytes; @@ -77,24 +107,32 @@ public StringLastAggregatorFactory( public Aggregator factorize(ColumnSelectorFactory metricFactory) { final BaseObjectColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); - return new StringLastAggregator( - metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), - valueSelector, - maxStringBytes, - StringFirstLastUtils.selectorNeedsFoldCheck(valueSelector, metricFactory.getColumnCapabilities(fieldName)) - ); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_AGGREGATOR; + } else { + return new StringLastAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector, + maxStringBytes, + StringFirstLastUtils.selectorNeedsFoldCheck(valueSelector, metricFactory.getColumnCapabilities(fieldName)) + ); + } } @Override public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory) { final BaseObjectColumnValueSelector valueSelector = metricFactory.makeColumnValueSelector(fieldName); - return new StringLastBufferAggregator( - metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), - valueSelector, - maxStringBytes, - StringFirstLastUtils.selectorNeedsFoldCheck(valueSelector, metricFactory.getColumnCapabilities(fieldName)) - ); + if (valueSelector instanceof NilColumnValueSelector) { + return NIL_BUFFER_AGGREGATOR; + } else { + return new StringLastBufferAggregator( + metricFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME), + valueSelector, + maxStringBytes, + StringFirstLastUtils.selectorNeedsFoldCheck(valueSelector, metricFactory.getColumnCapabilities(fieldName)) + ); + } } @Override diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastBufferAggregator.java b/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastBufferAggregator.java index 09e3276d3d56..9da985235472 100644 --- a/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastBufferAggregator.java +++ b/processing/src/main/java/org/apache/druid/query/aggregation/last/StringLastBufferAggregator.java @@ -72,7 +72,7 @@ public void aggregate(ByteBuffer buf, int position) valueSelector ); - if (inPair != null && inPair.rhs != null) { + if (inPair != null) { final long lastTime = buf.getLong(position); if (inPair.lhs >= lastTime) { StringFirstLastUtils.writePair( @@ -90,14 +90,12 @@ public void aggregate(ByteBuffer buf, int position) if (time >= lastTime) { final String value = DimensionHandlerUtils.convertObjectToString(valueSelector.getObject()); - if (value != null) { - StringFirstLastUtils.writePair( - buf, - position, - new SerializablePairLongString(time, value), - maxStringBytes - ); - } + StringFirstLastUtils.writePair( + buf, + position, + new SerializablePairLongString(time, value), + maxStringBytes + ); } } } diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregationTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregationTest.java index 19ceaa5b4e61..c85b6b11c779 100644 --- a/processing/src/test/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregationTest.java +++ b/processing/src/test/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregationTest.java @@ -30,14 +30,16 @@ import org.apache.druid.query.aggregation.TestObjectColumnSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.testing.InitializedNullHandlingTest; import org.easymock.EasyMock; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import java.nio.ByteBuffer; +import java.util.Comparator; -public class DoubleFirstAggregationTest +public class DoubleFirstAggregationTest extends InitializedNullHandlingTest { private DoubleFirstAggregatorFactory doubleFirstAggFactory; private DoubleFirstAggregatorFactory combiningAggFactory; @@ -118,6 +120,31 @@ public void testCombine() Assert.assertEquals(pair1, doubleFirstAggFactory.combine(pair1, pair2)); } + @Test + public void testComparator() + { + SerializablePair pair1 = new SerializablePair<>(1467225000L, 3.621); + SerializablePair pair2 = new SerializablePair<>(1467240000L, 785.4); + Comparator comparator = doubleFirstAggFactory.getComparator(); + Assert.assertEquals(-1, comparator.compare(pair1, pair2)); + Assert.assertEquals(0, comparator.compare(pair1, pair1)); + Assert.assertEquals(0, comparator.compare(pair2, pair2)); + Assert.assertEquals(1, comparator.compare(pair2, pair1)); + } + + @Test + public void testComparatorWithNulls() + { + SerializablePair pair1 = new SerializablePair<>(1467225000L, 3.621); + SerializablePair pair2 = new SerializablePair<>(1467240000L, null); + Comparator comparator = doubleFirstAggFactory.getComparator(); + Assert.assertEquals(1, comparator.compare(pair1, pair2)); + Assert.assertEquals(0, comparator.compare(pair1, pair1)); + Assert.assertEquals(0, comparator.compare(pair2, pair2)); + Assert.assertEquals(-1, comparator.compare(pair2, pair1)); + } + + @Test public void testDoubleFirstCombiningAggregator() { diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/first/FloatFirstAggregationTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/first/FloatFirstAggregationTest.java index 1e9721d83196..b9b37f83f8f7 100644 --- a/processing/src/test/java/org/apache/druid/query/aggregation/first/FloatFirstAggregationTest.java +++ b/processing/src/test/java/org/apache/druid/query/aggregation/first/FloatFirstAggregationTest.java @@ -30,14 +30,16 @@ import org.apache.druid.query.aggregation.TestObjectColumnSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.testing.InitializedNullHandlingTest; import org.easymock.EasyMock; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import java.nio.ByteBuffer; +import java.util.Comparator; -public class FloatFirstAggregationTest +public class FloatFirstAggregationTest extends InitializedNullHandlingTest { private FloatFirstAggregatorFactory floatFirstAggregatorFactory; private FloatFirstAggregatorFactory combiningAggFactory; @@ -65,8 +67,8 @@ public void setup() objectSelector = new TestObjectColumnSelector<>(pairs); colSelectorFactory = EasyMock.createMock(ColumnSelectorFactory.class); EasyMock.expect(colSelectorFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME)).andReturn(timeSelector); - EasyMock.expect(colSelectorFactory.makeColumnValueSelector("nilly")).andReturn(valueSelector); - EasyMock.expect(colSelectorFactory.makeColumnValueSelector("billy")).andReturn(objectSelector); + EasyMock.expect(colSelectorFactory.makeColumnValueSelector("nilly")).andReturn(valueSelector).atLeastOnce(); + EasyMock.expect(colSelectorFactory.makeColumnValueSelector("billy")).andReturn(objectSelector).atLeastOnce(); EasyMock.replay(colSelectorFactory); } @@ -113,11 +115,23 @@ public void testDoubleFirstBufferAggregator() @Test public void testCombine() { - SerializablePair pair1 = new SerializablePair<>(1467225000L, 3.621); - SerializablePair pair2 = new SerializablePair<>(1467240000L, 785.4); + SerializablePair pair1 = new SerializablePair<>(1467225000L, 3.621f); + SerializablePair pair2 = new SerializablePair<>(1467240000L, 785.4f); Assert.assertEquals(pair1, floatFirstAggregatorFactory.combine(pair1, pair2)); } + @Test + public void testComparatorWithNulls() + { + SerializablePair pair1 = new SerializablePair<>(1467225000L, 3.621f); + SerializablePair pair2 = new SerializablePair<>(1467240000L, null); + Comparator comparator = floatFirstAggregatorFactory.getComparator(); + Assert.assertEquals(1, comparator.compare(pair1, pair2)); + Assert.assertEquals(0, comparator.compare(pair1, pair1)); + Assert.assertEquals(0, comparator.compare(pair2, pair2)); + Assert.assertEquals(-1, comparator.compare(pair2, pair1)); + } + @Test public void testDoubleFirstCombiningAggregator() { diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/first/LongFirstAggregationTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/first/LongFirstAggregationTest.java index 4f3df593f13d..7fe925656e2b 100644 --- a/processing/src/test/java/org/apache/druid/query/aggregation/first/LongFirstAggregationTest.java +++ b/processing/src/test/java/org/apache/druid/query/aggregation/first/LongFirstAggregationTest.java @@ -29,14 +29,16 @@ import org.apache.druid.query.aggregation.TestObjectColumnSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.testing.InitializedNullHandlingTest; import org.easymock.EasyMock; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import java.nio.ByteBuffer; +import java.util.Comparator; -public class LongFirstAggregationTest +public class LongFirstAggregationTest extends InitializedNullHandlingTest { private LongFirstAggregatorFactory longFirstAggFactory; private LongFirstAggregatorFactory combiningAggFactory; @@ -117,6 +119,18 @@ public void testCombine() Assert.assertEquals(pair1, longFirstAggFactory.combine(pair1, pair2)); } + @Test + public void testComparatorWithNulls() + { + SerializablePair pair1 = new SerializablePair<>(1467225000L, 1263L); + SerializablePair pair2 = new SerializablePair<>(1467240000L, null); + Comparator comparator = longFirstAggFactory.getComparator(); + Assert.assertEquals(1, comparator.compare(pair1, pair2)); + Assert.assertEquals(0, comparator.compare(pair1, pair1)); + Assert.assertEquals(0, comparator.compare(pair2, pair2)); + Assert.assertEquals(-1, comparator.compare(pair2, pair1)); + } + @Test public void testLongFirstCombiningAggregator() { diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/first/StringFirstBufferAggregatorTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/first/StringFirstBufferAggregatorTest.java index 3b4ef692bf52..8d88677bb55b 100644 --- a/processing/src/test/java/org/apache/druid/query/aggregation/first/StringFirstBufferAggregatorTest.java +++ b/processing/src/test/java/org/apache/druid/query/aggregation/first/StringFirstBufferAggregatorTest.java @@ -189,7 +189,7 @@ public void testNoStringValue() SerializablePairLongString sp = ((SerializablePairLongString) agg.get(buf, position)); - Assert.assertEquals(1526724600L, (long) sp.lhs); - Assert.assertEquals("2.0", sp.rhs); + Assert.assertEquals(1526724000L, (long) sp.lhs); + Assert.assertEquals(null, sp.rhs); } } diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/first/StringFirstTimeseriesQueryTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/first/StringFirstTimeseriesQueryTest.java index 3ae838442fb2..445c6c9af419 100644 --- a/processing/src/test/java/org/apache/druid/query/aggregation/first/StringFirstTimeseriesQueryTest.java +++ b/processing/src/test/java/org/apache/druid/query/aggregation/first/StringFirstTimeseriesQueryTest.java @@ -43,6 +43,7 @@ import org.apache.druid.segment.incremental.IncrementalIndexStorageAdapter; import org.apache.druid.segment.incremental.IndexSizeExceededException; import org.apache.druid.segment.serde.ComplexMetrics; +import org.apache.druid.testing.InitializedNullHandlingTest; import org.joda.time.DateTime; import org.junit.Before; import org.junit.Test; @@ -50,7 +51,7 @@ import java.util.Collections; import java.util.List; -public class StringFirstTimeseriesQueryTest +public class StringFirstTimeseriesQueryTest extends InitializedNullHandlingTest { private static final String VISITOR_ID = "visitor_id"; private static final String CLIENT_TYPE = "client_type"; diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/last/DoubleLastAggregationTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/last/DoubleLastAggregationTest.java index 21d6cada8194..847194f12760 100644 --- a/processing/src/test/java/org/apache/druid/query/aggregation/last/DoubleLastAggregationTest.java +++ b/processing/src/test/java/org/apache/druid/query/aggregation/last/DoubleLastAggregationTest.java @@ -30,14 +30,16 @@ import org.apache.druid.query.aggregation.TestObjectColumnSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.testing.InitializedNullHandlingTest; import org.easymock.EasyMock; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import java.nio.ByteBuffer; +import java.util.Comparator; -public class DoubleLastAggregationTest +public class DoubleLastAggregationTest extends InitializedNullHandlingTest { private DoubleLastAggregatorFactory doubleLastAggFactory; private DoubleLastAggregatorFactory combiningAggFactory; @@ -118,6 +120,18 @@ public void testCombine() Assert.assertEquals(pair2, doubleLastAggFactory.combine(pair1, pair2)); } + @Test + public void testComparatorWithNulls() + { + SerializablePair pair1 = new SerializablePair<>(1467225000L, 3.621); + SerializablePair pair2 = new SerializablePair<>(1467240000L, null); + Comparator comparator = doubleLastAggFactory.getComparator(); + Assert.assertEquals(1, comparator.compare(pair1, pair2)); + Assert.assertEquals(0, comparator.compare(pair1, pair1)); + Assert.assertEquals(0, comparator.compare(pair2, pair2)); + Assert.assertEquals(-1, comparator.compare(pair2, pair1)); + } + @Test public void testDoubleLastCombiningAggregator() { diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/last/FloatLastAggregationTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/last/FloatLastAggregationTest.java index f094ecf24e67..86b6a998b8e3 100644 --- a/processing/src/test/java/org/apache/druid/query/aggregation/last/FloatLastAggregationTest.java +++ b/processing/src/test/java/org/apache/druid/query/aggregation/last/FloatLastAggregationTest.java @@ -30,14 +30,16 @@ import org.apache.druid.query.aggregation.TestObjectColumnSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.testing.InitializedNullHandlingTest; import org.easymock.EasyMock; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import java.nio.ByteBuffer; +import java.util.Comparator; -public class FloatLastAggregationTest +public class FloatLastAggregationTest extends InitializedNullHandlingTest { private FloatLastAggregatorFactory floatLastAggregatorFactory; private FloatLastAggregatorFactory combiningAggFactory; @@ -113,11 +115,23 @@ public void testDoubleLastBufferAggregator() @Test public void testCombine() { - SerializablePair pair1 = new SerializablePair<>(1467225000L, 3.621); - SerializablePair pair2 = new SerializablePair<>(1467240000L, 785.4); + SerializablePair pair1 = new SerializablePair<>(1467225000L, 3.621f); + SerializablePair pair2 = new SerializablePair<>(1467240000L, 785.4f); Assert.assertEquals(pair2, floatLastAggregatorFactory.combine(pair1, pair2)); } + @Test + public void testComparatorWithNulls() + { + SerializablePair pair1 = new SerializablePair<>(1467225000L, 3.621f); + SerializablePair pair2 = new SerializablePair<>(1467240000L, null); + Comparator comparator = floatLastAggregatorFactory.getComparator(); + Assert.assertEquals(1, comparator.compare(pair1, pair2)); + Assert.assertEquals(0, comparator.compare(pair1, pair1)); + Assert.assertEquals(0, comparator.compare(pair2, pair2)); + Assert.assertEquals(-1, comparator.compare(pair2, pair1)); + } + @Test public void testDoubleLastCombiningAggregator() { diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/last/LongLastAggregationTest.java b/processing/src/test/java/org/apache/druid/query/aggregation/last/LongLastAggregationTest.java index 1324bee261de..a9b2fadcc90e 100644 --- a/processing/src/test/java/org/apache/druid/query/aggregation/last/LongLastAggregationTest.java +++ b/processing/src/test/java/org/apache/druid/query/aggregation/last/LongLastAggregationTest.java @@ -29,14 +29,16 @@ import org.apache.druid.query.aggregation.TestObjectColumnSelector; import org.apache.druid.segment.ColumnSelectorFactory; import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.testing.InitializedNullHandlingTest; import org.easymock.EasyMock; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import java.nio.ByteBuffer; +import java.util.Comparator; -public class LongLastAggregationTest +public class LongLastAggregationTest extends InitializedNullHandlingTest { private LongLastAggregatorFactory longLastAggFactory; private LongLastAggregatorFactory combiningAggFactory; @@ -117,6 +119,18 @@ public void testCombine() Assert.assertEquals(pair2, longLastAggFactory.combine(pair1, pair2)); } + @Test + public void testComparatorWithNulls() + { + SerializablePair pair1 = new SerializablePair<>(1467225000L, 1263L); + SerializablePair pair2 = new SerializablePair<>(1467240000L, null); + Comparator comparator = longLastAggFactory.getComparator(); + Assert.assertEquals(1, comparator.compare(pair1, pair2)); + Assert.assertEquals(0, comparator.compare(pair1, pair1)); + Assert.assertEquals(0, comparator.compare(pair2, pair2)); + Assert.assertEquals(-1, comparator.compare(pair2, pair1)); + } + @Test public void testLongLastCombiningAggregator() { diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java index 1cc216171fb8..a94ef2bdb3dc 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java @@ -1254,7 +1254,7 @@ public void testEarliestAggregators() throws Exception .build() ), ImmutableList.of( - new Object[]{1L, 1.0f, NullHandling.sqlCompatible() ? "" : "10.1", 2L, 2.0f, "1"} + new Object[]{1L, 1.0f, "", 2L, 2.0f, "1"} ) ); } @@ -1465,11 +1465,406 @@ public void testStringEarliestInSubquery() throws Exception .build() ), ImmutableList.of( - new Object[]{NullHandling.sqlCompatible() ? 12.1 : 11.1} + // default mode subquery results: + //[, 10.1] + //[a, ] + //[abc, def] + // sql compatible mode subquery results: + //[null, 10.1] + //[, 2] + //[a, ] + //[abc, def] + new Object[]{NullHandling.sqlCompatible() ? 12.1 : 10.1} ) ); } + @Test + public void testEarliestAggregatorsNumericNulls() throws Exception + { + // Cannot vectorize EARLIEST aggregator. + skipVectorize(); + + testQuery( + "SELECT EARLIEST(l1), EARLIEST(d1), EARLIEST(f1) FROM druid.numfoo", + ImmutableList.of( + Druids.newTimeseriesQueryBuilder() + .dataSource(CalciteTests.DATASOURCE3) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .aggregators( + aggregators( + new LongFirstAggregatorFactory("a0", "l1"), + new DoubleFirstAggregatorFactory("a1", "d1"), + new FloatFirstAggregatorFactory("a2", "f1") + ) + ) + .context(TIMESERIES_CONTEXT_DEFAULT) + .build() + ), + ImmutableList.of( + new Object[]{7L, 1.0, 1.0f} + ) + ); + } + + @Test + public void testLatestAggregatorsNumericNull() throws Exception + { + // Cannot vectorize LATEST aggregator. + skipVectorize(); + + testQuery( + "SELECT LATEST(l1), LATEST(d1), LATEST(f1) FROM druid.numfoo", + ImmutableList.of( + Druids.newTimeseriesQueryBuilder() + .dataSource(CalciteTests.DATASOURCE3) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .aggregators( + aggregators( + new LongLastAggregatorFactory("a0", "l1"), + new DoubleLastAggregatorFactory("a1", "d1"), + new FloatLastAggregatorFactory("a2", "f1") + ) + ) + .context(TIMESERIES_CONTEXT_DEFAULT) + .build() + ), + ImmutableList.of( + new Object[]{NullHandling.defaultLongValue(), NullHandling.defaultDoubleValue(), NullHandling.defaultFloatValue()} + ) + ); + } + + @Test + public void testFirstLatestAggregatorsSkipNulls() throws Exception + { + // Cannot vectorize LATEST aggregator. + skipVectorize(); + + final DimFilter filter; + if (useDefault) { + filter = not(selector("dim1", null, null)); + } else { + filter = and( + not(selector("dim1", null, null)), + not(selector("l1", null, null)), + not(selector("d1", null, null)), + not(selector("f1", null, null)) + ); + } + testQuery( + "SELECT EARLIEST(dim1, 32), LATEST(l1), LATEST(d1), LATEST(f1) " + + "FROM druid.numfoo " + + "WHERE dim1 IS NOT NULL AND l1 IS NOT NULL AND d1 IS NOT NULL AND f1 is NOT NULL", + ImmutableList.of( + Druids.newTimeseriesQueryBuilder() + .dataSource(CalciteTests.DATASOURCE3) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .filters(filter) + .aggregators( + aggregators( + new StringFirstAggregatorFactory("a0", "dim1", 32), + new LongLastAggregatorFactory("a1", "l1"), + new DoubleLastAggregatorFactory("a2", "d1"), + new FloatLastAggregatorFactory("a3", "f1") + ) + ) + .context(TIMESERIES_CONTEXT_DEFAULT) + .build() + ), + ImmutableList.of( + // first row of dim1 is empty string, which is null in default mode, last non-null numeric rows are zeros + new Object[]{useDefault ? "10.1" : "", 0L, 0.0, 0.0f} + ) + ); + } + + @Test + public void testOrderByEarliestFloat() throws Exception + { + // Cannot vectorize EARLIEST aggregator. + skipVectorize(); + List expected; + if (NullHandling.replaceWithDefault()) { + expected = ImmutableList.of( + new Object[]{"1", 0.0f}, + new Object[]{"2", 0.0f}, + new Object[]{"abc", 0.0f}, + new Object[]{"def", 0.0f}, + new Object[]{"10.1", 0.1f}, + new Object[]{"", 1.0f} + ); + } else { + expected = ImmutableList.of( + new Object[]{"1", null}, + new Object[]{"abc", null}, + new Object[]{"def", null}, + new Object[]{"2", 0.0f}, + new Object[]{"10.1", 0.1f}, + new Object[]{"", 1.0f} + ); + } + testQuery( + "SELECT dim1, EARLIEST(f1) FROM druid.numfoo GROUP BY 1 ORDER BY 2 LIMIT 10", + ImmutableList.of( + new TopNQueryBuilder() + .dataSource(CalciteTests.DATASOURCE3) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .dimension(new DefaultDimensionSpec("dim1", "_d0")) + .aggregators( + aggregators( + new FloatFirstAggregatorFactory("a0", "f1") + ) + ) + .metric(new InvertedTopNMetricSpec(new NumericTopNMetricSpec("a0"))) + .threshold(10) + .context(QUERY_CONTEXT_DEFAULT) + .build() + ), + expected + ); + } + + @Test + public void testOrderByEarliestDouble() throws Exception + { + // Cannot vectorize EARLIEST aggregator. + skipVectorize(); + List expected; + if (NullHandling.replaceWithDefault()) { + expected = ImmutableList.of( + new Object[]{"1", 0.0}, + new Object[]{"2", 0.0}, + new Object[]{"abc", 0.0}, + new Object[]{"def", 0.0}, + new Object[]{"", 1.0}, + new Object[]{"10.1", 1.7} + ); + } else { + expected = ImmutableList.of( + new Object[]{"1", null}, + new Object[]{"abc", null}, + new Object[]{"def", null}, + new Object[]{"2", 0.0}, + new Object[]{"", 1.0}, + new Object[]{"10.1", 1.7} + ); + } + testQuery( + "SELECT dim1, EARLIEST(d1) FROM druid.numfoo GROUP BY 1 ORDER BY 2 LIMIT 10", + ImmutableList.of( + new TopNQueryBuilder() + .dataSource(CalciteTests.DATASOURCE3) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .dimension(new DefaultDimensionSpec("dim1", "_d0")) + .aggregators( + aggregators( + new DoubleFirstAggregatorFactory("a0", "d1") + ) + ) + .metric(new InvertedTopNMetricSpec(new NumericTopNMetricSpec("a0"))) + .threshold(10) + .context(QUERY_CONTEXT_DEFAULT) + .build() + ), + expected + ); + } + + @Test + public void testOrderByEarliestLong() throws Exception + { + // Cannot vectorize EARLIEST aggregator. + skipVectorize(); + List expected; + if (NullHandling.replaceWithDefault()) { + expected = ImmutableList.of( + new Object[]{"1", 0L}, + new Object[]{"2", 0L}, + new Object[]{"abc", 0L}, + new Object[]{"def", 0L}, + new Object[]{"", 7L}, + new Object[]{"10.1", 325323L} + ); + } else { + expected = ImmutableList.of( + new Object[]{"1", null}, + new Object[]{"abc", null}, + new Object[]{"def", null}, + new Object[]{"2", 0L}, + new Object[]{"", 7L}, + new Object[]{"10.1", 325323L} + ); + } + testQuery( + "SELECT dim1, EARLIEST(l1) FROM druid.numfoo GROUP BY 1 ORDER BY 2 LIMIT 10", + ImmutableList.of( + new TopNQueryBuilder() + .dataSource(CalciteTests.DATASOURCE3) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .dimension(new DefaultDimensionSpec("dim1", "_d0")) + .aggregators( + aggregators( + new LongFirstAggregatorFactory("a0", "l1") + ) + ) + .metric(new InvertedTopNMetricSpec(new NumericTopNMetricSpec("a0"))) + .threshold(10) + .context(QUERY_CONTEXT_DEFAULT) + .build() + ), + expected + ); + } + + @Test + public void testOrderByLatestFloat() throws Exception + { + // Cannot vectorize LATEST aggregator. + skipVectorize(); + List expected; + if (NullHandling.replaceWithDefault()) { + expected = ImmutableList.of( + new Object[]{"1", 0.0f}, + new Object[]{"2", 0.0f}, + new Object[]{"abc", 0.0f}, + new Object[]{"def", 0.0f}, + new Object[]{"10.1", 0.1f}, + new Object[]{"", 1.0f} + ); + } else { + expected = ImmutableList.of( + new Object[]{"1", null}, + new Object[]{"abc", null}, + new Object[]{"def", null}, + new Object[]{"2", 0.0f}, + new Object[]{"10.1", 0.1f}, + new Object[]{"", 1.0f} + ); + } + + testQuery( + "SELECT dim1, LATEST(f1) FROM druid.numfoo GROUP BY 1 ORDER BY 2 LIMIT 10", + ImmutableList.of( + new TopNQueryBuilder() + .dataSource(CalciteTests.DATASOURCE3) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .dimension(new DefaultDimensionSpec("dim1", "_d0")) + .aggregators( + aggregators( + new FloatLastAggregatorFactory("a0", "f1") + ) + ) + .metric(new InvertedTopNMetricSpec(new NumericTopNMetricSpec("a0"))) + .threshold(10) + .context(QUERY_CONTEXT_DEFAULT) + .build() + ), + expected + ); + } + + @Test + public void testOrderByLatestDouble() throws Exception + { + // Cannot vectorize LATEST aggregator. + skipVectorize(); + List expected; + if (NullHandling.replaceWithDefault()) { + expected = ImmutableList.of( + new Object[]{"1", 0.0}, + new Object[]{"2", 0.0}, + new Object[]{"abc", 0.0}, + new Object[]{"def", 0.0}, + new Object[]{"", 1.0}, + new Object[]{"10.1", 1.7} + ); + } else { + expected = ImmutableList.of( + new Object[]{"1", null}, + new Object[]{"abc", null}, + new Object[]{"def", null}, + new Object[]{"2", 0.0}, + new Object[]{"", 1.0}, + new Object[]{"10.1", 1.7} + ); + } + testQuery( + "SELECT dim1, LATEST(d1) FROM druid.numfoo GROUP BY 1 ORDER BY 2 LIMIT 10", + ImmutableList.of( + new TopNQueryBuilder() + .dataSource(CalciteTests.DATASOURCE3) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .dimension(new DefaultDimensionSpec("dim1", "_d0")) + .aggregators( + aggregators( + new DoubleLastAggregatorFactory("a0", "d1") + ) + ) + .metric(new InvertedTopNMetricSpec(new NumericTopNMetricSpec("a0"))) + .threshold(10) + .context(QUERY_CONTEXT_DEFAULT) + .build() + ), + expected + ); + } + + @Test + public void testOrderByLatestLong() throws Exception + { + // Cannot vectorize LATEST aggregator. + skipVectorize(); + List expected; + if (NullHandling.replaceWithDefault()) { + expected = ImmutableList.of( + new Object[]{"1", 0L}, + new Object[]{"2", 0L}, + new Object[]{"abc", 0L}, + new Object[]{"def", 0L}, + new Object[]{"", 7L}, + new Object[]{"10.1", 325323L} + ); + } else { + expected = ImmutableList.of( + new Object[]{"1", null}, + new Object[]{"abc", null}, + new Object[]{"def", null}, + new Object[]{"2", 0L}, + new Object[]{"", 7L}, + new Object[]{"10.1", 325323L} + ); + } + testQuery( + "SELECT dim1, LATEST(l1) FROM druid.numfoo GROUP BY 1 ORDER BY 2 LIMIT 10", + ImmutableList.of( + new TopNQueryBuilder() + .dataSource(CalciteTests.DATASOURCE3) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .dimension(new DefaultDimensionSpec("dim1", "_d0")) + .aggregators( + aggregators( + new LongLastAggregatorFactory("a0", "l1") + ) + ) + .metric(new InvertedTopNMetricSpec(new NumericTopNMetricSpec("a0"))) + .threshold(10) + .context(QUERY_CONTEXT_DEFAULT) + .build() + ), + expected + ); + } + @Test public void testGroupByLong() throws Exception {