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/09/30 14:59:03 UTC
(iotdb) 04/05: tmp
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
commit 696365082795aef09768033269b0c742a0a1cb09
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Sun Sep 29 18:16:39 2024 +0800
tmp
---
.../TableAggregationTableScanOperator.java | 226 +++++++++++++++++++--
.../plan/planner/TableOperatorGenerator.java | 168 ++++++++++++++-
2 files changed, 377 insertions(+), 17 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 8d548e5e79e..4a0b2873ffa 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
@@ -19,25 +19,75 @@
package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
-import org.apache.iotdb.db.queryengine.execution.aggregation.Aggregator;
+import org.apache.iotdb.commons.path.AlignedFullPath;
+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;
+import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanUtil;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.Aggregator;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByTimeParameter;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
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.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.LongColumn;
import org.apache.tsfile.write.schema.IMeasurementSchema;
+import java.io.IOException;
import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
-public class TableAggregationTableScanOperator extends TableScanOperator {
+import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.appendAggregationResult;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.constructAlignedPath;
+
+public class TableAggregationTableScanOperator extends AbstractSeriesAggregationScanOperator {
List<Aggregator> aggregators;
+ public static final LongColumn TIME_COLUMN_TEMPLATE =
+ new LongColumn(1, Optional.empty(), new long[] {0});
+
+ private final List<ColumnSchema> columnSchemas;
+
+ private final int[] columnsIndexArray;
+
+ private final int measurementColumnCount;
+
+ private final List<DeviceEntry> deviceEntries;
+
+ private final int deviceCount;
+
+ private final Ordering scanOrder;
+ private final SeriesScanOptions seriesScanOptions;
+
+ private final List<String> measurementColumnNames;
+
+ private final List<IMeasurementSchema> measurementSchemas;
+
+ private final List<TSDataType> measurementColumnTSDataTypes;
+
+ private TsBlockBuilder measurementDataBuilder;
+
+ // TODO calc maxTsBlockLineNum using date_bin
+ private final int maxTsBlockLineNum;
+
+ private TsBlock measurementDataBlock;
+
+ private QueryDataSource queryDataSource;
+
+ private int currentDeviceIndex;
+
public TableAggregationTableScanOperator(
- OperatorContext context,
PlanNodeId sourceId,
+ OperatorContext context,
List<ColumnSchema> columnSchemas,
int[] columnsIndexArray,
int measurementColumnCount,
@@ -46,19 +96,47 @@ public class TableAggregationTableScanOperator extends TableScanOperator {
SeriesScanOptions seriesScanOptions,
List<String> measurementColumnNames,
List<IMeasurementSchema> measurementSchemas,
- int maxTsBlockLineNum) {
+ int maxTsBlockLineNum,
+ int subSensorSize,
+ List<Aggregator> aggregators,
+ ITimeRangeIterator timeRangeIterator,
+ boolean ascending,
+ GroupByTimeParameter groupByTimeParameter,
+ long maxReturnSize,
+ boolean canUseStatistics) {
+
super(
- context,
sourceId,
- columnSchemas,
- columnsIndexArray,
- measurementColumnCount,
- deviceEntries,
- scanOrder,
- seriesScanOptions,
- measurementColumnNames,
- measurementSchemas,
- maxTsBlockLineNum);
+ context,
+ null,
+ subSensorSize,
+ aggregators,
+ timeRangeIterator,
+ ascending,
+ false,
+ groupByTimeParameter,
+ maxReturnSize,
+ canUseStatistics);
+
+ this.sourceId = sourceId;
+ this.operatorContext = context;
+ this.columnSchemas = columnSchemas;
+ this.columnsIndexArray = columnsIndexArray;
+ this.measurementColumnCount = measurementColumnCount;
+ this.deviceEntries = deviceEntries;
+ this.deviceCount = deviceEntries.size();
+ this.scanOrder = scanOrder;
+ this.seriesScanOptions = seriesScanOptions;
+ this.measurementColumnNames = measurementColumnNames;
+ this.measurementSchemas = measurementSchemas;
+ this.measurementColumnTSDataTypes =
+ measurementSchemas.stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
+ this.currentDeviceIndex = 0;
+
+ this.maxReturnSize = maxReturnSize;
+ this.maxTsBlockLineNum = maxTsBlockLineNum;
+
+ this.seriesScanUtil = constructAlignedSeriesScanUtil(deviceEntries.get(currentDeviceIndex));
}
@Override
@@ -66,14 +144,130 @@ public class TableAggregationTableScanOperator extends TableScanOperator {
return !isFinished();
}
+ @Override
+ public TsBlock next() throws Exception {
+ // start stopwatch, reset leftRuntimeOfOneNextCall
+ long start = System.nanoTime();
+ // leftRuntimeOfOneNextCall = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
+ // long maxRuntime = leftRuntimeOfOneNextCall;
+
+ while (
+ // System.nanoTime() - start < maxRuntime&&
+ (curTimeRange != null || timeRangeIterator.hasNextTimeRange())
+ && !resultTsBlockBuilder.isFull()) {
+ if (curTimeRange == null) {
+ // move to the next time window
+ curTimeRange = timeRangeIterator.nextTimeRange();
+ // clear previous aggregation result
+ for (Aggregator aggregator : aggregators) {
+ aggregator.reset();
+ }
+ }
+
+ // calculate aggregation result on current time window
+ // Keep curTimeRange if the calculation of this timeRange is not done
+ if (calculateAggregationResultForCurrentTimeRange()) {
+ curTimeRange = null;
+ }
+ }
+
+ if (resultTsBlockBuilder.getPositionCount() > 0) {
+ TsBlock resultTsBlock = resultTsBlockBuilder.build();
+ resultTsBlockBuilder.reset();
+ return resultTsBlock;
+ } else {
+ return null;
+ }
+ }
+
@Override
public boolean isFinished() throws Exception {
- return (retainedTsBlock == null);
- // && (currentDeviceIndex >= deviceCount || seriesScanOptions.limitConsumedUp());
+ return (retainedTsBlock == null)
+ && (currentDeviceIndex >= deviceCount || seriesScanOptions.limitConsumedUp());
}
@Override
public long ramBytesUsed() {
return 0;
}
+
+ private AlignedSeriesScanUtil constructAlignedSeriesScanUtil(DeviceEntry deviceEntry) {
+ AlignedFullPath alignedPath =
+ constructAlignedPath(deviceEntry, measurementColumnNames, measurementSchemas);
+
+ return new AlignedSeriesScanUtil(
+ alignedPath,
+ scanOrder,
+ seriesScanOptions,
+ operatorContext.getInstanceContext(),
+ true,
+ measurementColumnTSDataTypes);
+ }
+
+ /** Return true if we have the result of this timeRange. */
+ protected boolean calculateAggregationResultForCurrentTimeRange() {
+ try {
+ if (calcFromCachedData()) {
+ updateResultTsBlock();
+ return true;
+ }
+
+ if (readAndCalcFromPage()) {
+ updateResultTsBlock();
+ return true;
+ }
+
+ // only when all the page data has been consumed, we need to read the chunk data
+ if (!seriesScanUtil.hasNextPage() && readAndCalcFromChunk()) {
+ updateResultTsBlock();
+ return true;
+ }
+
+ // only when all the page and chunk data has been consumed, we need to read the file data
+ if (!seriesScanUtil.hasNextPage()
+ && !seriesScanUtil.hasNextChunk()
+ && readAndCalcFromFile()) {
+ updateResultTsBlock();
+ return true;
+ }
+
+ // If the TimeRange is (Long.MIN_VALUE, Long.MAX_VALUE), for Aggregators like countAggregator,
+ // we have to consume all the data before we finish the aggregation calculation.
+ if (seriesScanUtil.hasNextPage()
+ || seriesScanUtil.hasNextChunk()
+ || seriesScanUtil.hasNextFile()) {
+ return false;
+ } else {
+ currentDeviceIndex++;
+ }
+ updateResultTsBlock();
+ if (currentDeviceIndex < deviceCount) {
+ // construct AlignedSeriesScanUtil for next device
+ this.seriesScanUtil = constructAlignedSeriesScanUtil(deviceEntries.get(currentDeviceIndex));
+
+ // reset QueryDataSource
+ queryDataSource.reset();
+ this.seriesScanUtil.initQueryDataSource(queryDataSource);
+ }
+ return currentDeviceIndex >= deviceCount;
+ } catch (IOException e) {
+ throw new RuntimeException("Error while scanning the file", e);
+ }
+ }
+
+ @Override
+ public void initQueryDataSource(IQueryDataSource dataSource) {
+ this.queryDataSource = (QueryDataSource) dataSource;
+ this.seriesScanUtil.initQueryDataSource(queryDataSource);
+ this.resultTsBlockBuilder = new TsBlockBuilder(getResultDataTypes());
+ this.resultTsBlockBuilder.setMaxTsBlockLineNumber(this.maxTsBlockLineNum);
+ this.measurementDataBuilder = new TsBlockBuilder(this.measurementColumnTSDataTypes);
+ this.measurementDataBuilder.setMaxTsBlockLineNumber(this.maxTsBlockLineNum);
+ }
+
+ @Override
+ protected void updateResultTsBlock() {
+ appendAggregationResult(
+ resultTsBlockBuilder, aggregators, timeRangeIterator.currentOutputTime());
+ }
}
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 743bdc494ac..2f0152f160a 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
@@ -24,6 +24,8 @@ import org.apache.iotdb.commons.path.AlignedFullPath;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
+import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITimeRangeIterator;
+import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.SingleTimeWindowIterator;
import org.apache.iotdb.db.queryengine.execution.driver.DataDriverContext;
import org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeManager;
import org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeService;
@@ -1015,6 +1017,170 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
context.getNextOperatorId(),
node.getPlanNodeId(),
TableAggregationTableScanOperator.class.getSimpleName());
- throw new UnsupportedOperationException("Agg-BE not supported");
+
+ List<Aggregator> aggregators = new ArrayList<>();
+
+ // TODO fix childLayout
+ Map<Symbol, Integer> childLayout = new HashMap<>();
+
+ for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : node.getAggregations().entrySet()) {
+ Aggregator aggregator =
+ buildAggregator(childLayout, entry.getValue(), node.getStep(), context.getTypeProvider());
+ aggregators.add(aggregator);
+ }
+
+ List<Symbol> outputColumnNames = node.getOutputSymbols();
+ int outputColumnCount = outputColumnNames.size();
+ List<ColumnSchema> columnSchemas = new ArrayList<>(outputColumnCount);
+ int[] columnsIndexArray = new int[outputColumnCount];
+ Map<Symbol, ColumnSchema> columnSchemaMap = node.getAssignments();
+ Map<Symbol, Integer> idAndAttributeColumnsIndexMap = node.getIdAndAttributeIndexMap();
+ List<String> measurementColumnNames = new ArrayList<>();
+ List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
+ int measurementColumnCount = 0;
+ int idx = 0;
+ for (Symbol columnName : outputColumnNames) {
+ ColumnSchema schema =
+ requireNonNull(columnSchemaMap.get(columnName), columnName + " is null");
+
+ switch (schema.getColumnCategory()) {
+ case ID:
+ case ATTRIBUTE:
+ columnsIndexArray[idx++] =
+ requireNonNull(
+ idAndAttributeColumnsIndexMap.get(columnName), columnName + " is null");
+ columnSchemas.add(schema);
+ break;
+ case MEASUREMENT:
+ columnsIndexArray[idx++] = measurementColumnCount;
+ measurementColumnCount++;
+ measurementColumnNames.add(columnName.getName());
+ measurementSchemas.add(
+ new MeasurementSchema(schema.getName(), getTSDataType(schema.getType())));
+ columnSchemas.add(schema);
+ break;
+ case TIME:
+ columnsIndexArray[idx++] = -1;
+ columnSchemas.add(schema);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unexpected column category: " + schema.getColumnCategory());
+ }
+ }
+
+ Set<Symbol> outputSet = new HashSet<>(outputColumnNames);
+ for (Map.Entry<Symbol, ColumnSchema> entry : node.getAssignments().entrySet()) {
+ if (!outputSet.contains(entry.getKey())
+ && entry.getValue().getColumnCategory() == MEASUREMENT) {
+ measurementColumnCount++;
+ measurementColumnNames.add(entry.getKey().getName());
+ measurementSchemas.add(
+ new MeasurementSchema(
+ entry.getValue().getName(), getTSDataType(entry.getValue().getType())));
+ }
+ }
+
+ SeriesScanOptions.Builder scanOptionsBuilder =
+ node.getTimePredicate()
+ .map(timePredicate -> getSeriesScanOptionsBuilder(context, timePredicate))
+ .orElse(new SeriesScanOptions.Builder());
+ scanOptionsBuilder.withPushDownLimit(node.getPushDownLimit());
+ scanOptionsBuilder.withPushDownOffset(node.getPushDownOffset());
+ scanOptionsBuilder.withPushLimitToEachDevice(node.isPushLimitToEachDevice());
+ scanOptionsBuilder.withAllSensors(new HashSet<>(measurementColumnNames));
+
+ Expression pushDownPredicate = node.getPushDownPredicate();
+ if (pushDownPredicate != null) {
+ scanOptionsBuilder.withPushDownFilter(
+ convertPredicateToFilter(pushDownPredicate, measurementColumnNames, columnSchemaMap));
+ }
+
+ ITimeRangeIterator timeRangeIterator =
+ new SingleTimeWindowIterator(Long.MIN_VALUE, Long.MAX_VALUE);
+
+ return new TableAggregationTableScanOperator(
+ node.getPlanNodeId(),
+ operatorContext,
+ columnSchemas,
+ columnsIndexArray,
+ measurementColumnCount,
+ node.getDeviceEntries(),
+ node.getScanOrder(),
+ scanOptionsBuilder.build(),
+ measurementColumnNames,
+ measurementSchemas,
+ TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber(),
+ // TODO if it equals subSensor variable
+ measurementColumnCount,
+ aggregators,
+ timeRangeIterator,
+ false,
+ null,
+ calculateMaxAggregationResultSize(),
+ true);
+
+ // throw new UnsupportedOperationException("Agg-BE not supported");
}
+
+ public static long calculateMaxAggregationResultSize(
+ // List<? extends AggregationDescriptor> aggregationDescriptors,
+ // ITimeRangeIterator timeRangeIterator
+ ) {
+ // TODO perfect max aggregation result size logic
+ return TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes();
+
+ // long timeValueColumnsSizePerLine = TimeColumn.SIZE_IN_BYTES_PER_POSITION;
+ // for (AggregationDescriptor descriptor : aggregationDescriptors) {
+ // List<TSDataType> outPutDataTypes =
+ // descriptor.getOutputColumnNames().stream()
+ // .map(typeProvider::getTableModelType)
+ // .collect(Collectors.toList());
+ // for (TSDataType tsDataType : outPutDataTypes) {
+ // timeValueColumnsSizePerLine += getOutputColumnSizePerLine(tsDataType);
+ // }
+ // }
+ //
+ // return Math.min(
+ // TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes(),
+ // Math.min(
+ // TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber(),
+ // timeRangeIterator.getTotalIntervalNum())
+ // * timeValueColumnsSizePerLine);
+ }
+
+ // private void aaa(AggregationTableScanNode node) {
+ // Map<Symbol, Integer> childLayout = new HashMap<>();
+ //
+ // for (Symbol columnName : node.getoutputColumnNames) {
+ // ColumnSchema schema =
+ // requireNonNull(columnSchemaMap.get(columnName), columnName + " is null");
+ //
+ // switch (schema.getColumnCategory()) {
+ // case ID:
+ // case ATTRIBUTE:
+ // columnsIndexArray[idx++] =
+ // requireNonNull(
+ // idAndAttributeColumnsIndexMap.get(columnName), columnName + " is
+ // null");
+ // columnSchemas.add(schema);
+ // break;
+ // case MEASUREMENT:
+ // columnsIndexArray[idx++] = measurementColumnCount;
+ // measurementColumnCount++;
+ // measurementColumnNames.add(columnName.getName());
+ // measurementSchemas.add(
+ // new MeasurementSchema(schema.getName(), getTSDataType(schema.getType())));
+ // columnSchemas.add(schema);
+ // break;
+ // case TIME:
+ // columnsIndexArray[idx++] = -1;
+ // columnSchemas.add(schema);
+ // break;
+ // default:
+ // throw new IllegalArgumentException(
+ // "Unexpected column category: " + schema.getColumnCategory());
+ // }
+ // }
+ // }
}