Jackie-Jiang commented on code in PR #9848: URL: https://github.com/apache/pinot/pull/9848#discussion_r1035432319
########## pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseBooleanAggregateFunction.java: ########## @@ -0,0 +1,249 @@ +/** + * 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.pinot.core.query.aggregation.function; + +import java.util.Map; +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.PinotDataType; +import org.apache.pinot.core.common.BlockValSet; +import org.apache.pinot.core.query.aggregation.AggregationResultHolder; +import org.apache.pinot.core.query.aggregation.LongAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.ObjectAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.LongGroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder; +import org.apache.pinot.spi.data.FieldSpec; +import org.roaringbitmap.RoaringBitmap; + + +public abstract class BaseBooleanAggregateFunction extends BaseSingleInputAggregationFunction<Integer, Integer> { + + private final BooleanMerge _merger; + private final boolean _nullHandlingEnabled; + + protected enum BooleanMerge { + AND { + @Override + long merge(long left, int right) { + return left * right; + } + + @Override + boolean isTerminal(long agg) { + return agg == 0; + } + + @Override + long getDefaultValue() { + return 1; + } + }, + OR { + @Override + long merge(long left, int right) { + return left + right; Review Comment: This should be `left | right`. For boolean result, the value should be either 0 or 1. ########## pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseBooleanAggregateFunction.java: ########## @@ -0,0 +1,249 @@ +/** + * 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.pinot.core.query.aggregation.function; + +import java.util.Map; +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.PinotDataType; +import org.apache.pinot.core.common.BlockValSet; +import org.apache.pinot.core.query.aggregation.AggregationResultHolder; +import org.apache.pinot.core.query.aggregation.LongAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.ObjectAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.LongGroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder; +import org.apache.pinot.spi.data.FieldSpec; +import org.roaringbitmap.RoaringBitmap; + + +public abstract class BaseBooleanAggregateFunction extends BaseSingleInputAggregationFunction<Integer, Integer> { + + private final BooleanMerge _merger; + private final boolean _nullHandlingEnabled; + + protected enum BooleanMerge { + AND { + @Override + long merge(long left, int right) { + return left * right; + } + + @Override + boolean isTerminal(long agg) { + return agg == 0; + } + + @Override + long getDefaultValue() { + return 1; + } + }, + OR { + @Override + long merge(long left, int right) { + return left + right; + } + + @Override + boolean isTerminal(long agg) { + return agg > 0; + } + + @Override + long getDefaultValue() { + return 0; + } + }; + + abstract long merge(long left, int right); + + abstract boolean isTerminal(long agg); + + abstract long getDefaultValue(); + } + + protected BaseBooleanAggregateFunction(ExpressionContext expression, boolean nullHandlingEnabled, + BooleanMerge merger) { + super(expression); + _nullHandlingEnabled = nullHandlingEnabled; + _merger = merger; + } + + @Override + public AggregationResultHolder createAggregationResultHolder() { + return _nullHandlingEnabled + ? new ObjectAggregationResultHolder() + : new LongAggregationResultHolder(_merger.getDefaultValue()); + } + + @Override + public GroupByResultHolder createGroupByResultHolder(int initialCapacity, int maxCapacity) { + return _nullHandlingEnabled + ? new ObjectGroupByResultHolder(initialCapacity, maxCapacity) + : new LongGroupByResultHolder(initialCapacity, maxCapacity, _merger.getDefaultValue()); + } + + @Override + public void aggregate(int length, AggregationResultHolder aggregationResultHolder, + Map<ExpressionContext, BlockValSet> blockValSetMap) { + BlockValSet blockValSet = blockValSetMap.get(_expression); + + if (blockValSet.getValueType().getStoredType() != FieldSpec.DataType.INT) { + throw new IllegalArgumentException( + String.format("Unsupported data type %s for BOOL_AND", blockValSet.getValueType())); Review Comment: (minor) This can also be `BOOL_OR` ########## pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseBooleanAggregateFunction.java: ########## @@ -0,0 +1,249 @@ +/** + * 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.pinot.core.query.aggregation.function; + +import java.util.Map; +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.PinotDataType; +import org.apache.pinot.core.common.BlockValSet; +import org.apache.pinot.core.query.aggregation.AggregationResultHolder; +import org.apache.pinot.core.query.aggregation.LongAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.ObjectAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.LongGroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder; +import org.apache.pinot.spi.data.FieldSpec; +import org.roaringbitmap.RoaringBitmap; + + +public abstract class BaseBooleanAggregateFunction extends BaseSingleInputAggregationFunction<Integer, Integer> { + + private final BooleanMerge _merger; + private final boolean _nullHandlingEnabled; + + protected enum BooleanMerge { + AND { + @Override + long merge(long left, int right) { + return left * right; + } + + @Override + boolean isTerminal(long agg) { + return agg == 0; + } + + @Override + long getDefaultValue() { + return 1; + } + }, + OR { + @Override + long merge(long left, int right) { + return left + right; + } + + @Override + boolean isTerminal(long agg) { + return agg > 0; + } + + @Override + long getDefaultValue() { + return 0; + } + }; + + abstract long merge(long left, int right); + + abstract boolean isTerminal(long agg); + + abstract long getDefaultValue(); + } + + protected BaseBooleanAggregateFunction(ExpressionContext expression, boolean nullHandlingEnabled, + BooleanMerge merger) { + super(expression); + _nullHandlingEnabled = nullHandlingEnabled; + _merger = merger; + } + + @Override + public AggregationResultHolder createAggregationResultHolder() { + return _nullHandlingEnabled + ? new ObjectAggregationResultHolder() + : new LongAggregationResultHolder(_merger.getDefaultValue()); + } + + @Override + public GroupByResultHolder createGroupByResultHolder(int initialCapacity, int maxCapacity) { + return _nullHandlingEnabled + ? new ObjectGroupByResultHolder(initialCapacity, maxCapacity) + : new LongGroupByResultHolder(initialCapacity, maxCapacity, _merger.getDefaultValue()); + } + + @Override + public void aggregate(int length, AggregationResultHolder aggregationResultHolder, + Map<ExpressionContext, BlockValSet> blockValSetMap) { + BlockValSet blockValSet = blockValSetMap.get(_expression); + + if (blockValSet.getValueType().getStoredType() != FieldSpec.DataType.INT) { + throw new IllegalArgumentException( + String.format("Unsupported data type %s for BOOL_AND", blockValSet.getValueType())); + } + + int[] bools = blockValSet.getIntValuesSV(); + if (_nullHandlingEnabled) { + long agg = getLong(aggregationResultHolder.getResult()); + + RoaringBitmap nullBitmap = blockValSet.getNullBitmap(); + if (nullBitmap == null) { + nullBitmap = new RoaringBitmap(); + } else if (nullBitmap.getCardinality() > length) { + return; + } + + for (int i = 0; i < length; i++) { + if (!nullBitmap.contains(i)) { + agg = _merger.merge(agg, bools[i]); + aggregationResultHolder.setValue((Object) agg); + if (_merger.isTerminal(agg)) { Review Comment: Move this check to line 117 so that the block can be skipped. We might not want to do per-value terminal check because tight loop tends to have better performance. We can do early terminate on block level ########## pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseBooleanAggregateFunction.java: ########## @@ -0,0 +1,249 @@ +/** + * 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.pinot.core.query.aggregation.function; + +import java.util.Map; +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.PinotDataType; +import org.apache.pinot.core.common.BlockValSet; +import org.apache.pinot.core.query.aggregation.AggregationResultHolder; +import org.apache.pinot.core.query.aggregation.LongAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.ObjectAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.LongGroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder; +import org.apache.pinot.spi.data.FieldSpec; +import org.roaringbitmap.RoaringBitmap; + + +public abstract class BaseBooleanAggregateFunction extends BaseSingleInputAggregationFunction<Integer, Integer> { + + private final BooleanMerge _merger; + private final boolean _nullHandlingEnabled; + + protected enum BooleanMerge { + AND { + @Override + long merge(long left, int right) { Review Comment: We should store the value as int to avoid the implicit cast from int to long. Suggest adding the result holder for int type in this PR, and we may add it for long types if needed in the future. ########## pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseBooleanAggregateFunction.java: ########## @@ -0,0 +1,249 @@ +/** + * 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.pinot.core.query.aggregation.function; + +import java.util.Map; +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.PinotDataType; +import org.apache.pinot.core.common.BlockValSet; +import org.apache.pinot.core.query.aggregation.AggregationResultHolder; +import org.apache.pinot.core.query.aggregation.LongAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.ObjectAggregationResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.LongGroupByResultHolder; +import org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder; +import org.apache.pinot.spi.data.FieldSpec; +import org.roaringbitmap.RoaringBitmap; + + +public abstract class BaseBooleanAggregateFunction extends BaseSingleInputAggregationFunction<Integer, Integer> { + + private final BooleanMerge _merger; + private final boolean _nullHandlingEnabled; + + protected enum BooleanMerge { + AND { + @Override + long merge(long left, int right) { + return left * right; + } + + @Override + boolean isTerminal(long agg) { + return agg == 0; + } + + @Override + long getDefaultValue() { + return 1; + } + }, + OR { + @Override + long merge(long left, int right) { + return left + right; + } + + @Override + boolean isTerminal(long agg) { + return agg > 0; + } + + @Override + long getDefaultValue() { + return 0; + } + }; + + abstract long merge(long left, int right); + + abstract boolean isTerminal(long agg); + + abstract long getDefaultValue(); + } + + protected BaseBooleanAggregateFunction(ExpressionContext expression, boolean nullHandlingEnabled, + BooleanMerge merger) { + super(expression); + _nullHandlingEnabled = nullHandlingEnabled; + _merger = merger; + } + + @Override + public AggregationResultHolder createAggregationResultHolder() { + return _nullHandlingEnabled + ? new ObjectAggregationResultHolder() + : new LongAggregationResultHolder(_merger.getDefaultValue()); + } + + @Override + public GroupByResultHolder createGroupByResultHolder(int initialCapacity, int maxCapacity) { + return _nullHandlingEnabled + ? new ObjectGroupByResultHolder(initialCapacity, maxCapacity) + : new LongGroupByResultHolder(initialCapacity, maxCapacity, _merger.getDefaultValue()); + } + + @Override + public void aggregate(int length, AggregationResultHolder aggregationResultHolder, + Map<ExpressionContext, BlockValSet> blockValSetMap) { + BlockValSet blockValSet = blockValSetMap.get(_expression); + + if (blockValSet.getValueType().getStoredType() != FieldSpec.DataType.INT) { Review Comment: Do we want to support it on `INT` column? If not, we should explicitly check the value type instead of the stored type. The current logic won't work well when the input value is not 0/1. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org For additional commands, e-mail: commits-h...@pinot.apache.org
