You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2024/10/10 12:53:56 UTC
(iotdb) branch agg_table_scan updated: perfect agg table scan
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch agg_table_scan
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/agg_table_scan by this push:
new a206dd401ac perfect agg table scan
a206dd401ac is described below
commit a206dd401ac1a63cb03a7103f9d4432008d212e1
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Thu Oct 10 20:53:41 2024 +0800
perfect agg table scan
---
.../TableAggregationTableScanOperator.java | 88 +++++++++++++++++++---
.../relational/aggregation/TableAggregator.java | 20 ++---
.../plan/planner/TableOperatorGenerator.java | 29 +++++--
.../metadata/TableBuiltinAggregationFunction.java | 4 +
4 files changed, 111 insertions(+), 30 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
index 6829c9c0fa6..8a78a1a84ee 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
import org.apache.iotdb.commons.path.AlignedFullPath;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITimeRangeIterator;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractSeriesAggregationScanOperator;
@@ -36,6 +38,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource;
import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+import org.apache.commons.lang3.stream.Streams;
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
@@ -46,6 +49,7 @@ import org.apache.tsfile.read.common.block.TsBlockBuilder;
import org.apache.tsfile.read.common.block.column.LongColumn;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import java.io.IOException;
@@ -64,6 +68,9 @@ import static org.apache.tsfile.read.common.block.TsBlockUtil.skipPointsOutOfTim
public class TableAggregationTableScanOperator extends AbstractSeriesAggregationScanOperator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(TableAggregationTableScanOperator.class);
+
private final List<TableAggregator> aggregators;
public static final LongColumn TIME_COLUMN_TEMPLATE =
@@ -224,7 +231,12 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
@Override
public long ramBytesUsed() {
- return 0;
+ return INSTANCE_SIZE
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(seriesScanUtil)
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(sourceId)
+ + (resultTsBlockBuilder == null ? 0 : resultTsBlockBuilder.getRetainedSizeInBytes())
+ + RamUsageEstimator.sizeOfCollection(deviceEntries);
}
private AlignedSeriesScanUtil constructAlignedSeriesScanUtil(DeviceEntry deviceEntry) {
@@ -322,7 +334,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
* @return left - whether the aggregation calculation of the current time range has done; right -
* remaining tsBlock
*/
- public static Pair<Boolean, TsBlock> calculateAggregationFromRawData(
+ public Pair<Boolean, TsBlock> calculateAggregationFromRawData(
TsBlock inputTsBlock,
List<TableAggregator> aggregators,
TimeRange curTimeRange,
@@ -352,7 +364,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
isAllAggregatorsHasFinalResult(aggregators) || isTsBlockOutOfBound, inputTsBlock);
}
- private static TsBlock process(
+ private TsBlock process(
TsBlock inputTsBlock, TimeRange curTimeRange, List<TableAggregator> aggregators) {
// Get the row which need to be processed by aggregator
IWindow curWindow = new TimeWindow(curTimeRange);
@@ -366,11 +378,27 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
}
TsBlock inputRegion = inputTsBlock.getRegion(0, lastIndexToProcess + 1);
- for (TableAggregator aggregator : aggregators) {
+ for (int i = 0; i < aggregators.size(); i++) {
+ // ID or Attribute is null, skip this aggregation logic
+ if (TsTableColumnCategory.ID == columnSchemas.get(i).getColumnCategory()
+ && deviceEntries.get(currentDeviceIndex).getNthSegment(columnsIndexArray[i]) == null) {
+ continue;
+ }
+ if (TsTableColumnCategory.ATTRIBUTE == columnSchemas.get(i).getColumnCategory()
+ && deviceEntries
+ .get(currentDeviceIndex)
+ .getAttributeColumnValues()
+ .get(columnsIndexArray[i])
+ == null) {
+ continue;
+ }
+
+ TableAggregator aggregator = aggregators.get(i);
// current agg method has been calculated
if (aggregator.hasFinalResult()) {
continue;
}
+
aggregator.processBlock(inputRegion);
}
int lastReadRowIndex = lastIndexToProcess + 1;
@@ -391,11 +419,30 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
}
protected void calcFromStatistics(Statistics timeStatistics, Statistics[] valueStatistics) {
- for (TableAggregator aggregator : aggregators) {
+ for (int i = 0; i < aggregators.size(); i++) {
+ // ID or Attribute is null, skip this aggregation logic
+ if (TsTableColumnCategory.ID == columnSchemas.get(i).getColumnCategory()
+ && deviceEntries.get(currentDeviceIndex).getNthSegment(columnsIndexArray[i]) == null) {
+ continue;
+ }
+ if (TsTableColumnCategory.ATTRIBUTE == columnSchemas.get(i).getColumnCategory()
+ && deviceEntries
+ .get(currentDeviceIndex)
+ .getAttributeColumnValues()
+ .get(columnsIndexArray[i])
+ == null) {
+ continue;
+ }
+
+ TableAggregator aggregator = aggregators.get(i);
if (aggregator.hasFinalResult()) {
continue;
}
- aggregator.processStatistics(valueStatistics);
+ aggregator.processStatistics(
+ columnSchemas.get(i).getColumnCategory() == TsTableColumnCategory.MEASUREMENT
+ ? valueStatistics[i]
+ : timeStatistics);
+ // aggregator.processStatistics(valueStatistics);
}
}
@@ -520,8 +567,31 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
}
// calc from page data
- TsBlock tsBlock = seriesScanUtil.nextPage();
- if (tsBlock == null || tsBlock.isEmpty()) {
+ TsBlock originalTsBlock = seriesScanUtil.nextPage();
+ if (originalTsBlock == null) {
+ continue;
+ }
+ // TODO add optimization: if only agg measurement columns, no need to transfer
+ Column[] valueColumns = new Column[aggregators.size()];
+ for (int i = 0; i < columnsIndexArray.length; i++) {
+ ColumnSchema columnSchema = columnSchemas.get(i);
+ if (Streams.of(
+ TsTableColumnCategory.ID,
+ TsTableColumnCategory.ATTRIBUTE,
+ TsTableColumnCategory.TIME)
+ .anyMatch(columnSchema.getColumnCategory()::equals)) {
+ valueColumns[i] = originalTsBlock.getTimeColumn();
+ } else {
+ valueColumns[i] = originalTsBlock.getColumn(i);
+ }
+ }
+ TsBlock tsBlock =
+ new TsBlock(
+ originalTsBlock.getPositionCount(),
+ new RunLengthEncodedColumn(
+ TIME_COLUMN_TEMPLATE, originalTsBlock.getPositionCount()),
+ valueColumns);
+ if (tsBlock.isEmpty()) {
continue;
}
@@ -547,7 +617,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
}
/** Append a row of aggregation results to the result tsBlock. */
- public static void appendAggregationResult(
+ public void appendAggregationResult(
TsBlockBuilder tsBlockBuilder, List<? extends TableAggregator> aggregators) {
// TimeColumnBuilder timeColumnBuilder = tsBlockBuilder.getTimeColumnBuilder();
// Use start time of current time range as time column
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
index 8e9726d55ae..3f73d15345f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
@@ -27,8 +27,6 @@ import java.util.OptionalInt;
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
-import static org.apache.iotdb.db.queryengine.execution.aggregation.TreeAggregator.QUERY_EXECUTION_METRICS;
-import static org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.AGGREGATION_FROM_STATISTICS;
public class TableAggregator {
private final Accumulator accumulator;
@@ -74,19 +72,11 @@ public class TableAggregator {
}
}
- /** Used for SeriesAggregateScanOperator. */
- public void processStatistics(Statistics[] valueStatistics) {
- long startTime = System.nanoTime();
- try {
- // TODO verify the rightness
- for (int valueIndex : inputChannels) {
- // int valueIndex = inputLocations[0].getValueColumnIndex();
- accumulator.addStatistics(valueStatistics[valueIndex]);
- }
- } finally {
- QUERY_EXECUTION_METRICS.recordExecutionCost(
- AGGREGATION_FROM_STATISTICS, System.nanoTime() - startTime);
- }
+ /** Used for AggregateTableScanOperator. */
+ public void processStatistics(Statistics valueStatistics) {
+ // TODO verify the rightness
+ checkArgument(inputChannels.length == 1, "expected 1 input channel for processStatistics");
+ accumulator.addStatistics(valueStatistics);
}
public boolean hasFinalResult() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
index 43a4defd960..412bec723d3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
@@ -124,7 +124,10 @@ import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
+import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.ATTRIBUTE;
+import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.ID;
import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.MEASUREMENT;
+import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TIME;
import static org.apache.iotdb.db.queryengine.common.DataNodeEndPoints.isSameNode;
import static org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.MergeSortComparator.getComparatorForTable;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.constructAlignedPath;
@@ -1009,8 +1012,8 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return new TableAggregator(
accumulator,
step,
- getTSDataType(aggregation.getResolvedFunction().getSignature().getReturnType()),
- // getTSDataType(typeProvider.getTableModelType(aggregationSymbol)),
+ // getTSDataType(aggregation.getResolvedFunction().getSignature().getReturnType()),
+ getTSDataType(typeProvider.getTableModelType(aggregationSymbol)),
argumentChannels,
OptionalInt.empty());
}
@@ -1026,11 +1029,10 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
node.getPlanNodeId(),
AggregationTableScanNode.class.getSimpleName());
- List<TableAggregator> aggregators = new ArrayList<>();
-
- // TODO how to use the output symbols of AggregationTableScan?
- Map<Symbol, Integer> childLayout = makeLayoutFromOutputSymbols(node.getAssignments().keySet());
+ List<TableAggregator> aggregators = new ArrayList<>(node.getAggregations().size());
+ // TODO is assignments of AggTableScane hashmap ok?
+ Map<Symbol, Integer> childLayout = makeLayoutForAggTableScan(node);
for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : node.getAggregations().entrySet()) {
TableAggregator aggregator =
buildAggregator(
@@ -1148,6 +1150,21 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return aggTableScanOperator;
}
+ private Map<Symbol, Integer> makeLayoutForAggTableScan(AggregationTableScanNode node) {
+ Map<Symbol, Integer> childLayout;
+ ImmutableMap.Builder<Symbol, Integer> outputMappings = ImmutableMap.builder();
+ int channel = 0;
+
+ // TODO for aggregation function which has more than one arguements, this may by wrong
+ for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : node.getAggregations().entrySet()) {
+ Symbol symbol = Symbol.from(entry.getValue().getArguments().get(0));
+ outputMappings.put(symbol, channel);
+ channel++;
+ }
+ childLayout = outputMappings.buildOrThrow();
+ return childLayout;
+ }
+
public static long calculateMaxAggregationResultSize(
// List<? extends AggregationDescriptor> aggregationDescriptors,
// ITimeRangeIterator timeRangeIterator
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableBuiltinAggregationFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableBuiltinAggregationFunction.java
index 72f0cdadd02..eb45eb8f4a3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableBuiltinAggregationFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableBuiltinAggregationFunction.java
@@ -32,6 +32,7 @@ import java.util.stream.Collectors;
import static org.apache.tsfile.read.common.type.DoubleType.DOUBLE;
import static org.apache.tsfile.read.common.type.IntType.INT32;
+import static org.apache.tsfile.read.common.type.LongType.INT64;
public enum TableBuiltinAggregationFunction {
SUM("sum"),
@@ -111,7 +112,10 @@ public enum TableBuiltinAggregationFunction {
public static List<Type> getIntermediateTypes(String name, List<Type> originalArgumentTypes) {
if (AVG.functionName.equalsIgnoreCase(name)) {
return ImmutableList.of(DOUBLE, INT32);
+ } else if (COUNT.functionName.equalsIgnoreCase(name)) {
+ return ImmutableList.of(INT64);
} else {
+ // TODO consider other aggregations which changed the result type
return ImmutableList.copyOf(originalArgumentTypes);
}
}