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 <[email protected]> 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()); + // } + // } + // } }
