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/12/18 11:43:55 UTC
(iotdb) branch beyyes/last_cache updated (ead283f90bf -> e968a75889e)
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a change to branch beyyes/last_cache
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from ead283f90bf add basic impl code
add aa8e21c664f Fix ThreadPool Metric NPE #14459
add 9df0030ebca Add TsfileSize column to 'show regions' (#14436)
add 6efe1125c8f Fix query scan will return duplicated timestamp or unordered timestamp while TsFileResource degrading
add b5fe66ee754 Add eviction metrics in binary allocator (#14432)
add 7d6753a81c9 Bump ratis to 3.1.2-a862328-SNAPSHOT
add addbf964e4a Pipe: Do not stop runtime agent when removing datanode (#14474)
add ef6a06a2def Subscription: allow generate subsequent events with the same tablet batch to avoid large message & improve poll logic to avoid unnecessary nack (#14452)
add 59dac3272a6 [TABLE MODEL] Support user-defined aggregate function
add 876d8469515 fix checkNonAlignedDeviceSeries (#14461)
add 91b672da649 Allow hot reloading compaction from disabled status (#14470)
new de8d028848c Merge branch 'master' into beyyes/last_cache
new e968a75889e fix
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/iotdb/udf/AggregateFunctionExample.java | 80 +--
.../query/udf/example/relational/FirstTwoSum.java | 168 +++++
.../db/query/udf/example/relational/MyAvg.java | 126 ++++
.../{UDAFCount.java => relational/MyCount.java} | 56 +-
.../it/env/cluster/config/MppCommonConfig.java | 6 +
.../env/cluster/config/MppSharedCommonConfig.java | 7 +
.../it/env/remote/config/RemoteCommonConfig.java | 5 +
.../org/apache/iotdb/itbase/env/CommonConfig.java | 2 +
...thMisMatchIT.java => IoTDBFileTimeIndexIT.java} | 79 +--
.../it/db/it/udf/IoTDBSQLFunctionManagementIT.java | 58 +-
.../udf/IoTDBUserDefinedAggregateFunctionIT.java | 690 +++++++++++++++++++++
....java => IoTDBUserDefinedScalarFunctionIT.java} | 8 +-
.../relational/it/db/it/udf/SQLFunctionUtils.java | 87 +++
...ionConfig.java => AggregateFunctionConfig.java} | 21 +-
.../udf/api/relational/AggregateFunction.java | 81 ++-
.../iotdb/udf/api/relational/access/Record.java | 3 +-
.../consumer/SubscriptionConsumer.java | 3 +-
.../consumer/SubscriptionPullConsumer.java | 3 +-
.../consumer/SubscriptionPushConsumer.java | 3 +-
.../heartbeat/DataNodeHeartbeatHandler.java | 3 +
.../confignode/manager/load/cache/LoadCache.java | 11 +
.../manager/partition/PartitionManager.java | 8 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 23 +-
.../subtask/connector/PipeConnectorSubtask.java | 2 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 13 +-
.../execution/operator/source/SeriesScanUtil.java | 30 +-
.../TableAggregationTableScanOperator.java | 135 ++--
.../source/relational/TableLastQueryOperator.java | 137 +++-
.../relational/aggregation/AccumulatorFactory.java | 38 +-
.../UserDefinedAggregateFunctionAccumulator.java | 123 ++++
.../GroupedUserDefinedAggregateAccumulator.java | 134 ++++
.../config/executor/ClusterConfigTaskExecutor.java | 8 +-
.../execution/config/metadata/ShowRegionTask.java | 18 +-
.../plan/planner/TableOperatorGenerator.java | 134 ++--
.../relational/analyzer/ExpressionTreeUtils.java | 5 +-
.../relational/metadata/TableMetadataImpl.java | 26 +-
.../relational/planner/optimizations/Util.java | 2 -
.../iotdb/db/service/metrics/FileMetrics.java | 5 +
.../db/service/metrics/file/TsFileMetrics.java | 18 +
.../compaction/repair/RepairDataFileScanUtil.java | 2 +-
.../compaction/schedule/CompactionTaskManager.java | 11 +-
.../dataregion/read/QueryDataSource.java | 14 +-
.../dataregion/tsfile/TsFileResource.java | 12 +-
.../db/subscription/broker/SubscriptionBroker.java | 104 +++-
.../broker/SubscriptionPrefetchingQueue.java | 7 +-
.../db/subscription/broker/SubscriptionStates.java | 30 +-
.../event/SubscriptionCommitContextSupplier.java} | 9 +-
.../db/subscription/event/SubscriptionEvent.java | 14 +-
.../batch/SubscriptionPipeTabletEventBatch.java | 57 +-
.../SubscriptionEventExtendableResponse.java | 6 -
.../event/response/SubscriptionEventResponse.java | 7 +
.../response/SubscriptionEventTabletResponse.java | 60 +-
.../receiver/SubscriptionReceiverV1.java | 2 +-
.../commons/binaryallocator/BinaryAllocator.java | 18 +-
.../iotdb/commons/binaryallocator/arena/Arena.java | 39 +-
.../metric/BinaryAllocatorMetrics.java | 36 +-
.../commons/concurrent/ThreadPoolMetrics.java | 26 +-
.../schema/column/ColumnHeaderConstant.java | 4 +-
.../iotdb/commons/udf/access/RecordIterator.java | 10 +-
.../TableBuiltinAggregationFunction.java | 4 +-
.../iotdb/commons/udf/utils/TableUDFUtils.java | 6 +-
.../commons/udf/utils/UDFDataTypeTransformer.java | 45 +-
.../src/main/thrift/confignode.thrift | 1 +
pom.xml | 4 +-
64 files changed, 2395 insertions(+), 492 deletions(-)
copy integration-test/src/main/java/org/apache/iotdb/db/query/udf/example/UDAFCount.java => example/udf/src/main/java/org/apache/iotdb/udf/AggregateFunctionExample.java (52%)
create mode 100644 integration-test/src/main/java/org/apache/iotdb/db/query/udf/example/relational/FirstTwoSum.java
create mode 100644 integration-test/src/main/java/org/apache/iotdb/db/query/udf/example/relational/MyAvg.java
copy integration-test/src/main/java/org/apache/iotdb/db/query/udf/example/{UDAFCount.java => relational/MyCount.java} (55%)
copy integration-test/src/test/java/org/apache/iotdb/db/it/{aligned/IoTDBAlignedSeriesQueryWithMisMatchIT.java => IoTDBFileTimeIndexIT.java} (58%)
create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/udf/IoTDBUserDefinedAggregateFunctionIT.java
rename integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/udf/{scalar/IoTDBScalarFunctionIT.java => IoTDBUserDefinedScalarFunctionIT.java} (97%)
create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/udf/SQLFunctionUtils.java
copy iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/config/{ScalarFunctionConfig.java => AggregateFunctionConfig.java} (66%)
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/UserDefinedAggregateFunctionAccumulator.java
create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedUserDefinedAggregateAccumulator.java
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{pipe/extractor/dataregion/realtime/epoch/TsFileEpochStateMigrator.java => subscription/event/SubscriptionCommitContextSupplier.java} (79%)
(iotdb) 01/02: Merge branch 'master' into beyyes/last_cache
Posted by ca...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/last_cache
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit de8d028848cbbf00d3b2ec607a35ced1fec1b7b3
Merge: ead283f90bf 91b672da649
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Dec 18 14:37:07 2024 +0800
Merge branch 'master' into beyyes/last_cache
.../apache/iotdb/udf/AggregateFunctionExample.java | 117 ++++
.../query/udf/example/relational/FirstTwoSum.java | 168 +++++
.../db/query/udf/example/relational/MyAvg.java | 126 ++++
.../db/query/udf/example/relational/MyCount.java | 95 +++
.../it/env/cluster/config/MppCommonConfig.java | 6 +
.../env/cluster/config/MppSharedCommonConfig.java | 7 +
.../it/env/remote/config/RemoteCommonConfig.java | 5 +
.../org/apache/iotdb/itbase/env/CommonConfig.java | 2 +
.../apache/iotdb/db/it/IoTDBFileTimeIndexIT.java | 117 ++++
.../it/db/it/udf/IoTDBSQLFunctionManagementIT.java | 58 +-
.../udf/IoTDBUserDefinedAggregateFunctionIT.java | 690 +++++++++++++++++++++
....java => IoTDBUserDefinedScalarFunctionIT.java} | 8 +-
.../relational/it/db/it/udf/SQLFunctionUtils.java | 87 +++
.../customizer/config/AggregateFunctionConfig.java | 53 ++
.../udf/api/relational/AggregateFunction.java | 81 ++-
.../iotdb/udf/api/relational/access/Record.java | 3 +-
.../consumer/SubscriptionConsumer.java | 3 +-
.../consumer/SubscriptionPullConsumer.java | 3 +-
.../consumer/SubscriptionPushConsumer.java | 3 +-
.../heartbeat/DataNodeHeartbeatHandler.java | 3 +
.../confignode/manager/load/cache/LoadCache.java | 11 +
.../manager/partition/PartitionManager.java | 8 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 23 +-
.../subtask/connector/PipeConnectorSubtask.java | 2 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 13 +-
.../execution/operator/source/SeriesScanUtil.java | 30 +-
.../relational/aggregation/AccumulatorFactory.java | 38 +-
.../UserDefinedAggregateFunctionAccumulator.java | 123 ++++
.../GroupedUserDefinedAggregateAccumulator.java | 134 ++++
.../config/executor/ClusterConfigTaskExecutor.java | 8 +-
.../execution/config/metadata/ShowRegionTask.java | 18 +-
.../relational/analyzer/ExpressionTreeUtils.java | 5 +-
.../relational/metadata/TableMetadataImpl.java | 26 +-
.../relational/planner/optimizations/Util.java | 2 -
.../iotdb/db/service/metrics/FileMetrics.java | 5 +
.../db/service/metrics/file/TsFileMetrics.java | 18 +
.../compaction/repair/RepairDataFileScanUtil.java | 2 +-
.../compaction/schedule/CompactionTaskManager.java | 11 +-
.../dataregion/read/QueryDataSource.java | 14 +-
.../dataregion/tsfile/TsFileResource.java | 12 +-
.../db/subscription/broker/SubscriptionBroker.java | 104 +++-
.../broker/SubscriptionPrefetchingQueue.java | 7 +-
.../db/subscription/broker/SubscriptionStates.java | 30 +-
.../event/SubscriptionCommitContextSupplier.java | 10 +-
.../db/subscription/event/SubscriptionEvent.java | 14 +-
.../batch/SubscriptionPipeTabletEventBatch.java | 57 +-
.../SubscriptionEventExtendableResponse.java | 6 -
.../event/response/SubscriptionEventResponse.java | 7 +
.../response/SubscriptionEventTabletResponse.java | 60 +-
.../receiver/SubscriptionReceiverV1.java | 2 +-
.../commons/binaryallocator/BinaryAllocator.java | 18 +-
.../iotdb/commons/binaryallocator/arena/Arena.java | 39 +-
.../metric/BinaryAllocatorMetrics.java | 36 +-
.../commons/concurrent/ThreadPoolMetrics.java | 26 +-
.../schema/column/ColumnHeaderConstant.java | 4 +-
.../iotdb/commons/udf/access/RecordIterator.java | 10 +-
.../TableBuiltinAggregationFunction.java | 4 +-
.../iotdb/commons/udf/utils/TableUDFUtils.java | 6 +-
.../commons/udf/utils/UDFDataTypeTransformer.java | 45 +-
.../src/main/thrift/confignode.thrift | 1 +
pom.xml | 4 +-
61 files changed, 2424 insertions(+), 204 deletions(-)
(iotdb) 02/02: fix
Posted by ca...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/last_cache
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e968a75889e9ecb5bc5ca78878f56eda4f1ffdcf
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Dec 18 19:43:36 2024 +0800
fix
---
.../TableAggregationTableScanOperator.java | 135 ++++++++++----------
.../source/relational/TableLastQueryOperator.java | 137 +++++++++++++++++----
.../plan/planner/TableOperatorGenerator.java | 134 ++++++++------------
3 files changed, 234 insertions(+), 172 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 8eb700770d5..2dd7f25a7c2 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
@@ -24,7 +24,7 @@ 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.ITableTimeRangeIterator;
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.AbstractDataSourceOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanUtil;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
import org.apache.iotdb.db.queryengine.execution.operator.window.IWindow;
@@ -72,34 +72,40 @@ import static org.apache.iotdb.db.queryengine.execution.operator.source.relation
import static org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanGraphPrinter.DEVICE_NUMBER;
import static org.apache.tsfile.read.common.block.TsBlockUtil.skipPointsOutOfTimeRange;
-public class TableAggregationTableScanOperator extends AbstractSeriesAggregationScanOperator {
+public class TableAggregationTableScanOperator extends AbstractDataSourceOperator {
private static final long INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(TableAggregationTableScanOperator.class);
- private final List<TableAggregator> tableAggregators;
+ private boolean finished = false;
+ private TsBlock inputTsBlock;
- private final List<ColumnSchema> groupingKeySchemas;
- private final int[] groupingKeyIndex;
+ protected List<TableAggregator> tableAggregators;
+ protected final List<ColumnSchema> groupingKeySchemas;
+ protected final int[] groupingKeyIndex;
- private final List<ColumnSchema> columnSchemas;
- private final int[] columnsIndexArray;
-
- private final List<DeviceEntry> deviceEntries;
- private final int deviceCount;
+ protected final List<DeviceEntry> deviceEntries;
+ protected final int deviceCount;
private int currentDeviceIndex;
+ private List<String> measurementColumnNames;
+ private Set<String> allSensors;
+ private List<IMeasurementSchema> measurementSchemas;
+ private List<TSDataType> measurementColumnTSDataTypes;
+ private int measurementCount;
- private final Ordering scanOrder;
- private final SeriesScanOptions seriesScanOptions;
+ private List<ColumnSchema> aggColumnSchemas;
+ private int[] aggColumnsIndexArray;
- private final List<String> measurementColumnNames;
- private final Set<String> allSensors;
- private final List<IMeasurementSchema> measurementSchemas;
- private final List<TSDataType> measurementColumnTSDataTypes;
+ private SeriesScanOptions seriesScanOptions;
+ private final boolean ascending;
+ private final Ordering scanOrder;
+ // Some special data types(like BLOB) cannot use statistics
+ protected final boolean canUseStatistics;
+ private final long cachedRawDataSize;
// stores all inputChannels of tableAggregators,
// e.g. for aggregation `last(s1), count(s2), count(s1)`, the inputChannels should be [0, 1, 0]
- private final List<Integer> aggregatorInputChannels;
+ private List<Integer> aggregatorInputChannels;
private QueryDataSource queryDataSource;
@@ -110,53 +116,42 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
public TableAggregationTableScanOperator(
PlanNodeId sourceId,
OperatorContext context,
- List<ColumnSchema> columnSchemas,
- int[] columnsIndexArray,
+ List<ColumnSchema> aggColumnSchemas,
+ int[] aggColumnsIndexArray,
List<DeviceEntry> deviceEntries,
- Ordering scanOrder,
SeriesScanOptions seriesScanOptions,
List<String> measurementColumnNames,
Set<String> allSensors,
List<IMeasurementSchema> measurementSchemas,
- int measurementCount,
List<TableAggregator> tableAggregators,
List<ColumnSchema> groupingKeySchemas,
int[] groupingKeyIndex,
ITableTimeRangeIterator tableTimeRangeIterator,
boolean ascending,
- long maxReturnSize,
boolean canUseStatistics,
List<Integer> aggregatorInputChannels) {
- super(
- sourceId,
- context,
- null,
- measurementCount,
- null,
- null,
- ascending,
- false,
- null,
- maxReturnSize,
- (1L + measurementCount) * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte(),
- canUseStatistics);
-
+ this.sourceId = sourceId;
+ this.operatorContext = context;
+ this.canUseStatistics = canUseStatistics;
this.tableAggregators = tableAggregators;
this.groupingKeySchemas = groupingKeySchemas;
this.groupingKeyIndex = groupingKeyIndex;
-
- this.sourceId = sourceId;
- this.operatorContext = context;
- this.columnSchemas = columnSchemas;
- this.columnsIndexArray = columnsIndexArray;
+ this.aggColumnSchemas = aggColumnSchemas;
+ this.aggColumnsIndexArray = aggColumnsIndexArray;
this.deviceEntries = deviceEntries;
this.deviceCount = deviceEntries.size();
this.operatorContext.recordSpecifiedInfo(DEVICE_NUMBER, Integer.toString(this.deviceCount));
- this.scanOrder = scanOrder;
+ this.ascending = ascending;
+ this.scanOrder = ascending ? Ordering.ASC : Ordering.DESC;
this.seriesScanOptions = seriesScanOptions;
this.measurementColumnNames = measurementColumnNames;
+ this.measurementCount = measurementColumnNames.size();
+ this.cachedRawDataSize =
+ (1L + this.measurementCount)
+ * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
this.allSensors = allSensors;
+ // this.allSensors = new HashSet<>(measurementColumnNames);
this.measurementSchemas = measurementSchemas;
this.measurementColumnTSDataTypes =
measurementSchemas.stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
@@ -164,12 +159,6 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
this.operatorContext.recordSpecifiedInfo(CURRENT_DEVICE_INDEX_STRING, Integer.toString(0));
this.aggregatorInputChannels = aggregatorInputChannels;
this.timeIterator = tableTimeRangeIterator;
- if (tableTimeRangeIterator.getType()
- == ITableTimeRangeIterator.TimeIteratorType.SINGLE_TIME_ITERATOR) {
- curTimeRange = new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE);
- }
-
- this.maxReturnSize = maxReturnSize;
constructAlignedSeriesScanUtil();
}
@@ -182,6 +171,23 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
return finished;
}
+ @Override
+ public long calculateMaxPeekMemory() {
+ return cachedRawDataSize + maxReturnSize;
+ }
+
+ @Override
+ public long calculateMaxReturnSize() {
+ return maxReturnSize;
+ }
+
+ @Override
+ public long calculateRetainedSizeAfterCallingNext() {
+ return timeIterator.getType() == ITableTimeRangeIterator.TimeIteratorType.DATE_BIN_TIME_ITERATOR
+ ? cachedRawDataSize
+ : 0;
+ }
+
@Override
public boolean hasNext() throws Exception {
return timeIterator.hasCachedTimeRange()
@@ -216,7 +222,6 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
// return true if current time window is calc finished
if (calculateAggregationResultForCurrentTimeRange()) {
timeIterator.resetCurTimeRange();
- // curTimeRange = null;
}
}
@@ -275,7 +280,6 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
}
/** Return true if we have the result of this timeRange. */
- @Override
protected boolean calculateAggregationResultForCurrentTimeRange() {
try {
if (calcFromCachedData()) {
@@ -338,14 +342,12 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
}
}
- @Override
protected void updateResultTsBlock() {
appendAggregationResult(resultTsBlockBuilder, tableAggregators);
// after appendAggregationResult invoked, aggregators must be cleared
resetTableAggregators();
}
- @Override
protected boolean calcFromCachedData() {
return calcUsingRawData(inputTsBlock);
}
@@ -411,7 +413,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
continue;
}
valueColumns[idx] =
- buildValueColumn(columnSchemas.get(idx).getColumnCategory(), inputRegion, idx);
+ buildValueColumn(aggColumnSchemas.get(idx).getColumnCategory(), inputRegion, idx);
}
TsBlock tsBlock =
@@ -448,7 +450,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
(String)
deviceEntries
.get(currentDeviceIndex)
- .getNthSegment(columnsIndexArray[columnIdx] + 1);
+ .getNthSegment(aggColumnsIndexArray[columnIdx] + 1);
return getIdOrAttrColumn(
inputRegion.getTimeColumn().getPositionCount(),
id == null ? null : new Binary(id, TSFileConfig.STRING_CHARSET));
@@ -457,10 +459,10 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
deviceEntries
.get(currentDeviceIndex)
.getAttributeColumnValues()
- .get(columnsIndexArray[columnIdx]);
+ .get(aggColumnsIndexArray[columnIdx]);
return getIdOrAttrColumn(inputRegion.getTimeColumn().getPositionCount(), attr);
case MEASUREMENT:
- return inputRegion.getColumn(columnsIndexArray[columnIdx]);
+ return inputRegion.getColumn(aggColumnsIndexArray[columnIdx]);
default:
throw new IllegalStateException("Unsupported column type: " + columnSchemaCategory);
}
@@ -492,7 +494,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
idx++;
TsTableColumnCategory columnSchemaCategory =
- columnSchemas.get(aggregatorInputChannels.get(idx)).getColumnCategory();
+ aggColumnSchemas.get(aggregatorInputChannels.get(idx)).getColumnCategory();
statisticsArray[i] =
buildStatistics(
columnSchemaCategory,
@@ -519,7 +521,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
(String)
deviceEntries
.get(currentDeviceIndex)
- .getNthSegment(columnsIndexArray[columnIdx] + 1);
+ .getNthSegment(aggColumnsIndexArray[columnIdx] + 1);
return getStatistics(
timeStatistics, id == null ? null : new Binary(id, TSFileConfig.STRING_CHARSET));
case ATTRIBUTE:
@@ -527,10 +529,10 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
deviceEntries
.get(currentDeviceIndex)
.getAttributeColumnValues()
- .get(columnsIndexArray[columnIdx]);
+ .get(aggColumnsIndexArray[columnIdx]);
return getStatistics(timeStatistics, attr);
case MEASUREMENT:
- return valueStatistics[columnsIndexArray[columnIdx]];
+ return valueStatistics[aggColumnsIndexArray[columnIdx]];
default:
throw new IllegalStateException("Unsupported column type: " + columnSchemaCategory);
}
@@ -550,7 +552,6 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
}
@SuppressWarnings({"squid:S3776", "squid:S135", "squid:S3740"})
- @Override
public boolean readAndCalcFromFile() throws IOException {
// start stopwatch
long start = System.nanoTime();
@@ -573,8 +574,8 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
if (timeIterator
.getCurTimeRange()
.contains(fileTimeStatistics.getStartTime(), fileTimeStatistics.getEndTime())) {
- Statistics[] statisticsList = new Statistics[subSensorSize];
- for (int i = 0; i < subSensorSize; i++) {
+ Statistics[] statisticsList = new Statistics[measurementCount];
+ for (int i = 0; i < measurementCount; i++) {
statisticsList[i] = seriesScanUtil.currentFileStatistics(i);
}
calcFromStatistics(fileTimeStatistics, statisticsList);
@@ -620,8 +621,8 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
.getCurTimeRange()
.contains(chunkTimeStatistics.getStartTime(), chunkTimeStatistics.getEndTime())) {
// calc from chunkMetaData
- Statistics[] statisticsList = new Statistics[subSensorSize];
- for (int i = 0; i < subSensorSize; i++) {
+ Statistics[] statisticsList = new Statistics[measurementCount];
+ for (int i = 0; i < measurementCount; i++) {
statisticsList[i] = seriesScanUtil.currentChunkStatistics(i);
}
calcFromStatistics(chunkTimeStatistics, statisticsList);
@@ -669,8 +670,8 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
if (timeIterator
.getCurTimeRange()
.contains(pageTimeStatistics.getStartTime(), pageTimeStatistics.getEndTime())) {
- Statistics[] statisticsList = new Statistics[subSensorSize];
- for (int i = 0; i < subSensorSize; i++) {
+ Statistics[] statisticsList = new Statistics[measurementCount];
+ for (int i = 0; i < measurementCount; i++) {
statisticsList[i] = seriesScanUtil.currentPageStatistics(i);
}
calcFromStatistics(pageTimeStatistics, statisticsList);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableLastQueryOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableLastQueryOperator.java
index 481e575340c..e6a369b7319 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableLastQueryOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableLastQueryOperator.java
@@ -19,8 +19,11 @@
package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
-import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractDataSourceOperator;
+import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITableTimeRangeIterator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+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.relational.metadata.QualifiedObjectName;
@@ -40,15 +43,18 @@ import org.apache.tsfile.utils.TsPrimitiveType;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.OptionalLong;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.TIME_COLUMN_TEMPLATE;
import static org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache.EMPTY_TIME_VALUE_PAIR;
-public class TableLastQueryOperator extends AbstractDataSourceOperator {
+public class TableLastQueryOperator extends TableAggregationTableScanOperator {
private static final long INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(TableLastQueryOperator.class);
@@ -58,16 +64,16 @@ public class TableLastQueryOperator extends AbstractDataSourceOperator {
private boolean finished = false;
// TODO not need all table aggregators when match last cache
- private final List<TableAggregator> tableAggregators;
+ // private final List<TableAggregator> tableAggregators;
private List<TableAggregator> unCachedTableAggregators;
- private final List<ColumnSchema> groupingKeySchemas;
+ // private final List<ColumnSchema> groupingKeySchemas;
private final QualifiedObjectName qualifiedObjectName;
- private final List<DeviceEntry> deviceEntries;
+ // private final List<DeviceEntry> deviceEntries;
private int currentDeviceIndex;
- private final List<String> measurementColumnNames;
- private final List<IMeasurementSchema> measurementSchemas;
- private final List<TSDataType> measurementColumnTSDataTypes;
+ // private final List<String> measurementColumnNames;
+ // private final List<IMeasurementSchema> measurementSchemas;
+ // private final List<TSDataType> measurementColumnTSDataTypes;
private QueryDataSource queryDataSource;
@@ -80,25 +86,79 @@ public class TableLastQueryOperator extends AbstractDataSourceOperator {
private boolean hashLastBy;
private boolean needCacheTimeColumn;
private boolean calcCacheForCurrentDevice;
- private List<String> currentUnCacheMeasurements = new ArrayList<>();
+ private Set<String> currentUnCacheMeasurements = new HashSet<>();
+
+ private final List<TableAggregator> originalTableAggregators;
+ private final List<String> originalMeasurementColumnNames;
+ private final Set<String> originalAllSensors;
+ private final List<IMeasurementSchema> originalMeasurementSchemas;
+ // private final List<TSDataType> originalMeasurementColumnTSDataTypes;
+ private final List<ColumnSchema> originalColumnSchemas;
+ private final int[] originalColumnsIndexArray;
+ private final SeriesScanOptions originalSeriesScanOptions;
+ // stores all inputChannels of tableAggregators,
+ // e.g. for aggregation `last(s1), count(s2), count(s1)`, the inputChannels should be [0, 1, 0]
+ private final List<Integer> originalAggregatorInputChannels;
public TableLastQueryOperator(
- List<TableAggregator> tableAggregators,
- List<ColumnSchema> groupingKeySchemas,
- QualifiedObjectName qualifiedObjectName,
+ PlanNodeId sourceId,
+ OperatorContext context,
+ List<ColumnSchema> columnSchemas,
+ int[] columnsIndexArray,
List<DeviceEntry> deviceEntries,
+ SeriesScanOptions seriesScanOptions,
List<String> measurementColumnNames,
+ Set<String> allSensors,
List<IMeasurementSchema> measurementSchemas,
- List<TSDataType> measurementColumnTSDataTypes) {
- this.tableAggregators = tableAggregators;
- this.groupingKeySchemas = groupingKeySchemas;
- this.qualifiedObjectName = qualifiedObjectName;
- this.deviceEntries = deviceEntries;
- this.measurementColumnNames = measurementColumnNames;
- this.measurementSchemas = measurementSchemas;
- this.measurementColumnTSDataTypes = measurementColumnTSDataTypes;
- this.isLastBy = new boolean[tableAggregators.size()];
+ int measurementCount,
+ List<TableAggregator> tableAggregators,
+ List<ColumnSchema> groupingKeySchemas,
+ int[] groupingKeyIndex,
+ ITableTimeRangeIterator tableTimeRangeIterator,
+ boolean ascending,
+ boolean canUseStatistics,
+ List<Integer> aggregatorInputChannels,
+ QualifiedObjectName qualifiedObjectName) {
+
+ super(
+ sourceId,
+ context,
+ null,
+ null,
+ deviceEntries,
+ null,
+ Collections.emptyList(),
+ null,
+ Collections.emptyList(),
+ null,
+ groupingKeySchemas,
+ groupingKeyIndex,
+ tableTimeRangeIterator,
+ ascending,
+ canUseStatistics,
+ null);
+
+ this.originalTableAggregators = tableAggregators;
+ this.originalAggregatorInputChannels = aggregatorInputChannels;
+
+ this.originalColumnSchemas = columnSchemas;
+ this.originalColumnsIndexArray = columnsIndexArray;
+
+ this.originalMeasurementColumnNames = measurementColumnNames;
+ this.originalMeasurementSchemas = measurementSchemas;
+
+ this.originalSeriesScanOptions = seriesScanOptions;
+ this.originalAllSensors = allSensors;
+
+ // this.tableAggregators = tableAggregators;
+ // this.groupingKeySchemas = groupingKeySchemas;
+ // this.deviceEntries = deviceEntries;
+ // this.measurementColumnNames = measurementColumnNames;
+ // this.measurementSchemas = measurementSchemas;
+ // this.measurementColumnTSDataTypes = measurementColumnTSDataTypes;
+ this.isLastBy = new boolean[tableAggregators.size()];
+ this.qualifiedObjectName = qualifiedObjectName;
this.lastColumns = new ArrayList<>();
this.lastByColumns = new ArrayList<>();
this.indexOfLastColumnInAggregators = new int[lastColumns.size()];
@@ -154,6 +214,22 @@ public class TableLastQueryOperator extends AbstractDataSourceOperator {
DeviceEntry currentDeviceEntry = deviceEntries.get(currentDeviceIndex);
if (!calcCacheForCurrentDevice) {
+
+ for (int i = 0; i < originalTableAggregators.size(); i++) {
+ TableAggregator tableAggregator = originalTableAggregators.get(i);
+ if (isLastBy[i]) {
+ Optional<Pair<OptionalLong, TsPrimitiveType[]>> lastByResult =
+ TABLE_DEVICE_SCHEMA_CACHE.getLastRow(
+ qualifiedObjectName.getDatabaseName(),
+ currentDeviceEntry.getDeviceID(),
+ "",
+ lastByColumns);
+ } else {
+
+ }
+ }
+
+ // process last(time), last_by(x,time)...
if (!lastByColumns.isEmpty()) {
Optional<Pair<OptionalLong, TsPrimitiveType[]>> lastByResult =
TABLE_DEVICE_SCHEMA_CACHE.getLastRow(
@@ -161,15 +237,24 @@ public class TableLastQueryOperator extends AbstractDataSourceOperator {
currentDeviceEntry.getDeviceID(),
"",
lastByColumns);
+
if (!lastByResult.isPresent()) {
// all missed
-
+ currentUnCacheMeasurements.addAll(lastByColumns);
} else {
+ TsPrimitiveType[] lastByResultPair = lastByResult.get().getRight();
+ for (int i = 0; i < lastByColumns.size(); i++) {
+ if (lastByResultPair[i] == null) {
+ currentUnCacheMeasurements.add(lastByColumns.get(i));
+ } else {
+ }
+ }
}
// TODO verify id and attr columns
}
+ // process last(x), last(y), ...
if (!lastColumns.isEmpty()) {
for (int i = 0; i < lastColumns.size(); i++) {
String measurement = lastByColumns.get(i);
@@ -178,11 +263,14 @@ public class TableLastQueryOperator extends AbstractDataSourceOperator {
qualifiedObjectName.getDatabaseName(),
currentDeviceEntry.getDeviceID(),
measurement);
+
if (timeValuePair == null) {
currentUnCacheMeasurements.add(measurement);
} else {
+
ColumnBuilder columnBuilder =
resultTsBlockBuilder.getColumnBuilder(indexOfLastColumnInAggregators[i]);
+
if (timeValuePair == EMPTY_TIME_VALUE_PAIR) {
columnBuilder.appendNull();
} else {
@@ -197,6 +285,9 @@ public class TableLastQueryOperator extends AbstractDataSourceOperator {
// read last value from File or MemTable, update last cache
if (!currentUnCacheMeasurements.isEmpty()) {
+
+ // avoid invoke twice
+
// TABLE_DEVICE_SCHEMA_CACHE.initOrInvalidateLastCache();
// TABLE_DEVICE_SCHEMA_CACHE.updateLastCacheIfExists
@@ -214,7 +305,7 @@ public class TableLastQueryOperator extends AbstractDataSourceOperator {
}
@Override
- protected List<TSDataType> getResultDataTypes() {
+ public List<TSDataType> getResultDataTypes() {
int groupingKeySize = groupingKeySchemas != null ? groupingKeySchemas.size() : 0;
List<TSDataType> resultDataTypes = new ArrayList<>(groupingKeySize + tableAggregators.size());
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 40866a9ce30..fa671f243e5 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
@@ -1745,72 +1745,71 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
public Operator visitAggregationTableScan(
AggregationTableScanNode node, LocalExecutionPlanContext context) {
+ List<String> measurementColumnNames = new ArrayList<>();
+ List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
+ Map<String, Integer> measurementColumnsIndexMap = new HashMap<>();
+
List<TableAggregator> aggregators = new ArrayList<>(node.getAggregations().size());
- Map<Symbol, Integer> columnLayout = new HashMap<>(node.getAggregations().size());
+ List<Integer> aggregatorInputChannels =
+ new ArrayList<>(
+ (int)
+ node.getAggregations().values().stream()
+ .mapToLong(aggregation -> aggregation.getArguments().size())
+ .sum());
+ int aggDistinctArgumentCount =
+ (int)
+ node.getAggregations().values().stream()
+ .flatMap(aggregation -> aggregation.getArguments().stream())
+ .map(Symbol::from)
+ .distinct()
+ .count();
+ List<ColumnSchema> aggColumnSchemas = new ArrayList<>(aggDistinctArgumentCount);
+ Map<Symbol, Integer> aggColumnLayout = new HashMap<>(aggDistinctArgumentCount);
+ int[] aggColumnsIndexArray = new int[aggDistinctArgumentCount];
- int distinctArgumentCount = node.getAssignments().size();
- int aggregationsCount = node.getAggregations().size();
- List<Integer> aggregatorInputChannels = new ArrayList<>();
+ String timeColumnName = null;
int channel = 0;
int measurementColumnCount = 0;
- Map<Symbol, Integer> idAndAttributeColumnsIndexMap = node.getIdAndAttributeIndexMap();
- Map<Symbol, ColumnSchema> columnSchemaMap = node.getAssignments();
- List<ColumnSchema> columnSchemas = new ArrayList<>(aggregationsCount);
- int[] columnsIndexArray = new int[distinctArgumentCount];
- List<String> measurementColumnNames = new ArrayList<>();
- Map<String, Integer> measurementColumnsIndexMap = new HashMap<>();
- String timeColumnName = null;
- List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
-
for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : node.getAggregations().entrySet()) {
- AggregationNode.Aggregation aggregation = entry.getValue();
-
- for (Expression argument : aggregation.getArguments()) {
+ for (Expression argument : entry.getValue().getArguments()) {
Symbol symbol = Symbol.from(argument);
- ColumnSchema schema = requireNonNull(columnSchemaMap.get(symbol), symbol + " is null");
- switch (schema.getColumnCategory()) {
- case ID:
- case ATTRIBUTE:
- if (!columnLayout.containsKey(symbol)) {
- columnsIndexArray[channel] =
- requireNonNull(idAndAttributeColumnsIndexMap.get(symbol), symbol + " is null");
- columnSchemas.add(schema);
- }
- break;
- case MEASUREMENT:
- if (!columnLayout.containsKey(symbol)) {
- columnsIndexArray[channel] = measurementColumnCount;
+ ColumnSchema schema =
+ requireNonNull(node.getAssignments().get(symbol), symbol + " is null");
+ if (!aggColumnLayout.containsKey(symbol)) {
+ switch (schema.getColumnCategory()) {
+ case ID:
+ case ATTRIBUTE:
+ aggColumnsIndexArray[channel] =
+ requireNonNull(node.getIdAndAttributeIndexMap().get(symbol), symbol + " is null");
+ break;
+ case MEASUREMENT:
+ aggColumnsIndexArray[channel] = measurementColumnCount;
measurementColumnCount++;
measurementColumnNames.add(schema.getName());
measurementSchemas.add(
new MeasurementSchema(schema.getName(), getTSDataType(schema.getType())));
- columnSchemas.add(schema);
measurementColumnsIndexMap.put(symbol.getName(), measurementColumnCount - 1);
- }
- break;
- case TIME:
- if (!columnLayout.containsKey(symbol)) {
- columnsIndexArray[channel] = -1;
- columnSchemas.add(schema);
+ break;
+ case TIME:
+ aggColumnsIndexArray[channel] = -1;
timeColumnName = symbol.getName();
- }
- break;
- default:
- throw new IllegalArgumentException(
- "Unexpected column category: " + schema.getColumnCategory());
- }
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unexpected column category: " + schema.getColumnCategory());
+ }
- if (!columnLayout.containsKey(symbol)) {
+ aggColumnSchemas.add(schema);
aggregatorInputChannels.add(channel);
- columnLayout.put(symbol, channel++);
+ aggColumnLayout.put(symbol, channel++);
} else {
- aggregatorInputChannels.add(columnLayout.get(symbol));
+ aggregatorInputChannels.add(aggColumnLayout.get(symbol));
}
}
}
for (Map.Entry<Symbol, ColumnSchema> entry : node.getAssignments().entrySet()) {
- if (!columnLayout.containsKey(entry.getKey())
+ if (!aggColumnLayout.containsKey(entry.getKey())
&& entry.getValue().getColumnCategory() == MEASUREMENT) {
measurementColumnCount++;
measurementColumnNames.add(entry.getValue().getName());
@@ -1830,7 +1829,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : node.getAggregations().entrySet()) {
aggregators.add(
buildAggregator(
- columnLayout,
+ aggColumnLayout,
entry.getKey(),
entry.getValue(),
node.getStep(),
@@ -1848,9 +1847,9 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
for (int i = 0; i < node.getGroupingKeys().size(); i++) {
Symbol groupingKey = node.getGroupingKeys().get(i);
- if (idAndAttributeColumnsIndexMap.containsKey(groupingKey)) {
- groupingKeySchemas.add(columnSchemaMap.get(groupingKey));
- groupingKeyIndex[i] = idAndAttributeColumnsIndexMap.get(groupingKey);
+ if (node.getIdAndAttributeIndexMap().containsKey(groupingKey)) {
+ groupingKeySchemas.add(node.getAssignments().get(groupingKey));
+ groupingKeyIndex[i] = node.getIdAndAttributeIndexMap().get(groupingKey);
} else {
if (node.getProjection() != null
&& !node.getProjection().getMap().isEmpty()
@@ -1894,7 +1893,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
SeriesScanOptions seriesScanOptions =
buildSeriesScanOptions(
context,
- columnSchemaMap,
+ node.getAssignments(),
measurementColumnNames,
measurementColumnsIndexMap,
timeColumnName,
@@ -1910,21 +1909,18 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
new TableAggregationTableScanOperator(
node.getPlanNodeId(),
operatorContext,
- columnSchemas,
- columnsIndexArray,
+ aggColumnSchemas,
+ aggColumnsIndexArray,
node.getDeviceEntries(),
- scanAscending ? Ordering.ASC : Ordering.DESC,
seriesScanOptions,
measurementColumnNames,
allSensors,
measurementSchemas,
- measurementColumnCount,
aggregators,
groupingKeySchemas,
groupingKeyIndex,
timeRangeIterator,
scanAscending,
- calculateMaxAggregationResultSize(),
canUseStatistic,
aggregatorInputChannels);
@@ -2062,30 +2058,4 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
// TODO complete this method
return true;
}
-
- 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);
- }
}