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 a4b2ea87e63d..6550b024690f 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.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 @@ protected VectorAggregator factorizeVector( 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 241d7911af6a..cb95d52ab699 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.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 @@ protected VectorAggregator factorizeVector( 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 4bfeba615cdb..553f8103800a 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.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 @@ protected VectorAggregator factorizeVector( 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 d720658a6b3f..3355414f6c84 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.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 @@ protected VectorAggregator factorizeVector( 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 1304d272ede9..1caa9b6b622d 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.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 @@ protected VectorAggregator factorizeVector( 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 3073217986cf..d6ff9547f443 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.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 @@ protected VectorAggregator factorizeVector( 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 000000000000..8a691f8ddd85 --- /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 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 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 000000000000..baa18c0924bb --- /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 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 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 000000000000..05a5481d9586 --- /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 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 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 000000000000..05b61334c80d --- /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 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 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 000000000000..aad1edfc0873 --- /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 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 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 000000000000..5d77a63a57da --- /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 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 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 000000000000..d6bd0c735d1b --- /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 000000000000..256021b24ef1 --- /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 19e54b075a36..d4fbd27f3995 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.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 @@ */ 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 @@ private static void runFloat(int size, NullPattern pattern, int position, int st } } - 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; - } - } }