This is an automated email from the ASF dual-hosted git repository.
clintropolis pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 4bf97a82b5f feat: simd min/max vector aggregators (#19660)
4bf97a82b5f is described below
commit 4bf97a82b5fdc70e1993e7060651b1c9279cfb8b
Author: Clint Wylie <[email protected]>
AuthorDate: Tue Jul 14 18:28:25 2026 -0700
feat: simd min/max vector aggregators (#19660)
---
.../benchmark/query/SqlExpressionBenchmark.java | 16 +-
.../aggregation/DoubleMaxAggregatorFactory.java | 5 +
.../aggregation/DoubleMinAggregatorFactory.java | 5 +
.../aggregation/FloatMaxAggregatorFactory.java | 5 +
.../aggregation/FloatMinAggregatorFactory.java | 5 +
.../aggregation/LongMaxAggregatorFactory.java | 5 +
.../aggregation/LongMinAggregatorFactory.java | 5 +
.../simd/SimdDoubleMaxVectorAggregator.java | 97 +++++++
.../simd/SimdDoubleMinVectorAggregator.java | 97 +++++++
.../simd/SimdFloatMaxVectorAggregator.java | 97 +++++++
.../simd/SimdFloatMinVectorAggregator.java | 97 +++++++
.../simd/SimdLongMaxVectorAggregator.java | 97 +++++++
.../simd/SimdLongMinVectorAggregator.java | 97 +++++++
.../simd/SimdAggregatorTestHelpers.java | 191 ++++++++++++
.../simd/SimdMinMaxVectorAggregatorTest.java | 319 +++++++++++++++++++++
.../simd/SimdSumVectorAggregatorTest.java | 148 +---------
...lVectorizedExpressionResultConsistencyTest.java | 4 +-
17 files changed, 1145 insertions(+), 145 deletions(-)
diff --git
a/benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlExpressionBenchmark.java
b/benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlExpressionBenchmark.java
index 1c2b3c723f5..28a3ac8672a 100644
---
a/benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlExpressionBenchmark.java
+++
b/benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlExpressionBenchmark.java
@@ -160,10 +160,17 @@ public class SqlExpressionBenchmark extends
SqlBaseQueryBenchmark
"SELECT NVL(long1, long3), SUM(double1) FROM expressions GROUP BY 1
ORDER BY 2",
"SELECT NVL(long1, long5 + long3), SUM(double1) FROM expressions GROUP
BY 1 ORDER BY 2",
"SELECT CASE WHEN MOD(long1, 2) = 0 THEN -1 WHEN MOD(long1, 2) = 1 THEN
long2 / MOD(long1, 2) ELSE long3 END FROM expressions GROUP BY 1",
- // cast
+ // 59-61 cast
"SELECT CAST(string1 as BIGINT) + CAST(string3 as DOUBLE) + long3,
COUNT(*) FROM expressions GROUP BY 1 ORDER BY 2",
"SELECT COUNT(*), SUM(CAST(string1 as BIGINT) + CAST(string3 as BIGINT))
FROM expressions WHERE double3 < 1010.0 AND double3 > 100.0",
- "SELECT COUNT(*) FROM expressions WHERE __time >= TIMESTAMP '2000-01-01
00:00:00' AND __time < TIMESTAMP '2000-01-02 00:00:00' AND
(UPPER(COALESCE(string3,'')) LIKE '1%' OR TRIM(UPPER(COALESCE(string3,'')))
LIKE '1%' OR SUBSTRING(UPPER(COALESCE(string3,'')),1,1) IN
('1','2','3','4','5') OR ('X' || UPPER(COALESCE(string3,''))) LIKE 'X1%') AND
(UPPER(COALESCE(string5,'')) LIKE '2%' OR TRIM(UPPER(COALESCE(string5,'')))
LIKE '2%' OR SUBSTRING(UPPER(COALESCE(string5,'')),1,1) IN ('1','2 [...]
+ "SELECT COUNT(*) FROM expressions WHERE __time >= TIMESTAMP '2000-01-01
00:00:00' AND __time < TIMESTAMP '2000-01-02 00:00:00' AND
(UPPER(COALESCE(string3,'')) LIKE '1%' OR TRIM(UPPER(COALESCE(string3,'')))
LIKE '1%' OR SUBSTRING(UPPER(COALESCE(string3,'')),1,1) IN
('1','2','3','4','5') OR ('X' || UPPER(COALESCE(string3,''))) LIKE 'X1%') AND
(UPPER(COALESCE(string5,'')) LIKE '2%' OR TRIM(UPPER(COALESCE(string5,'')))
LIKE '2%' OR SUBSTRING(UPPER(COALESCE(string5,'')),1,1) IN ('1','2 [...]
+ // ===========================
+ // more non-expression reference queries: min/max aggregators
+ // ===========================
+ // 62: min/max on a nullable long column (exercises the null-aware SIMD
path)
+ "SELECT MIN(long5), MAX(long5) FROM expressions",
+ // 63: min/max across long/double/float, non-null columns
+ "SELECT MIN(long1), MAX(long1), MIN(double1), MAX(double1), MIN(float3),
MAX(float3) FROM expressions"
);
@Param({
@@ -240,7 +247,10 @@ public class SqlExpressionBenchmark extends
SqlBaseQueryBenchmark
"57",
"58",
"59",
- "60"
+ "60",
+ "61",
+ "62",
+ "63"
})
private String query;
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/DoubleMaxAggregatorFactory.java
b/processing/src/main/java/org/apache/druid/query/aggregation/DoubleMaxAggregatorFactory.java
index a4b2ea87e63..6550b024690 100644
---
a/processing/src/main/java/org/apache/druid/query/aggregation/DoubleMaxAggregatorFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/DoubleMaxAggregatorFactory.java
@@ -24,6 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Supplier;
import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.math.expr.ExpressionProcessing;
+import org.apache.druid.query.aggregation.simd.SimdDoubleMaxVectorAggregator;
import org.apache.druid.segment.BaseDoubleColumnValueSelector;
import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
import org.apache.druid.segment.vector.VectorValueSelector;
@@ -81,6 +83,9 @@ public class DoubleMaxAggregatorFactory extends
SimpleDoubleAggregatorFactory
VectorValueSelector selector
)
{
+ if (ExpressionProcessing.useVectorApi()) {
+ return new SimdDoubleMaxVectorAggregator(selector);
+ }
return new DoubleMaxVectorAggregator(selector);
}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/DoubleMinAggregatorFactory.java
b/processing/src/main/java/org/apache/druid/query/aggregation/DoubleMinAggregatorFactory.java
index 241d7911af6..cb95d52ab69 100644
---
a/processing/src/main/java/org/apache/druid/query/aggregation/DoubleMinAggregatorFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/DoubleMinAggregatorFactory.java
@@ -24,6 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Supplier;
import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.math.expr.ExpressionProcessing;
+import org.apache.druid.query.aggregation.simd.SimdDoubleMinVectorAggregator;
import org.apache.druid.segment.BaseDoubleColumnValueSelector;
import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
import org.apache.druid.segment.vector.VectorValueSelector;
@@ -81,6 +83,9 @@ public class DoubleMinAggregatorFactory extends
SimpleDoubleAggregatorFactory
VectorValueSelector selector
)
{
+ if (ExpressionProcessing.useVectorApi()) {
+ return new SimdDoubleMinVectorAggregator(selector);
+ }
return new DoubleMinVectorAggregator(selector);
}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/FloatMaxAggregatorFactory.java
b/processing/src/main/java/org/apache/druid/query/aggregation/FloatMaxAggregatorFactory.java
index 4bfeba615cd..553f8103800 100644
---
a/processing/src/main/java/org/apache/druid/query/aggregation/FloatMaxAggregatorFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/FloatMaxAggregatorFactory.java
@@ -24,6 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Supplier;
import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.math.expr.ExpressionProcessing;
+import org.apache.druid.query.aggregation.simd.SimdFloatMaxVectorAggregator;
import org.apache.druid.segment.BaseFloatColumnValueSelector;
import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
import org.apache.druid.segment.vector.VectorValueSelector;
@@ -81,6 +83,9 @@ public class FloatMaxAggregatorFactory extends
SimpleFloatAggregatorFactory
VectorValueSelector selector
)
{
+ if (ExpressionProcessing.useVectorApi()) {
+ return new SimdFloatMaxVectorAggregator(selector);
+ }
return new FloatMaxVectorAggregator(selector);
}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/FloatMinAggregatorFactory.java
b/processing/src/main/java/org/apache/druid/query/aggregation/FloatMinAggregatorFactory.java
index d720658a6b3..3355414f6c8 100644
---
a/processing/src/main/java/org/apache/druid/query/aggregation/FloatMinAggregatorFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/FloatMinAggregatorFactory.java
@@ -24,6 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Supplier;
import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.math.expr.ExpressionProcessing;
+import org.apache.druid.query.aggregation.simd.SimdFloatMinVectorAggregator;
import org.apache.druid.segment.BaseFloatColumnValueSelector;
import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
import org.apache.druid.segment.vector.VectorValueSelector;
@@ -81,6 +83,9 @@ public class FloatMinAggregatorFactory extends
SimpleFloatAggregatorFactory
VectorValueSelector selector
)
{
+ if (ExpressionProcessing.useVectorApi()) {
+ return new SimdFloatMinVectorAggregator(selector);
+ }
return new FloatMinVectorAggregator(selector);
}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/LongMaxAggregatorFactory.java
b/processing/src/main/java/org/apache/druid/query/aggregation/LongMaxAggregatorFactory.java
index 1304d272ede..1caa9b6b622 100644
---
a/processing/src/main/java/org/apache/druid/query/aggregation/LongMaxAggregatorFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/LongMaxAggregatorFactory.java
@@ -24,6 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Supplier;
import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.math.expr.ExpressionProcessing;
+import org.apache.druid.query.aggregation.simd.SimdLongMaxVectorAggregator;
import org.apache.druid.segment.BaseLongColumnValueSelector;
import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
import org.apache.druid.segment.vector.VectorValueSelector;
@@ -81,6 +83,9 @@ public class LongMaxAggregatorFactory extends
SimpleLongAggregatorFactory
VectorValueSelector selector
)
{
+ if (ExpressionProcessing.useVectorApi()) {
+ return new SimdLongMaxVectorAggregator(selector);
+ }
return new LongMaxVectorAggregator(selector);
}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/LongMinAggregatorFactory.java
b/processing/src/main/java/org/apache/druid/query/aggregation/LongMinAggregatorFactory.java
index 3073217986c..d6ff9547f44 100644
---
a/processing/src/main/java/org/apache/druid/query/aggregation/LongMinAggregatorFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/LongMinAggregatorFactory.java
@@ -24,6 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Supplier;
import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.math.expr.ExpressionProcessing;
+import org.apache.druid.query.aggregation.simd.SimdLongMinVectorAggregator;
import org.apache.druid.segment.BaseLongColumnValueSelector;
import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
import org.apache.druid.segment.vector.VectorValueSelector;
@@ -81,6 +83,9 @@ public class LongMinAggregatorFactory extends
SimpleLongAggregatorFactory
VectorValueSelector selector
)
{
+ if (ExpressionProcessing.useVectorApi()) {
+ return new SimdLongMinVectorAggregator(selector);
+ }
return new LongMinVectorAggregator(selector);
}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdDoubleMaxVectorAggregator.java
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdDoubleMaxVectorAggregator.java
new file mode 100644
index 00000000000..8a691f8ddd8
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdDoubleMaxVectorAggregator.java
@@ -0,0 +1,97 @@
+/*
+ * 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.simd;
+
+import jdk.incubator.vector.DoubleVector;
+import jdk.incubator.vector.VectorMask;
+import jdk.incubator.vector.VectorOperators;
+import jdk.incubator.vector.VectorSpecies;
+import org.apache.druid.query.aggregation.DoubleMaxVectorAggregator;
+import org.apache.druid.query.aggregation.NullAwareVectorAggregator;
+import org.apache.druid.segment.vector.VectorValueSelector;
+
+import java.nio.ByteBuffer;
+
+/**
+ * SIMD specialization of {@link DoubleMaxVectorAggregator}'s ungrouped
contiguous-range aggregation. The hot loop
+ * issues a hardcoded {@link DoubleVector#max} and a {@code
reduceLanes(VectorOperators.MAX)} so the JIT emits the
+ * platform's double-max and double-max-reduce intrinsics. Null lanes preserve
the lane's seeded
+ * {@link Double#NEGATIVE_INFINITY} via masked {@code lanewise} so the
reduction is unaffected by them.
+ */
+public final class SimdDoubleMaxVectorAggregator extends
DoubleMaxVectorAggregator implements NullAwareVectorAggregator
+{
+ private static final VectorSpecies<Double> SPECIES =
DoubleVector.SPECIES_PREFERRED;
+
+ private final VectorValueSelector selector;
+
+ public SimdDoubleMaxVectorAggregator(VectorValueSelector selector)
+ {
+ super(selector);
+ this.selector = selector;
+ }
+
+ @Override
+ public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
+ {
+ final double[] vector = selector.getDoubleVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ DoubleVector vacc = DoubleVector.broadcast(SPECIES,
Double.NEGATIVE_INFINITY);
+ for (; i < upperBound; i += laneCount) {
+ vacc = vacc.max(DoubleVector.fromArray(SPECIES, vector, i));
+ }
+ double localMax = vacc.reduceLanes(VectorOperators.MAX);
+ for (; i < endRow; i++) {
+ localMax = Math.max(localMax, vector[i]);
+ }
+ buf.putDouble(position, Math.max(buf.getDouble(position), localMax));
+ }
+
+ @Override
+ public boolean aggregate(ByteBuffer buf, int position, int startRow, int
endRow, boolean[] nullVector)
+ {
+ final double[] vector = selector.getDoubleVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ DoubleVector vacc = DoubleVector.broadcast(SPECIES,
Double.NEGATIVE_INFINITY);
+ int nonNullCount = 0;
+ for (; i < upperBound; i += laneCount) {
+ final VectorMask<Double> notNull = VectorMask.fromArray(SPECIES,
nullVector, i).not();
+ vacc = vacc.lanewise(VectorOperators.MAX,
DoubleVector.fromArray(SPECIES, vector, i), notNull);
+ nonNullCount += notNull.trueCount();
+ }
+ double localMax = vacc.reduceLanes(VectorOperators.MAX);
+ for (; i < endRow; i++) {
+ if (!nullVector[i]) {
+ localMax = Math.max(localMax, vector[i]);
+ nonNullCount++;
+ }
+ }
+ if (nonNullCount > 0) {
+ buf.putDouble(position, Math.max(buf.getDouble(position), localMax));
+ return true;
+ }
+ return false;
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdDoubleMinVectorAggregator.java
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdDoubleMinVectorAggregator.java
new file mode 100644
index 00000000000..baa18c0924b
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdDoubleMinVectorAggregator.java
@@ -0,0 +1,97 @@
+/*
+ * 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.simd;
+
+import jdk.incubator.vector.DoubleVector;
+import jdk.incubator.vector.VectorMask;
+import jdk.incubator.vector.VectorOperators;
+import jdk.incubator.vector.VectorSpecies;
+import org.apache.druid.query.aggregation.DoubleMinVectorAggregator;
+import org.apache.druid.query.aggregation.NullAwareVectorAggregator;
+import org.apache.druid.segment.vector.VectorValueSelector;
+
+import java.nio.ByteBuffer;
+
+/**
+ * SIMD specialization of {@link DoubleMinVectorAggregator}'s ungrouped
contiguous-range aggregation. The hot loop
+ * issues a hardcoded {@link DoubleVector#min} and a {@code
reduceLanes(VectorOperators.MIN)} so the JIT emits the
+ * platform's double-min and double-min-reduce intrinsics. Null lanes preserve
the lane's seeded
+ * {@link Double#POSITIVE_INFINITY} via masked {@code lanewise} so the
reduction is unaffected by them.
+ */
+public final class SimdDoubleMinVectorAggregator extends
DoubleMinVectorAggregator implements NullAwareVectorAggregator
+{
+ private static final VectorSpecies<Double> SPECIES =
DoubleVector.SPECIES_PREFERRED;
+
+ private final VectorValueSelector selector;
+
+ public SimdDoubleMinVectorAggregator(VectorValueSelector selector)
+ {
+ super(selector);
+ this.selector = selector;
+ }
+
+ @Override
+ public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
+ {
+ final double[] vector = selector.getDoubleVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ DoubleVector vacc = DoubleVector.broadcast(SPECIES,
Double.POSITIVE_INFINITY);
+ for (; i < upperBound; i += laneCount) {
+ vacc = vacc.min(DoubleVector.fromArray(SPECIES, vector, i));
+ }
+ double localMin = vacc.reduceLanes(VectorOperators.MIN);
+ for (; i < endRow; i++) {
+ localMin = Math.min(localMin, vector[i]);
+ }
+ buf.putDouble(position, Math.min(buf.getDouble(position), localMin));
+ }
+
+ @Override
+ public boolean aggregate(ByteBuffer buf, int position, int startRow, int
endRow, boolean[] nullVector)
+ {
+ final double[] vector = selector.getDoubleVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ DoubleVector vacc = DoubleVector.broadcast(SPECIES,
Double.POSITIVE_INFINITY);
+ int nonNullCount = 0;
+ for (; i < upperBound; i += laneCount) {
+ final VectorMask<Double> notNull = VectorMask.fromArray(SPECIES,
nullVector, i).not();
+ vacc = vacc.lanewise(VectorOperators.MIN,
DoubleVector.fromArray(SPECIES, vector, i), notNull);
+ nonNullCount += notNull.trueCount();
+ }
+ double localMin = vacc.reduceLanes(VectorOperators.MIN);
+ for (; i < endRow; i++) {
+ if (!nullVector[i]) {
+ localMin = Math.min(localMin, vector[i]);
+ nonNullCount++;
+ }
+ }
+ if (nonNullCount > 0) {
+ buf.putDouble(position, Math.min(buf.getDouble(position), localMin));
+ return true;
+ }
+ return false;
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdFloatMaxVectorAggregator.java
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdFloatMaxVectorAggregator.java
new file mode 100644
index 00000000000..05a5481d958
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdFloatMaxVectorAggregator.java
@@ -0,0 +1,97 @@
+/*
+ * 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.simd;
+
+import jdk.incubator.vector.FloatVector;
+import jdk.incubator.vector.VectorMask;
+import jdk.incubator.vector.VectorOperators;
+import jdk.incubator.vector.VectorSpecies;
+import org.apache.druid.query.aggregation.FloatMaxVectorAggregator;
+import org.apache.druid.query.aggregation.NullAwareVectorAggregator;
+import org.apache.druid.segment.vector.VectorValueSelector;
+
+import java.nio.ByteBuffer;
+
+/**
+ * SIMD specialization of {@link FloatMaxVectorAggregator}'s ungrouped
contiguous-range aggregation. The hot loop
+ * issues a hardcoded {@link FloatVector#max} and a {@code
reduceLanes(VectorOperators.MAX)} so the JIT emits the
+ * platform's float-max and float-max-reduce intrinsics. Null lanes preserve
the lane's seeded
+ * {@link Float#NEGATIVE_INFINITY} via masked {@code lanewise} so the
reduction is unaffected by them.
+ */
+public final class SimdFloatMaxVectorAggregator extends
FloatMaxVectorAggregator implements NullAwareVectorAggregator
+{
+ private static final VectorSpecies<Float> SPECIES =
FloatVector.SPECIES_PREFERRED;
+
+ private final VectorValueSelector selector;
+
+ public SimdFloatMaxVectorAggregator(VectorValueSelector selector)
+ {
+ super(selector);
+ this.selector = selector;
+ }
+
+ @Override
+ public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
+ {
+ final float[] vector = selector.getFloatVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ FloatVector vacc = FloatVector.broadcast(SPECIES, Float.NEGATIVE_INFINITY);
+ for (; i < upperBound; i += laneCount) {
+ vacc = vacc.max(FloatVector.fromArray(SPECIES, vector, i));
+ }
+ float localMax = vacc.reduceLanes(VectorOperators.MAX);
+ for (; i < endRow; i++) {
+ localMax = Math.max(localMax, vector[i]);
+ }
+ buf.putFloat(position, Math.max(buf.getFloat(position), localMax));
+ }
+
+ @Override
+ public boolean aggregate(ByteBuffer buf, int position, int startRow, int
endRow, boolean[] nullVector)
+ {
+ final float[] vector = selector.getFloatVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ FloatVector vacc = FloatVector.broadcast(SPECIES, Float.NEGATIVE_INFINITY);
+ int nonNullCount = 0;
+ for (; i < upperBound; i += laneCount) {
+ final VectorMask<Float> notNull = VectorMask.fromArray(SPECIES,
nullVector, i).not();
+ vacc = vacc.lanewise(VectorOperators.MAX, FloatVector.fromArray(SPECIES,
vector, i), notNull);
+ nonNullCount += notNull.trueCount();
+ }
+ float localMax = vacc.reduceLanes(VectorOperators.MAX);
+ for (; i < endRow; i++) {
+ if (!nullVector[i]) {
+ localMax = Math.max(localMax, vector[i]);
+ nonNullCount++;
+ }
+ }
+ if (nonNullCount > 0) {
+ buf.putFloat(position, Math.max(buf.getFloat(position), localMax));
+ return true;
+ }
+ return false;
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdFloatMinVectorAggregator.java
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdFloatMinVectorAggregator.java
new file mode 100644
index 00000000000..05b61334c80
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdFloatMinVectorAggregator.java
@@ -0,0 +1,97 @@
+/*
+ * 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.simd;
+
+import jdk.incubator.vector.FloatVector;
+import jdk.incubator.vector.VectorMask;
+import jdk.incubator.vector.VectorOperators;
+import jdk.incubator.vector.VectorSpecies;
+import org.apache.druid.query.aggregation.FloatMinVectorAggregator;
+import org.apache.druid.query.aggregation.NullAwareVectorAggregator;
+import org.apache.druid.segment.vector.VectorValueSelector;
+
+import java.nio.ByteBuffer;
+
+/**
+ * SIMD specialization of {@link FloatMinVectorAggregator}'s ungrouped
contiguous-range aggregation. The hot loop
+ * issues a hardcoded {@link FloatVector#min} and a {@code
reduceLanes(VectorOperators.MIN)} so the JIT emits the
+ * platform's float-min and float-min-reduce intrinsics. Null lanes preserve
the lane's seeded
+ * {@link Float#POSITIVE_INFINITY} via masked {@code lanewise} so the
reduction is unaffected by them.
+ */
+public final class SimdFloatMinVectorAggregator extends
FloatMinVectorAggregator implements NullAwareVectorAggregator
+{
+ private static final VectorSpecies<Float> SPECIES =
FloatVector.SPECIES_PREFERRED;
+
+ private final VectorValueSelector selector;
+
+ public SimdFloatMinVectorAggregator(VectorValueSelector selector)
+ {
+ super(selector);
+ this.selector = selector;
+ }
+
+ @Override
+ public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
+ {
+ final float[] vector = selector.getFloatVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ FloatVector vacc = FloatVector.broadcast(SPECIES, Float.POSITIVE_INFINITY);
+ for (; i < upperBound; i += laneCount) {
+ vacc = vacc.min(FloatVector.fromArray(SPECIES, vector, i));
+ }
+ float localMin = vacc.reduceLanes(VectorOperators.MIN);
+ for (; i < endRow; i++) {
+ localMin = Math.min(localMin, vector[i]);
+ }
+ buf.putFloat(position, Math.min(buf.getFloat(position), localMin));
+ }
+
+ @Override
+ public boolean aggregate(ByteBuffer buf, int position, int startRow, int
endRow, boolean[] nullVector)
+ {
+ final float[] vector = selector.getFloatVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ FloatVector vacc = FloatVector.broadcast(SPECIES, Float.POSITIVE_INFINITY);
+ int nonNullCount = 0;
+ for (; i < upperBound; i += laneCount) {
+ final VectorMask<Float> notNull = VectorMask.fromArray(SPECIES,
nullVector, i).not();
+ vacc = vacc.lanewise(VectorOperators.MIN, FloatVector.fromArray(SPECIES,
vector, i), notNull);
+ nonNullCount += notNull.trueCount();
+ }
+ float localMin = vacc.reduceLanes(VectorOperators.MIN);
+ for (; i < endRow; i++) {
+ if (!nullVector[i]) {
+ localMin = Math.min(localMin, vector[i]);
+ nonNullCount++;
+ }
+ }
+ if (nonNullCount > 0) {
+ buf.putFloat(position, Math.min(buf.getFloat(position), localMin));
+ return true;
+ }
+ return false;
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdLongMaxVectorAggregator.java
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdLongMaxVectorAggregator.java
new file mode 100644
index 00000000000..aad1edfc087
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdLongMaxVectorAggregator.java
@@ -0,0 +1,97 @@
+/*
+ * 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.simd;
+
+import jdk.incubator.vector.LongVector;
+import jdk.incubator.vector.VectorMask;
+import jdk.incubator.vector.VectorOperators;
+import jdk.incubator.vector.VectorSpecies;
+import org.apache.druid.query.aggregation.LongMaxVectorAggregator;
+import org.apache.druid.query.aggregation.NullAwareVectorAggregator;
+import org.apache.druid.segment.vector.VectorValueSelector;
+
+import java.nio.ByteBuffer;
+
+/**
+ * SIMD specialization of {@link LongMaxVectorAggregator}'s ungrouped
contiguous-range aggregation. The hot loop
+ * issues a hardcoded {@link LongVector#max} and a {@code
reduceLanes(VectorOperators.MAX)} so the JIT emits the
+ * platform's long-max and long-max-reduce intrinsics. Null lanes preserve the
lane's seeded {@link Long#MIN_VALUE}
+ * via masked {@code lanewise} so the reduction is unaffected by them.
+ */
+public final class SimdLongMaxVectorAggregator extends LongMaxVectorAggregator
implements NullAwareVectorAggregator
+{
+ private static final VectorSpecies<Long> SPECIES =
LongVector.SPECIES_PREFERRED;
+
+ private final VectorValueSelector selector;
+
+ public SimdLongMaxVectorAggregator(VectorValueSelector selector)
+ {
+ super(selector);
+ this.selector = selector;
+ }
+
+ @Override
+ public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
+ {
+ final long[] vector = selector.getLongVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ LongVector vacc = LongVector.broadcast(SPECIES, Long.MIN_VALUE);
+ for (; i < upperBound; i += laneCount) {
+ vacc = vacc.max(LongVector.fromArray(SPECIES, vector, i));
+ }
+ long localMax = vacc.reduceLanes(VectorOperators.MAX);
+ for (; i < endRow; i++) {
+ localMax = Math.max(localMax, vector[i]);
+ }
+ buf.putLong(position, Math.max(buf.getLong(position), localMax));
+ }
+
+ @Override
+ public boolean aggregate(ByteBuffer buf, int position, int startRow, int
endRow, boolean[] nullVector)
+ {
+ final long[] vector = selector.getLongVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ LongVector vacc = LongVector.broadcast(SPECIES, Long.MIN_VALUE);
+ int nonNullCount = 0;
+ for (; i < upperBound; i += laneCount) {
+ final VectorMask<Long> notNull = VectorMask.fromArray(SPECIES,
nullVector, i).not();
+ vacc = vacc.lanewise(VectorOperators.MAX, LongVector.fromArray(SPECIES,
vector, i), notNull);
+ nonNullCount += notNull.trueCount();
+ }
+ long localMax = vacc.reduceLanes(VectorOperators.MAX);
+ for (; i < endRow; i++) {
+ if (!nullVector[i]) {
+ localMax = Math.max(localMax, vector[i]);
+ nonNullCount++;
+ }
+ }
+ if (nonNullCount > 0) {
+ buf.putLong(position, Math.max(buf.getLong(position), localMax));
+ return true;
+ }
+ return false;
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdLongMinVectorAggregator.java
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdLongMinVectorAggregator.java
new file mode 100644
index 00000000000..5d77a63a57d
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/query/aggregation/simd/SimdLongMinVectorAggregator.java
@@ -0,0 +1,97 @@
+/*
+ * 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.simd;
+
+import jdk.incubator.vector.LongVector;
+import jdk.incubator.vector.VectorMask;
+import jdk.incubator.vector.VectorOperators;
+import jdk.incubator.vector.VectorSpecies;
+import org.apache.druid.query.aggregation.LongMinVectorAggregator;
+import org.apache.druid.query.aggregation.NullAwareVectorAggregator;
+import org.apache.druid.segment.vector.VectorValueSelector;
+
+import java.nio.ByteBuffer;
+
+/**
+ * SIMD specialization of {@link LongMinVectorAggregator}'s ungrouped
contiguous-range aggregation. The hot loop
+ * issues a hardcoded {@link LongVector#min} and a {@code
reduceLanes(VectorOperators.MIN)} so the JIT emits the
+ * platform's long-min and long-min-reduce intrinsics. Null lanes preserve the
lane's seeded {@link Long#MAX_VALUE}
+ * via masked {@code lanewise} so the reduction is unaffected by them.
+ */
+public final class SimdLongMinVectorAggregator extends LongMinVectorAggregator
implements NullAwareVectorAggregator
+{
+ private static final VectorSpecies<Long> SPECIES =
LongVector.SPECIES_PREFERRED;
+
+ private final VectorValueSelector selector;
+
+ public SimdLongMinVectorAggregator(VectorValueSelector selector)
+ {
+ super(selector);
+ this.selector = selector;
+ }
+
+ @Override
+ public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
+ {
+ final long[] vector = selector.getLongVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ LongVector vacc = LongVector.broadcast(SPECIES, Long.MAX_VALUE);
+ for (; i < upperBound; i += laneCount) {
+ vacc = vacc.min(LongVector.fromArray(SPECIES, vector, i));
+ }
+ long localMin = vacc.reduceLanes(VectorOperators.MIN);
+ for (; i < endRow; i++) {
+ localMin = Math.min(localMin, vector[i]);
+ }
+ buf.putLong(position, Math.min(buf.getLong(position), localMin));
+ }
+
+ @Override
+ public boolean aggregate(ByteBuffer buf, int position, int startRow, int
endRow, boolean[] nullVector)
+ {
+ final long[] vector = selector.getLongVector();
+
+ final int laneCount = SPECIES.length();
+ final int upperBound = startRow + SPECIES.loopBound(endRow - startRow);
+ int i = startRow;
+ LongVector vacc = LongVector.broadcast(SPECIES, Long.MAX_VALUE);
+ int nonNullCount = 0;
+ for (; i < upperBound; i += laneCount) {
+ final VectorMask<Long> notNull = VectorMask.fromArray(SPECIES,
nullVector, i).not();
+ vacc = vacc.lanewise(VectorOperators.MIN, LongVector.fromArray(SPECIES,
vector, i), notNull);
+ nonNullCount += notNull.trueCount();
+ }
+ long localMin = vacc.reduceLanes(VectorOperators.MIN);
+ for (; i < endRow; i++) {
+ if (!nullVector[i]) {
+ localMin = Math.min(localMin, vector[i]);
+ nonNullCount++;
+ }
+ }
+ if (nonNullCount > 0) {
+ buf.putLong(position, Math.min(buf.getLong(position), localMin));
+ return true;
+ }
+ return false;
+ }
+}
diff --git
a/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdAggregatorTestHelpers.java
b/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdAggregatorTestHelpers.java
new file mode 100644
index 00000000000..d6bd0c735d1
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdAggregatorTestHelpers.java
@@ -0,0 +1,191 @@
+/*
+ * 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.simd;
+
+import org.apache.druid.segment.vector.VectorValueSelector;
+
+import javax.annotation.Nullable;
+import java.util.Random;
+import java.util.function.IntPredicate;
+
+/**
+ * Shared fixtures for the SIMD vector aggregator equivalence tests. Holds the
vector-size axis, a set of null
+ * patterns designed to stress specific regions of the SIMD chunk loop, a
minimal in-memory
+ * {@link VectorValueSelector}, and small pseudo-random data generators.
+ */
+final class SimdAggregatorTestHelpers
+{
+ /**
+ * Vector sizes picked to exercise distinct regimes against typical SIMD
species lengths: 1 is tail-only,
+ * 8 fills an exact AVX2/AVX-512 chunk, 17 leaves a ragged tail after one or
more chunks, 64 spans many chunks,
+ * and 1023 is large with a ragged tail.
+ */
+ static final int[] VECTOR_SIZES = {1, 8, 17, 64, 1023};
+
+ private SimdAggregatorTestHelpers()
+ {
+ }
+
+ /**
+ * Returns a {@code boolean[]} of length {@code startRow + realNulls.length}
with the real null pattern copied
+ * to indices {@code [startRow, startRow + realNulls.length)}. The leading
slots are left as {@code false}
+ * (non-null) so that if the aggregator incorrectly reads past {@code
startRow} the poison values would slip in.
+ */
+ static boolean[] padNulls(boolean[] realNulls, int startRow)
+ {
+ final boolean[] padded = new boolean[startRow + realNulls.length];
+ System.arraycopy(realNulls, 0, padded, startRow, realNulls.length);
+ return padded;
+ }
+
+ static long[] randomLongs(int size, int seed)
+ {
+ final Random r = new Random(0xC0FFEEL + seed);
+ final long[] out = new long[size];
+ for (int i = 0; i < size; i++) {
+ out[i] = r.nextInt() & 0xFFFFFL;
+ }
+ return out;
+ }
+
+ static double[] randomDoubles(int size, int seed)
+ {
+ final Random r = new Random(0xC0FFEEL + seed);
+ final double[] out = new double[size];
+ for (int i = 0; i < size; i++) {
+ out[i] = (r.nextDouble() - 0.5) * 1000.0;
+ }
+ return out;
+ }
+
+ static float[] randomFloats(int size, int seed)
+ {
+ final Random r = new Random(0xC0FFEEL + seed);
+ final float[] out = new float[size];
+ for (int i = 0; i < size; i++) {
+ out[i] = (r.nextFloat() - 0.5f) * 1000.0f;
+ }
+ return out;
+ }
+
+ /**
+ * Null distributions designed to stress the SIMD null-aware path. {@link
#NONE} returns {@code null} from
+ * {@link #toMask(int)} to model a column with no null vector at all (the
common non-nullable case). The rest
+ * exercise sparse, dense, all-null, alternating, and
chunk-boundary-adjacent placements.
+ */
+ enum NullPattern
+ {
+ NONE(i -> false),
+ ALL(i -> true),
+ ALTERNATING(i -> (i & 1) == 0),
+ SPARSE(i -> i % 7 == 0),
+ FIRST_THREE(i -> i < 3),
+ CHUNK_BOUNDARY(i -> i == 7 || i == 8);
+
+ private final IntPredicate predicate;
+
+ NullPattern(IntPredicate predicate)
+ {
+ this.predicate = predicate;
+ }
+
+ @Nullable
+ boolean[] toMask(int size)
+ {
+ if (this == NONE) {
+ return null;
+ }
+ final boolean[] mask = new boolean[size];
+ for (int i = 0; i < size; i++) {
+ mask[i] = predicate.test(i);
+ }
+ return mask;
+ }
+ }
+
+ /**
+ * Minimal in-memory {@link VectorValueSelector} backed by pre-built
primitive arrays. Only the accessor for the
+ * type used by a given test is non-null; the others return whatever was
passed at construction (typically null).
+ */
+ static final class FakeVectorValueSelector implements VectorValueSelector
+ {
+ private final int size;
+ @Nullable
+ private final long[] longs;
+ @Nullable
+ private final double[] doubles;
+ @Nullable
+ private final float[] floats;
+ @Nullable
+ private final boolean[] nulls;
+
+ FakeVectorValueSelector(
+ int size,
+ @Nullable long[] longs,
+ @Nullable double[] doubles,
+ @Nullable float[] floats,
+ @Nullable boolean[] nulls
+ )
+ {
+ this.size = size;
+ this.longs = longs;
+ this.doubles = doubles;
+ this.floats = floats;
+ this.nulls = nulls;
+ }
+
+ @Override
+ public long[] getLongVector()
+ {
+ return longs;
+ }
+
+ @Override
+ public float[] getFloatVector()
+ {
+ return floats;
+ }
+
+ @Override
+ public double[] getDoubleVector()
+ {
+ return doubles;
+ }
+
+ @Nullable
+ @Override
+ public boolean[] getNullVector()
+ {
+ return nulls;
+ }
+
+ @Override
+ public int getMaxVectorSize()
+ {
+ return size;
+ }
+
+ @Override
+ public int getCurrentVectorSize()
+ {
+ return size;
+ }
+ }
+}
diff --git
a/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdMinMaxVectorAggregatorTest.java
b/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdMinMaxVectorAggregatorTest.java
new file mode 100644
index 00000000000..256021b24ef
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdMinMaxVectorAggregatorTest.java
@@ -0,0 +1,319 @@
+/*
+ * 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.simd;
+
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.aggregation.DoubleMaxVectorAggregator;
+import org.apache.druid.query.aggregation.DoubleMinVectorAggregator;
+import org.apache.druid.query.aggregation.FloatMaxVectorAggregator;
+import org.apache.druid.query.aggregation.FloatMinVectorAggregator;
+import org.apache.druid.query.aggregation.LongMaxVectorAggregator;
+import org.apache.druid.query.aggregation.LongMinVectorAggregator;
+import
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.FakeVectorValueSelector;
+import
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.NullPattern;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.VECTOR_SIZES;
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.padNulls;
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.randomDoubles;
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.randomFloats;
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.randomLongs;
+
+/**
+ * Equivalence tests for SIMD Min/Max vector aggregators. For each (op, type,
vector size, null pattern) tuple,
+ * drives the SIMD aggregator directly and compares against either the scalar
parent (no-null path) or a manually
+ * computed reference (null-aware path). When every row is null the null-aware
path must report false and leave
+ * the buffer's seeded identity value untouched.
+ *
+ * Each scenario is exercised twice: once with {@code (position=0, startRow=0,
endRow=size)} and once with
+ * {@code (position=1, startRow=1, endRow=size+1)} where the row at index 0
holds a deliberately op-defeating
+ * "poison" value (an extreme low for min, extreme high for max) that would
visibly change the result if the
+ * aggregator incorrectly read past {@code startRow}, and the buffer slot
starts at byte offset 1 so any
+ * indexing off the position parameter shows up.
+ */
+public class SimdMinMaxVectorAggregatorTest extends InitializedNullHandlingTest
+{
+ @Test
+ public void testLongMin()
+ {
+ for (int size : VECTOR_SIZES) {
+ for (NullPattern p : NullPattern.values()) {
+ runLong(size, p, true, 0, 0);
+ runLong(size, p, true, 1, 1);
+ }
+ }
+ }
+
+ @Test
+ public void testLongMax()
+ {
+ for (int size : VECTOR_SIZES) {
+ for (NullPattern p : NullPattern.values()) {
+ runLong(size, p, false, 0, 0);
+ runLong(size, p, false, 1, 1);
+ }
+ }
+ }
+
+ @Test
+ public void testDoubleMin()
+ {
+ for (int size : VECTOR_SIZES) {
+ for (NullPattern p : NullPattern.values()) {
+ runDouble(size, p, true, 0, 0);
+ runDouble(size, p, true, 1, 1);
+ }
+ }
+ }
+
+ @Test
+ public void testDoubleMax()
+ {
+ for (int size : VECTOR_SIZES) {
+ for (NullPattern p : NullPattern.values()) {
+ runDouble(size, p, false, 0, 0);
+ runDouble(size, p, false, 1, 1);
+ }
+ }
+ }
+
+ @Test
+ public void testFloatMin()
+ {
+ for (int size : VECTOR_SIZES) {
+ for (NullPattern p : NullPattern.values()) {
+ runFloat(size, p, true, 0, 0);
+ runFloat(size, p, true, 1, 1);
+ }
+ }
+ }
+
+ @Test
+ public void testFloatMax()
+ {
+ for (int size : VECTOR_SIZES) {
+ for (NullPattern p : NullPattern.values()) {
+ runFloat(size, p, false, 0, 0);
+ runFloat(size, p, false, 1, 1);
+ }
+ }
+ }
+
+ private static void runLong(int size, NullPattern pattern, boolean isMin,
int position, int startRow)
+ {
+ final int arrLen = startRow + size;
+ final long[] values = new long[arrLen];
+ final long poison = isMin ? Long.MIN_VALUE : Long.MAX_VALUE;
+ for (int i = 0; i < startRow; i++) {
+ values[i] = poison;
+ }
+ System.arraycopy(randomLongs(size, isMin ? 0 : 1), 0, values, startRow,
size);
+
+ final boolean[] realNulls = pattern.toMask(size);
+ final boolean[] nulls = realNulls == null ? null : padNulls(realNulls,
startRow);
+
+ final FakeVectorValueSelector selector = new
FakeVectorValueSelector(arrLen, values, null, null, nulls);
+ final int endRow = startRow + size;
+ final String msg = StringUtils.format(
+ "type[long] op[%s] size[%s] nulls[%s] pos[%s] start[%s]",
+ isMin ? "min" : "max", size, pattern, position, startRow
+ );
+ final long identity = isMin ? Long.MAX_VALUE : Long.MIN_VALUE;
+
+ if (nulls == null) {
+ final ByteBuffer scalarBuf = ByteBuffer.allocate(position + Long.BYTES);
+ final ByteBuffer simdBuf = ByteBuffer.allocate(position + Long.BYTES);
+ if (isMin) {
+ final LongMinVectorAggregator scalar = new
LongMinVectorAggregator(selector);
+ final SimdLongMinVectorAggregator simd = new
SimdLongMinVectorAggregator(selector);
+ scalar.init(scalarBuf, position);
+ simd.init(simdBuf, position);
+ scalar.aggregate(scalarBuf, position, startRow, endRow);
+ simd.aggregate(simdBuf, position, startRow, endRow);
+ } else {
+ final LongMaxVectorAggregator scalar = new
LongMaxVectorAggregator(selector);
+ final SimdLongMaxVectorAggregator simd = new
SimdLongMaxVectorAggregator(selector);
+ scalar.init(scalarBuf, position);
+ simd.init(simdBuf, position);
+ scalar.aggregate(scalarBuf, position, startRow, endRow);
+ simd.aggregate(simdBuf, position, startRow, endRow);
+ }
+ Assert.assertEquals(msg, scalarBuf.getLong(position),
simdBuf.getLong(position));
+ } else {
+ long expected = identity;
+ boolean anyNonNull = false;
+ for (int i = startRow; i < endRow; i++) {
+ if (!nulls[i]) {
+ expected = isMin ? Math.min(expected, values[i]) :
Math.max(expected, values[i]);
+ anyNonNull = true;
+ }
+ }
+ final ByteBuffer simdBuf = ByteBuffer.allocate(position + Long.BYTES);
+ final boolean reported;
+ if (isMin) {
+ final SimdLongMinVectorAggregator simd = new
SimdLongMinVectorAggregator(selector);
+ simd.init(simdBuf, position);
+ reported = simd.aggregate(simdBuf, position, startRow, endRow, nulls);
+ } else {
+ final SimdLongMaxVectorAggregator simd = new
SimdLongMaxVectorAggregator(selector);
+ simd.init(simdBuf, position);
+ reported = simd.aggregate(simdBuf, position, startRow, endRow, nulls);
+ }
+ Assert.assertEquals(msg + " (anyNonNull)", anyNonNull, reported);
+ Assert.assertEquals(msg, expected, simdBuf.getLong(position));
+ }
+ }
+
+ private static void runDouble(int size, NullPattern pattern, boolean isMin,
int position, int startRow)
+ {
+ final int arrLen = startRow + size;
+ final double[] values = new double[arrLen];
+ final double poison = isMin ? Double.NEGATIVE_INFINITY :
Double.POSITIVE_INFINITY;
+ for (int i = 0; i < startRow; i++) {
+ values[i] = poison;
+ }
+ System.arraycopy(randomDoubles(size, isMin ? 2 : 3), 0, values, startRow,
size);
+
+ final boolean[] realNulls = pattern.toMask(size);
+ final boolean[] nulls = realNulls == null ? null : padNulls(realNulls,
startRow);
+
+ final FakeVectorValueSelector selector = new
FakeVectorValueSelector(arrLen, null, values, null, nulls);
+ final int endRow = startRow + size;
+ final String msg = StringUtils.format(
+ "type[double] op[%s] size[%s] nulls[%s] pos[%s] start[%s]",
+ isMin ? "min" : "max", size, pattern, position, startRow
+ );
+ final double identity = isMin ? Double.POSITIVE_INFINITY :
Double.NEGATIVE_INFINITY;
+
+ if (nulls == null) {
+ final ByteBuffer scalarBuf = ByteBuffer.allocate(position +
Double.BYTES);
+ final ByteBuffer simdBuf = ByteBuffer.allocate(position + Double.BYTES);
+ if (isMin) {
+ final DoubleMinVectorAggregator scalar = new
DoubleMinVectorAggregator(selector);
+ final SimdDoubleMinVectorAggregator simd = new
SimdDoubleMinVectorAggregator(selector);
+ scalar.init(scalarBuf, position);
+ simd.init(simdBuf, position);
+ scalar.aggregate(scalarBuf, position, startRow, endRow);
+ simd.aggregate(simdBuf, position, startRow, endRow);
+ } else {
+ final DoubleMaxVectorAggregator scalar = new
DoubleMaxVectorAggregator(selector);
+ final SimdDoubleMaxVectorAggregator simd = new
SimdDoubleMaxVectorAggregator(selector);
+ scalar.init(scalarBuf, position);
+ simd.init(simdBuf, position);
+ scalar.aggregate(scalarBuf, position, startRow, endRow);
+ simd.aggregate(simdBuf, position, startRow, endRow);
+ }
+ // min/max produces a value that was present in the input -- exact
equality is fine.
+ Assert.assertEquals(msg, scalarBuf.getDouble(position),
simdBuf.getDouble(position), 0.0);
+ } else {
+ double expected = identity;
+ boolean anyNonNull = false;
+ for (int i = startRow; i < endRow; i++) {
+ if (!nulls[i]) {
+ expected = isMin ? Math.min(expected, values[i]) :
Math.max(expected, values[i]);
+ anyNonNull = true;
+ }
+ }
+ final ByteBuffer simdBuf = ByteBuffer.allocate(position + Double.BYTES);
+ final boolean reported;
+ if (isMin) {
+ final SimdDoubleMinVectorAggregator simd = new
SimdDoubleMinVectorAggregator(selector);
+ simd.init(simdBuf, position);
+ reported = simd.aggregate(simdBuf, position, startRow, endRow, nulls);
+ } else {
+ final SimdDoubleMaxVectorAggregator simd = new
SimdDoubleMaxVectorAggregator(selector);
+ simd.init(simdBuf, position);
+ reported = simd.aggregate(simdBuf, position, startRow, endRow, nulls);
+ }
+ Assert.assertEquals(msg + " (anyNonNull)", anyNonNull, reported);
+ Assert.assertEquals(msg, expected, simdBuf.getDouble(position), 0.0);
+ }
+ }
+
+ private static void runFloat(int size, NullPattern pattern, boolean isMin,
int position, int startRow)
+ {
+ final int arrLen = startRow + size;
+ final float[] values = new float[arrLen];
+ final float poison = isMin ? Float.NEGATIVE_INFINITY :
Float.POSITIVE_INFINITY;
+ for (int i = 0; i < startRow; i++) {
+ values[i] = poison;
+ }
+ System.arraycopy(randomFloats(size, isMin ? 4 : 5), 0, values, startRow,
size);
+
+ final boolean[] realNulls = pattern.toMask(size);
+ final boolean[] nulls = realNulls == null ? null : padNulls(realNulls,
startRow);
+
+ final FakeVectorValueSelector selector = new
FakeVectorValueSelector(arrLen, null, null, values, nulls);
+ final int endRow = startRow + size;
+ final String msg = StringUtils.format(
+ "type[float] op[%s] size[%s] nulls[%s] pos[%s] start[%s]",
+ isMin ? "min" : "max", size, pattern, position, startRow
+ );
+ final float identity = isMin ? Float.POSITIVE_INFINITY :
Float.NEGATIVE_INFINITY;
+
+ if (nulls == null) {
+ final ByteBuffer scalarBuf = ByteBuffer.allocate(position + Float.BYTES);
+ final ByteBuffer simdBuf = ByteBuffer.allocate(position + Float.BYTES);
+ if (isMin) {
+ final FloatMinVectorAggregator scalar = new
FloatMinVectorAggregator(selector);
+ final SimdFloatMinVectorAggregator simd = new
SimdFloatMinVectorAggregator(selector);
+ scalar.init(scalarBuf, position);
+ simd.init(simdBuf, position);
+ scalar.aggregate(scalarBuf, position, startRow, endRow);
+ simd.aggregate(simdBuf, position, startRow, endRow);
+ } else {
+ final FloatMaxVectorAggregator scalar = new
FloatMaxVectorAggregator(selector);
+ final SimdFloatMaxVectorAggregator simd = new
SimdFloatMaxVectorAggregator(selector);
+ scalar.init(scalarBuf, position);
+ simd.init(simdBuf, position);
+ scalar.aggregate(scalarBuf, position, startRow, endRow);
+ simd.aggregate(simdBuf, position, startRow, endRow);
+ }
+ Assert.assertEquals(msg, scalarBuf.getFloat(position),
simdBuf.getFloat(position), 0.0f);
+ } else {
+ float expected = identity;
+ boolean anyNonNull = false;
+ for (int i = startRow; i < endRow; i++) {
+ if (!nulls[i]) {
+ expected = isMin ? Math.min(expected, values[i]) :
Math.max(expected, values[i]);
+ anyNonNull = true;
+ }
+ }
+ final ByteBuffer simdBuf = ByteBuffer.allocate(position + Float.BYTES);
+ final boolean reported;
+ if (isMin) {
+ final SimdFloatMinVectorAggregator simd = new
SimdFloatMinVectorAggregator(selector);
+ simd.init(simdBuf, position);
+ reported = simd.aggregate(simdBuf, position, startRow, endRow, nulls);
+ } else {
+ final SimdFloatMaxVectorAggregator simd = new
SimdFloatMaxVectorAggregator(selector);
+ simd.init(simdBuf, position);
+ reported = simd.aggregate(simdBuf, position, startRow, endRow, nulls);
+ }
+ Assert.assertEquals(msg + " (anyNonNull)", anyNonNull, reported);
+ Assert.assertEquals(msg, expected, simdBuf.getFloat(position), 0.0f);
+ }
+ }
+
+}
diff --git
a/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdSumVectorAggregatorTest.java
b/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdSumVectorAggregatorTest.java
index 19e54b075a3..d4fbd27f399 100644
---
a/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdSumVectorAggregatorTest.java
+++
b/processing/src/test/java/org/apache/druid/query/aggregation/simd/SimdSumVectorAggregatorTest.java
@@ -23,15 +23,19 @@ import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.query.aggregation.DoubleSumVectorAggregator;
import org.apache.druid.query.aggregation.FloatSumVectorAggregator;
import org.apache.druid.query.aggregation.LongSumVectorAggregator;
-import org.apache.druid.segment.vector.VectorValueSelector;
+import
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.FakeVectorValueSelector;
+import
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.NullPattern;
import org.apache.druid.testing.InitializedNullHandlingTest;
import org.junit.Assert;
import org.junit.Test;
-import javax.annotation.Nullable;
import java.nio.ByteBuffer;
-import java.util.Random;
-import java.util.function.IntPredicate;
+
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.VECTOR_SIZES;
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.padNulls;
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.randomDoubles;
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.randomFloats;
+import static
org.apache.druid.query.aggregation.simd.SimdAggregatorTestHelpers.randomLongs;
/**
* For each (sum type, vector size, null pattern) combination, drives the SIMD
and scalar sum vector aggregators
@@ -49,7 +53,6 @@ import java.util.function.IntPredicate;
*/
public class SimdSumVectorAggregatorTest extends InitializedNullHandlingTest
{
- private static final int[] VECTOR_SIZES = {1, 8, 17, 64, 1023};
private static final long POISON_LONG = Long.MAX_VALUE / 2;
private static final double POISON_DOUBLE = 1e15;
private static final float POISON_FLOAT = 1e10f;
@@ -254,139 +257,4 @@ public class SimdSumVectorAggregatorTest extends
InitializedNullHandlingTest
}
}
- private static boolean[] padNulls(boolean[] realNulls, int startRow)
- {
- final boolean[] padded = new boolean[startRow + realNulls.length];
- System.arraycopy(realNulls, 0, padded, startRow, realNulls.length);
- return padded;
- }
-
- private static long[] randomLongs(int size, int seed)
- {
- final Random r = new Random(0xC0FFEEL + seed);
- final long[] out = new long[size];
- for (int i = 0; i < size; i++) {
- out[i] = r.nextInt() & 0xFFFFFL;
- }
- return out;
- }
-
- private static double[] randomDoubles(int size, int seed)
- {
- final Random r = new Random(0xC0FFEEL + seed);
- final double[] out = new double[size];
- for (int i = 0; i < size; i++) {
- out[i] = (r.nextDouble() - 0.5) * 1000.0;
- }
- return out;
- }
-
- private static float[] randomFloats(int size, int seed)
- {
- final Random r = new Random(0xC0FFEEL + seed);
- final float[] out = new float[size];
- for (int i = 0; i < size; i++) {
- out[i] = (r.nextFloat() - 0.5f) * 1000.0f;
- }
- return out;
- }
-
- private enum NullPattern
- {
- NONE(i -> false),
- ALL(i -> true),
- ALTERNATING(i -> (i & 1) == 0),
- SPARSE(i -> i % 7 == 0),
- FIRST_THREE(i -> i < 3),
- CHUNK_BOUNDARY(i -> i == 7 || i == 8);
-
- private final IntPredicate predicate;
-
- NullPattern(IntPredicate predicate)
- {
- this.predicate = predicate;
- }
-
- @Nullable
- boolean[] toMask(int size)
- {
- if (this == NONE) {
- return null; // models a column with no null vector at all
- }
- final boolean[] mask = new boolean[size];
- for (int i = 0; i < size; i++) {
- mask[i] = predicate.test(i);
- }
- return mask;
- }
- }
-
- /**
- * Minimal in-memory {@link VectorValueSelector} backed by pre-built
primitive arrays for tests. Only the
- * accessor for the type used by a given test is non-null.
- */
- private static final class FakeVectorValueSelector implements
VectorValueSelector
- {
- private final int size;
- @Nullable
- private final long[] longs;
- @Nullable
- private final double[] doubles;
- @Nullable
- private final float[] floats;
- @Nullable
- private final boolean[] nulls;
-
- FakeVectorValueSelector(
- int size,
- @Nullable long[] longs,
- @Nullable double[] doubles,
- @Nullable float[] floats,
- @Nullable boolean[] nulls
- )
- {
- this.size = size;
- this.longs = longs;
- this.doubles = doubles;
- this.floats = floats;
- this.nulls = nulls;
- }
-
- @Override
- public long[] getLongVector()
- {
- return longs;
- }
-
- @Override
- public float[] getFloatVector()
- {
- return floats;
- }
-
- @Override
- public double[] getDoubleVector()
- {
- return doubles;
- }
-
- @Nullable
- @Override
- public boolean[] getNullVector()
- {
- return nulls;
- }
-
- @Override
- public int getMaxVectorSize()
- {
- return size;
- }
-
- @Override
- public int getCurrentVectorSize()
- {
- return size;
- }
- }
}
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/SqlVectorizedExpressionResultConsistencyTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/SqlVectorizedExpressionResultConsistencyTest.java
index 0ea81fc1c11..75157793dd0 100644
---
a/sql/src/test/java/org/apache/druid/sql/calcite/SqlVectorizedExpressionResultConsistencyTest.java
+++
b/sql/src/test/java/org/apache/druid/sql/calcite/SqlVectorizedExpressionResultConsistencyTest.java
@@ -225,8 +225,8 @@ public class SqlVectorizedExpressionResultConsistencyTest
extends InitializedNul
nonVectorObject,
vectorObject
),
- ((Double) nonVectorObject).doubleValue(),
- ((Double) vectorObject).doubleValue(),
+ ((Number) nonVectorObject).doubleValue(),
+ ((Number) vectorObject).doubleValue(),
0.01
);
} else {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
