JackieTien97 commented on code in PR #13997:
URL: https://github.com/apache/iotdb/pull/13997#discussion_r1830598319
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java:
##########
@@ -56,7 +56,7 @@ public class PushAggregationIntoTableScan implements
PlanOptimizer {
@Override
public PlanNode optimize(PlanNode plan, PlanOptimizer.Context context) {
if (!(context.getAnalysis().getStatement() instanceof Query)
- || context.getAnalysis().noAggregates()) {
+ || !context.getAnalysis().isAggregationQuery()) {
Review Comment:
!isAggregationQuery is not equal to previous `noAggregates`
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/utils/TypeUtil.java:
##########
@@ -499,4 +499,96 @@ public static long hash(Type type, Column column, int
position) {
public static FlatHashStrategy getFlatHashStrategy(List<Type> hashTypes) {
return new HashStrategy(hashTypes);
}
+
+ public static boolean rowNotDistinctFromRow(
+ Type[] types, int leftPosition, Column[] leftBlock, int rightPosition,
Column[] rightBlock) {
+ for (int i = 0; i < types.length; i++) {
+ boolean leftNull = leftBlock[i].isNull(leftPosition);
+ boolean rightNull = rightBlock[i].isNull(rightPosition);
+
+ if (leftNull != rightNull) {
+ return false;
+ }
+ if (leftNull) {
+ continue;
+ }
+
+ switch (types[i].getTypeEnum()) {
+ case BOOLEAN:
+ if (leftBlock[i].getBoolean(leftPosition) !=
rightBlock[i].getBoolean(rightPosition)) {
+ return false;
+ }
+ break;
+ case INT32:
+ case DATE:
+ if (leftBlock[i].getInt(leftPosition) !=
rightBlock[i].getInt(rightPosition)) {
+ return false;
+ }
+ break;
+ case INT64:
+ case TIMESTAMP:
+ if (leftBlock[i].getLong(leftPosition) !=
rightBlock[i].getLong(rightPosition)) {
+ return false;
+ }
+ break;
+ case FLOAT:
+ if (leftBlock[i].getFloat(leftPosition) !=
rightBlock[i].getFloat(rightPosition)) {
+ return false;
+ }
+ break;
+ case DOUBLE:
+ if (leftBlock[i].getDouble(leftPosition) !=
rightBlock[i].getDouble(rightPosition)) {
+ return false;
+ }
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ if (!leftBlock[i]
+ .getBinary(leftPosition)
+ .equals(rightBlock[i].getBinary(rightPosition))) {
+ return false;
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ return true;
+ }
+
+ public static void appendTo(Type type, Column input, int position,
ColumnBuilder columnBuilder) {
Review Comment:
don't need this method, you can call `ColumnBuilder.write(Column column, int
index)`, but remember to do
if (input.isNull(position)) {
columnBuilder.appendNull();
}
before calling `ColumnBuilder.write(Column column, int index)` method
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/builder/InMemoryHashAggregationBuilder.java:
##########
@@ -120,6 +131,18 @@ public void updateMemory() {
// updateMemory.update();
}
+ @Override
+ public void reset() {
+ // TODO reset of GroupByHash
+ groupByHash =
+ createGroupByHash(
+ groupByOutputTypes, hashChannel.isPresent(), expectedGroups,
updateMemory);
Review Comment:
why not adding a reset method in `GroupByHash`? `Trino` don't have that?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped;
+
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
+import org.apache.iotdb.db.queryengine.execution.operator.AbstractOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
+import
org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager;
+
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.appendTo;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.rowNotDistinctFromRow;
+
+public class StreamingAggregationOperator extends AbstractOperator {
+ private static final long INSTANCE_SIZE =
+
RamUsageEstimator.shallowSizeOfInstance(StreamingAggregationOperator.class);
+
+ private final OperatorContext operatorContext;
+
+ private final Operator child;
+
+ private final List<TableAggregator> aggregators;
+
+ private final Type[] groupByTypes;
+ private final int[] groupByChannels;
+
+ private final TsBlockBuilder resultBuilder;
+ private final ColumnBuilder[] resultColumnsBuilder;
+
+ private boolean finished = false;
+
+ // cached current group to judge row equality
+ private TsBlock currentGroup;
+
+ // more than one group in input block
+ private final Deque<TsBlock> outputs = new LinkedList<>();
+
+ public StreamingAggregationOperator(
+ OperatorContext operatorContext,
+ Operator child,
+ List<Type> groupByTypes,
+ List<Integer> groupByChannels,
+ List<TableAggregator> aggregators,
+ long maxPartialMemory,
+ boolean spillEnabled,
+ long unSpillMemoryLimit) {
+ this.operatorContext = operatorContext;
+ this.child = child;
+ this.groupByTypes = groupByTypes.toArray(new Type[0]);
+ this.groupByChannels = Ints.toArray(groupByChannels);
+ this.aggregators = aggregators;
+ this.resultBuilder =
+ new TsBlockBuilder(
+ Stream.concat(
+
groupByTypes.stream().map(InternalTypeManager::getTSDataType),
+ aggregators.stream().map(TableAggregator::getType))
+ .collect(Collectors.toList()));
+ this.resultColumnsBuilder = resultBuilder.getValueColumnBuilders();
+ checkArgument(!spillEnabled, "spill is not supported");
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ return !finished || retainedTsBlock != null || !outputs.isEmpty();
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ if (retainedTsBlock != null) {
+ return getResultFromRetainedTsBlock();
+ }
+
+ if (!outputs.isEmpty()) {
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ TsBlock block;
+ if (child.hasNextWithTimer()) {
+ block = child.nextWithTimer();
+ if (block == null) {
+ return null;
+ }
+
+ processInput(block);
+
+ if (outputs.isEmpty()) {
+ return null;
+ }
+ } else {
+ // last evaluate
+ if (currentGroup != null) {
+ evaluateAndFlushGroup(currentGroup, 0, true);
Review Comment:
really need to evaluate it again? For my understanding currentGroup is just
kept to be comapred to next TsBlock, it must be totally process on its own
iteration.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformAggregationToStreamable.java:
##########
@@ -50,7 +50,7 @@ public class TransformAggregationToStreamable implements
PlanOptimizer {
@Override
public PlanNode optimize(PlanNode plan, PlanOptimizer.Context context) {
if (!(context.getAnalysis().getStatement() instanceof Query)
- || context.getAnalysis().noAggregates()) {
+ || !context.getAnalysis().isAggregationQuery()) {
Review Comment:
!isAggregationQuery is not equal to previous `noAggregates`
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAvgAccumulator.java:
##########
@@ -88,7 +89,7 @@ public void addInput(int[] groupIds, Column[] arguments) {
@Override
public void addIntermediate(int[] groupIds, Column argument) {
checkArgument(
- argument instanceof BinaryColumn,
+ argument instanceof BinaryColumn || argument instanceof
RunLengthEncodedColumn,
Review Comment:
also need to do argument.getValue() instanceof BinaryColumn if it is
`RunLengthEncodedColumn`.
So, why not directly using argument.getDataType() == BLOB
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped;
+
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
+import org.apache.iotdb.db.queryengine.execution.operator.AbstractOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
+import
org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager;
+
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.appendTo;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.rowNotDistinctFromRow;
+
+public class StreamingAggregationOperator extends AbstractOperator {
+ private static final long INSTANCE_SIZE =
+
RamUsageEstimator.shallowSizeOfInstance(StreamingAggregationOperator.class);
+
+ private final OperatorContext operatorContext;
+
+ private final Operator child;
+
+ private final List<TableAggregator> aggregators;
+
+ private final Type[] groupByTypes;
+ private final int[] groupByChannels;
+
+ private final TsBlockBuilder resultBuilder;
+ private final ColumnBuilder[] resultColumnsBuilder;
+
+ private boolean finished = false;
+
+ // cached current group to judge row equality
+ private TsBlock currentGroup;
+
+ // more than one group in input block
+ private final Deque<TsBlock> outputs = new LinkedList<>();
+
+ public StreamingAggregationOperator(
+ OperatorContext operatorContext,
+ Operator child,
+ List<Type> groupByTypes,
+ List<Integer> groupByChannels,
+ List<TableAggregator> aggregators,
+ long maxPartialMemory,
+ boolean spillEnabled,
+ long unSpillMemoryLimit) {
+ this.operatorContext = operatorContext;
+ this.child = child;
+ this.groupByTypes = groupByTypes.toArray(new Type[0]);
+ this.groupByChannels = Ints.toArray(groupByChannels);
+ this.aggregators = aggregators;
+ this.resultBuilder =
+ new TsBlockBuilder(
+ Stream.concat(
+
groupByTypes.stream().map(InternalTypeManager::getTSDataType),
+ aggregators.stream().map(TableAggregator::getType))
+ .collect(Collectors.toList()));
+ this.resultColumnsBuilder = resultBuilder.getValueColumnBuilders();
+ checkArgument(!spillEnabled, "spill is not supported");
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ return !finished || retainedTsBlock != null || !outputs.isEmpty();
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ if (retainedTsBlock != null) {
+ return getResultFromRetainedTsBlock();
+ }
+
+ if (!outputs.isEmpty()) {
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ TsBlock block;
+ if (child.hasNextWithTimer()) {
+ block = child.nextWithTimer();
+ if (block == null) {
+ return null;
+ }
+
+ processInput(block);
+
+ if (outputs.isEmpty()) {
+ return null;
+ }
+ } else {
+ // last evaluate
+ if (currentGroup != null) {
+ evaluateAndFlushGroup(currentGroup, 0, true);
+ currentGroup = null;
+ }
+ finished = true;
+ }
+ resultTsBlock = outputs.removeFirst();
Review Comment:
what if outputs is empty?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analysis.java:
##########
@@ -411,6 +411,10 @@ public boolean isAggregation(QuerySpecification node) {
return groupingSets.containsKey(NodeRef.of(node));
}
+ public boolean isAggregationQuery() {
+ return !groupingSets.isEmpty() || !aggregates.isEmpty();
+ }
+
Review Comment:
this method seems not to be equal to previous !noAggregates()?
Why adding this method, and other places call noAggregates right?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/utils/TypeUtil.java:
##########
@@ -499,4 +499,96 @@ public static long hash(Type type, Column column, int
position) {
public static FlatHashStrategy getFlatHashStrategy(List<Type> hashTypes) {
return new HashStrategy(hashTypes);
}
+
+ public static boolean rowNotDistinctFromRow(
Review Comment:
don't need this method, you can refer to groupKeyComparator in
`AbstractGapFillWGroupOperator`
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped;
+
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
+import org.apache.iotdb.db.queryengine.execution.operator.AbstractOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
+import
org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager;
+
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.appendTo;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.rowNotDistinctFromRow;
+
+public class StreamingAggregationOperator extends AbstractOperator {
+ private static final long INSTANCE_SIZE =
+
RamUsageEstimator.shallowSizeOfInstance(StreamingAggregationOperator.class);
+
+ private final OperatorContext operatorContext;
+
+ private final Operator child;
+
+ private final List<TableAggregator> aggregators;
+
+ private final Type[] groupByTypes;
+ private final int[] groupByChannels;
+
+ private final TsBlockBuilder resultBuilder;
+ private final ColumnBuilder[] resultColumnsBuilder;
+
+ private boolean finished = false;
+
+ // cached current group to judge row equality
+ private TsBlock currentGroup;
+
+ // more than one group in input block
+ private final Deque<TsBlock> outputs = new LinkedList<>();
+
+ public StreamingAggregationOperator(
+ OperatorContext operatorContext,
+ Operator child,
+ List<Type> groupByTypes,
+ List<Integer> groupByChannels,
+ List<TableAggregator> aggregators,
+ long maxPartialMemory,
+ boolean spillEnabled,
+ long unSpillMemoryLimit) {
+ this.operatorContext = operatorContext;
+ this.child = child;
+ this.groupByTypes = groupByTypes.toArray(new Type[0]);
+ this.groupByChannels = Ints.toArray(groupByChannels);
+ this.aggregators = aggregators;
+ this.resultBuilder =
+ new TsBlockBuilder(
+ Stream.concat(
+
groupByTypes.stream().map(InternalTypeManager::getTSDataType),
+ aggregators.stream().map(TableAggregator::getType))
+ .collect(Collectors.toList()));
+ this.resultColumnsBuilder = resultBuilder.getValueColumnBuilders();
+ checkArgument(!spillEnabled, "spill is not supported");
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ return !finished || retainedTsBlock != null || !outputs.isEmpty();
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ if (retainedTsBlock != null) {
+ return getResultFromRetainedTsBlock();
+ }
+
+ if (!outputs.isEmpty()) {
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ TsBlock block;
+ if (child.hasNextWithTimer()) {
+ block = child.nextWithTimer();
+ if (block == null) {
+ return null;
+ }
+
+ processInput(block);
+
+ if (outputs.isEmpty()) {
+ return null;
+ }
+ } else {
+ // last evaluate
+ if (currentGroup != null) {
+ evaluateAndFlushGroup(currentGroup, 0, true);
+ currentGroup = null;
+ }
+ finished = true;
+ }
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ private void processInput(TsBlock page) {
+ requireNonNull(page, "page is null");
+
+ Column[] groupByPage = page.getColumns(groupByChannels);
+ if (currentGroup != null) {
+ if (!rowNotDistinctFromRow(
+ groupByTypes, 0, currentGroup.getColumns(groupByChannels), 0,
groupByPage)) {
+ // page starts with new group, so flush it
+ evaluateAndFlushGroup(currentGroup, 0, false);
+ }
+ currentGroup = null;
+ }
+
+ int startPosition = 0;
+ while (true) {
+ // may be equal to page.getPositionCount() if the end is not found in
this page
+ int nextGroupStart = findNextGroupStart(startPosition, groupByPage);
+ addRowsToAggregators(page, startPosition, nextGroupStart - 1);
+
+ if (nextGroupStart < page.getPositionCount()) {
+ // current group stops somewhere in the middle of the page, so flush it
+ evaluateAndFlushGroup(page, startPosition, false);
+ startPosition = nextGroupStart;
+ } else {
+ // late materialization requires that page being locally stored is
materialized before the
+ // next one is fetched
+ currentGroup = page.getRegion(page.getPositionCount() - 1, 1);
+ return;
+ }
+ }
+ }
+
+ private void addRowsToAggregators(TsBlock page, int startPosition, int
endPosition) {
+ TsBlock region = page.getRegion(startPosition, endPosition - startPosition
+ 1);
+ for (TableAggregator aggregator : aggregators) {
+ aggregator.processBlock(region);
+ }
+ }
+
+ private void resetAggregationBuilder() {
+ for (TableAggregator aggregator : aggregators) {
+ aggregator.reset();
+ }
+ }
+
+ private void evaluateAndFlushGroup(TsBlock page, int position, boolean
lastCalculate) {
+ resultBuilder.declarePosition();
+ for (int i = 0; i < groupByTypes.length; i++) {
+ Column input = page.getColumn(groupByChannels[i]);
+ Type type = groupByTypes[i];
+ appendTo(type, input, position, resultColumnsBuilder[i]);
+ }
+ int offset = groupByTypes.length;
+ for (int i = 0; i < aggregators.size(); i++) {
+ aggregators.get(i).evaluate(resultColumnsBuilder[offset + i]);
+ }
+
+ if (lastCalculate || resultBuilder.isFull()) {
+ Column[] result = new Column[resultColumnsBuilder.length];
+ for (int i = 0; i < resultColumnsBuilder.length; i++) {
+ result[i] = resultColumnsBuilder[i].build();
+ }
+
+ outputs.add(
+ TsBlock.wrapBlocksWithoutCopy(
+ resultBuilder.getPositionCount(),
+ new RunLengthEncodedColumn(
+ TableScanOperator.TIME_COLUMN_TEMPLATE,
resultBuilder.getPositionCount()),
+ result));
Review Comment:
```suggestion
outputs.add(resultBuilder.build(new RunLengthEncodedColumn(
TableScanOperator.TIME_COLUMN_TEMPLATE,
resultBuilder.getPositionCount())));
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped;
+
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
+import org.apache.iotdb.db.queryengine.execution.operator.AbstractOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
+import
org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager;
+
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.appendTo;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.rowNotDistinctFromRow;
+
+public class StreamingAggregationOperator extends AbstractOperator {
+ private static final long INSTANCE_SIZE =
+
RamUsageEstimator.shallowSizeOfInstance(StreamingAggregationOperator.class);
+
+ private final OperatorContext operatorContext;
+
+ private final Operator child;
+
+ private final List<TableAggregator> aggregators;
+
+ private final Type[] groupByTypes;
+ private final int[] groupByChannels;
+
+ private final TsBlockBuilder resultBuilder;
+ private final ColumnBuilder[] resultColumnsBuilder;
+
+ private boolean finished = false;
+
+ // cached current group to judge row equality
+ private TsBlock currentGroup;
Review Comment:
better to use SortKey and Comparator<SortKey> as
AbstractGapFillWGroupOperator to do so.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped;
+
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
+import org.apache.iotdb.db.queryengine.execution.operator.AbstractOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
+import
org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager;
+
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.appendTo;
+import static
org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.rowNotDistinctFromRow;
+
+public class StreamingAggregationOperator extends AbstractOperator {
+ private static final long INSTANCE_SIZE =
+
RamUsageEstimator.shallowSizeOfInstance(StreamingAggregationOperator.class);
+
+ private final OperatorContext operatorContext;
+
+ private final Operator child;
+
+ private final List<TableAggregator> aggregators;
+
+ private final Type[] groupByTypes;
+ private final int[] groupByChannels;
+
+ private final TsBlockBuilder resultBuilder;
+ private final ColumnBuilder[] resultColumnsBuilder;
+
+ private boolean finished = false;
+
+ // cached current group to judge row equality
+ private TsBlock currentGroup;
+
+ // more than one group in input block
+ private final Deque<TsBlock> outputs = new LinkedList<>();
+
+ public StreamingAggregationOperator(
+ OperatorContext operatorContext,
+ Operator child,
+ List<Type> groupByTypes,
+ List<Integer> groupByChannels,
+ List<TableAggregator> aggregators,
+ long maxPartialMemory,
+ boolean spillEnabled,
+ long unSpillMemoryLimit) {
+ this.operatorContext = operatorContext;
+ this.child = child;
+ this.groupByTypes = groupByTypes.toArray(new Type[0]);
+ this.groupByChannels = Ints.toArray(groupByChannels);
+ this.aggregators = aggregators;
+ this.resultBuilder =
+ new TsBlockBuilder(
+ Stream.concat(
+
groupByTypes.stream().map(InternalTypeManager::getTSDataType),
+ aggregators.stream().map(TableAggregator::getType))
+ .collect(Collectors.toList()));
+ this.resultColumnsBuilder = resultBuilder.getValueColumnBuilders();
+ checkArgument(!spillEnabled, "spill is not supported");
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ return !finished || retainedTsBlock != null || !outputs.isEmpty();
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ if (retainedTsBlock != null) {
+ return getResultFromRetainedTsBlock();
+ }
+
+ if (!outputs.isEmpty()) {
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ TsBlock block;
+ if (child.hasNextWithTimer()) {
+ block = child.nextWithTimer();
+ if (block == null) {
+ return null;
+ }
+
+ processInput(block);
+
+ if (outputs.isEmpty()) {
+ return null;
+ }
+ } else {
+ // last evaluate
+ if (currentGroup != null) {
+ evaluateAndFlushGroup(currentGroup, 0, true);
+ currentGroup = null;
+ }
+ finished = true;
+ }
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ private void processInput(TsBlock page) {
+ requireNonNull(page, "page is null");
+
+ Column[] groupByPage = page.getColumns(groupByChannels);
+ if (currentGroup != null) {
+ if (!rowNotDistinctFromRow(
+ groupByTypes, 0, currentGroup.getColumns(groupByChannels), 0,
groupByPage)) {
+ // page starts with new group, so flush it
+ evaluateAndFlushGroup(currentGroup, 0, false);
+ }
+ currentGroup = null;
+ }
+
+ int startPosition = 0;
+ while (true) {
+ // may be equal to page.getPositionCount() if the end is not found in
this page
+ int nextGroupStart = findNextGroupStart(startPosition, groupByPage);
+ addRowsToAggregators(page, startPosition, nextGroupStart - 1);
+
+ if (nextGroupStart < page.getPositionCount()) {
+ // current group stops somewhere in the middle of the page, so flush it
+ evaluateAndFlushGroup(page, startPosition, false);
+ startPosition = nextGroupStart;
+ } else {
+ // late materialization requires that page being locally stored is
materialized before the
+ // next one is fetched
+ currentGroup = page.getRegion(page.getPositionCount() - 1, 1);
+ return;
+ }
+ }
+ }
+
+ private void addRowsToAggregators(TsBlock page, int startPosition, int
endPosition) {
+ TsBlock region = page.getRegion(startPosition, endPosition - startPosition
+ 1);
+ for (TableAggregator aggregator : aggregators) {
+ aggregator.processBlock(region);
+ }
+ }
+
+ private void resetAggregationBuilder() {
+ for (TableAggregator aggregator : aggregators) {
+ aggregator.reset();
+ }
+ }
+
+ private void evaluateAndFlushGroup(TsBlock page, int position, boolean
lastCalculate) {
Review Comment:
Do we really need this `lastCalculate` parameter?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
