clintropolis commented on code in PR #14408:
URL: https://github.com/apache/druid/pull/14408#discussion_r1280072841
##########
processing/src/main/java/org/apache/druid/query/aggregation/first/DoubleFirstAggregatorFactory.java:
##########
@@ -125,6 +137,21 @@ public BufferAggregator
factorizeBuffered(ColumnSelectorFactory metricFactory)
}
}
+ @Override
+ public VectorAggregator factorizeVector(
+ VectorColumnSelectorFactory columnSelectorFactory
+ )
+ {
+ ColumnCapabilities capabilities =
columnSelectorFactory.getColumnCapabilities(fieldName);
+ if (capabilities != null && capabilities.isNumeric()) {
Review Comment:
nit: can use `Types.isNumeric(capabilities)`
https://github.com/apache/druid/blob/master/processing/src/main/java/org/apache/druid/segment/column/Types.java#L120
##########
processing/src/main/java/org/apache/druid/query/aggregation/first/LongFirstAggregatorFactory.java:
##########
@@ -122,6 +128,25 @@ public BufferAggregator
factorizeBuffered(ColumnSelectorFactory metricFactory)
}
}
+ @Override
+ public VectorAggregator factorizeVector(VectorColumnSelectorFactory
columnSelectorFactory)
+ {
+ ColumnCapabilities capabilities =
columnSelectorFactory.getColumnCapabilities(fieldName);
+ if (capabilities != null && capabilities.isNumeric()) {
Review Comment:
nit: can use `Types.isNumeric`
##########
processing/src/main/java/org/apache/druid/query/aggregation/first/FloatFirstAggregatorFactory.java:
##########
@@ -123,6 +129,24 @@ public BufferAggregator
factorizeBuffered(ColumnSelectorFactory metricFactory)
}
}
+ @Override
+ public VectorAggregator factorizeVector(VectorColumnSelectorFactory
columnSelectorFactory)
+ {
+ ColumnCapabilities capabilities =
columnSelectorFactory.getColumnCapabilities(fieldName);
+ if (capabilities != null && capabilities.isNumeric()) {
Review Comment:
nit: can use `Types.isNumeric`
##########
processing/src/test/java/org/apache/druid/query/aggregation/first/DoubleFirstVectorAggregationTest.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.java.util.common.Pair;
+import org.apache.druid.query.aggregation.VectorAggregator;
+import org.apache.druid.query.dimension.DimensionSpec;
+import org.apache.druid.segment.column.ColumnCapabilities;
+import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.vector.BaseDoubleVectorValueSelector;
+import org.apache.druid.segment.vector.BaseLongVectorValueSelector;
+import org.apache.druid.segment.vector.MultiValueDimensionVectorSelector;
+import org.apache.druid.segment.vector.NoFilterVectorOffset;
+import org.apache.druid.segment.vector.ReadableVectorInspector;
+import org.apache.druid.segment.vector.SingleValueDimensionVectorSelector;
+import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
+import org.apache.druid.segment.vector.VectorObjectSelector;
+import org.apache.druid.segment.vector.VectorValueSelector;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+import java.util.concurrent.ThreadLocalRandom;
+
+public class DoubleFirstVectorAggregationTest extends
InitializedNullHandlingTest
+{
+ private static final double EPSILON = 1e-5;
+ private static final double[] VALUES = new double[]{7.8d, 11, 23.67, 60};
+ private static final boolean[] NULLS = new boolean[]{false, false, true,
false};
+ private long[] times = {2436, 6879, 7888, 8224};
+
+ private static final String NAME = "NAME";
+ private static final String FIELD_NAME = "FIELD_NAME";
+ private static final String TIME_COL = "__time";
+
+ private VectorValueSelector selector;
+
+ private BaseLongVectorValueSelector timeSelector;
+ private ByteBuffer buf;
+
+ private DoubleFirstVectorAggregator target;
+
+ private DoubleFirstAggregatorFactory doubleFirstAggregatorFactory;
+
+ private VectorColumnSelectorFactory selectorFactory;
+
+ @Before
+ public void setup()
+ {
+ byte[] randomBytes = new byte[1024];
+ ThreadLocalRandom.current().nextBytes(randomBytes);
+ buf = ByteBuffer.wrap(randomBytes);
+ timeSelector = new BaseLongVectorValueSelector(new
NoFilterVectorOffset(times.length, 0, times.length)
+ {
+ })
+ {
+ @Override
+ public long[] getLongVector()
+ {
+ return times;
+ }
+
+ @Nullable
+ @Override
+ public boolean[] getNullVector()
+ {
+ return NULLS;
Review Comment:
nit: time column will never have null values afaik
##########
processing/src/main/java/org/apache/druid/query/aggregation/first/StringFirstVectorAggregator.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.java.util.common.DateTimes;
+import org.apache.druid.query.aggregation.SerializablePairLongString;
+import org.apache.druid.query.aggregation.VectorAggregator;
+import org.apache.druid.segment.DimensionHandlerUtils;
+import org.apache.druid.segment.vector.BaseLongVectorValueSelector;
+import org.apache.druid.segment.vector.VectorObjectSelector;
+
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+
+public class StringFirstVectorAggregator implements VectorAggregator
+{
+ private static final SerializablePairLongString INIT = new
SerializablePairLongString(
+ DateTimes.MAX.getMillis(),
+ null
+ );
+ private final BaseLongVectorValueSelector timeSelector;
+ private final VectorObjectSelector valueSelector;
+ private final int maxStringBytes;
+
+
+ public StringFirstVectorAggregator(
+ BaseLongVectorValueSelector timeSelector,
+ VectorObjectSelector valueSelector,
+ int maxStringBytes
+ )
+ {
+ this.timeSelector = timeSelector;
+ this.valueSelector = valueSelector;
+ this.maxStringBytes = maxStringBytes;
+ }
+
+ @Override
+ public void init(ByteBuffer buf, int position)
+ {
+ StringFirstLastUtils.writePair(buf, position, INIT, maxStringBytes);
+ }
+
+ @Override
+ public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
+ {
+ if (timeSelector == null) {
+ return;
+ }
+ long[] times = timeSelector.getLongVector();
+ Object[] objectsWhichMightBeStrings = valueSelector.getObjectVector();
+ long firstTime = buf.getLong(position);
+ int index;
+ for (int i = startRow; i < endRow; i++) {
+ if (times[i] > firstTime) {
+ break;
+ }
+ index = i;
+ final boolean foldNeeded =
StringFirstLastUtils.objectNeedsFoldCheck(objectsWhichMightBeStrings[index]);
+ if (foldNeeded) {
+ final SerializablePairLongString inPair =
StringFirstLastUtils.readPairFromVectorSelectorsAtIndex(
+ timeSelector,
+ valueSelector,
+ index
+ );
+ if (inPair != null) {
+ firstTime = buf.getLong(position);
+ if (inPair.lhs < firstTime) {
+ StringFirstLastUtils.writePair(
+ buf,
+ position,
+ new SerializablePairLongString(inPair.lhs, inPair.rhs),
+ maxStringBytes
+ );
+ }
+ }
+ } else {
+ final long time = times[index];
+ if (time < firstTime) {
+ final String value =
DimensionHandlerUtils.convertObjectToString(objectsWhichMightBeStrings[index]);
+ firstTime = time;
+ StringFirstLastUtils.writePair(
+ buf,
+ position,
+ new SerializablePairLongString(time, value),
+ maxStringBytes
+ );
+ }
+ }
+ }
+
+ }
+
+ @Override
+ public void aggregate(ByteBuffer buf, int numRows, int[] positions,
@Nullable int[] rows, int positionOffset)
+ {
+ long[] timeVector = timeSelector.getLongVector();
+ Object[] objectsWhichMightBeStrings = valueSelector.getObjectVector();
+
+ // iterate once over the object vector to find first non null element and
+ // determine if the type is Pair or not
+ boolean foldNeeded = false;
+ for (Object obj : objectsWhichMightBeStrings) {
Review Comment:
i forget, are selectors that spit out `SerializablePairLongString` always
not null values? if not, do we need to check that we actually found something
inside of the loop that wasn't null? im thinking of the case of when the column
is sparse and has lots of nulls, and the whole vector for this aggregate call
is all nulls. I guess it ends up in the not-folding case, which is ok if the
serializable pairs are never null
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
