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/16 14:12:04 UTC
(iotdb) branch last_first_table_model updated (97ccc28a359 -> 0e38315b711)
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a change to branch last_first_table_model
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from 97ccc28a359 tmp
add 0fe56983088 Add order by to make the rightness of group by integration test
add faf58883fee Pipe: Support syncing table model data between clusters (skeleton) (#13731)
add 4d1552947cc IoTV2: Introduce pipeTaskRestartTimes for ensuring replicate events as order (#13730)
add d6a64705d6f Subscription: improve event consistency by operating within the `compute` method (#13787)
new 5688ff761e5 Merge branch 'master' into last_first_table_model
new 0e38315b711 fix last
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:
.../db/it/IoTDBMultiIDsWithAttributesTableIT.java | 8 +-
.../evolvable/batch/PipeTabletEventPlainBatch.java | 44 +++-
...hReq.java => PipeTransferTabletBatchReqV2.java} | 79 +++---
.../request/PipeTransferTabletBinaryReq.java | 10 +-
...Req.java => PipeTransferTabletBinaryReqV2.java} | 101 +++++---
.../request/PipeTransferTabletInsertNodeReq.java | 12 +-
...java => PipeTransferTabletInsertNodeReqV2.java} | 89 ++++---
.../request/PipeTransferTabletRawReq.java | 5 +-
...RawReq.java => PipeTransferTabletRawReqV2.java} | 83 ++++---
.../request/PipeTransferTsFileSealWithModReq.java | 37 ++-
.../pipeconsensus/PipeConsensusAsyncConnector.java | 5 +-
.../pipeconsensus/PipeConsensusSyncConnector.java | 5 +-
.../PipeConsensusTransferBatchReqBuilder.java | 1 +
.../async/IoTDBDataRegionAsyncConnector.java | 31 ++-
.../async/handler/PipeTransferTsFileHandler.java | 16 +-
.../thrift/sync/IoTDBDataRegionSyncConnector.java | 46 ++--
.../db/pipe/event/common/row/PipeRowCollector.java | 2 +-
.../common/tablet/PipeRawTabletInsertionEvent.java | 2 +-
.../query/TsFileInsertionEventQueryParser.java | 12 +-
.../scan/TsFileInsertionEventScanParser.java | 2 +-
.../pipe/metric/PipeDataNodeReceiverMetrics.java | 84 +++++++
.../pipeconsensus/PipeConsensusReceiver.java | 73 ++++--
.../protocol/thrift/IoTDBDataNodeReceiver.java | 130 +++++++++-
.../TableAggregationTableScanOperator.java | 105 ++++++--
.../relational/aggregation/AvgAccumulator.java | 10 +-
.../relational/aggregation/CountAccumulator.java | 6 +-
.../relational/aggregation/LastAccumulator.java | 18 +-
.../relational/aggregation/LastByAccumulator.java | 2 +-
.../relational/aggregation/SumAccumulator.java | 8 +-
.../relational/aggregation/TableAccumulator.java | 2 +-
.../relational/aggregation/TableAggregator.java | 8 +-
.../plan/planner/TableOperatorGenerator.java | 102 ++++----
.../plan/statement/crud/LoadTsFileStatement.java | 4 +
.../broker/SubscriptionPrefetchingQueue.java | 152 +++++++-----
.../broker/SubscriptionPrefetchingTabletQueue.java | 148 ++++++-----
.../broker/SubscriptionPrefetchingTsFileQueue.java | 274 +++++++++++----------
.../connector/PipeDataNodeThriftRequestTest.java | 203 +++++++++++++++
.../pipe/event/TsFileInsertionEventParserTest.java | 23 +-
.../payload/thrift/request/PipeRequestType.java | 5 +
.../thrift/request/PipeTransferFileSealReqV2.java | 6 +-
.../commons/pipe/event/PipeInsertionEvent.java | 2 +-
.../org/apache/iotdb/commons/utils/FileUtils.java | 4 +-
.../src/main/thrift/pipeconsensus.thrift | 3 +-
43 files changed, 1375 insertions(+), 587 deletions(-)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/{PipeTransferTabletBatchReq.java => PipeTransferTabletBatchReqV2.java} (71%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/{PipeTransferTabletBinaryReq.java => PipeTransferTabletBinaryReqV2.java} (51%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/{PipeTransferTabletInsertNodeReq.java => PipeTransferTabletInsertNodeReqV2.java} (53%)
copy iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/{PipeTransferTabletRawReq.java => PipeTransferTabletRawReqV2.java} (67%)
(iotdb) 01/02: Merge branch 'master' into last_first_table_model
Posted by ca...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch last_first_table_model
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5688ff761e58e097b2262300d21368ca659de574
Merge: 97ccc28a359 d6a64705d6f
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Oct 16 19:55:41 2024 +0800
Merge branch 'master' into last_first_table_model
.../db/it/IoTDBMultiIDsWithAttributesTableIT.java | 8 +-
.../evolvable/batch/PipeTabletEventPlainBatch.java | 44 +++-
.../request/PipeTransferTabletBatchReqV2.java | 241 ++++++++++++++++++
.../request/PipeTransferTabletBinaryReq.java | 10 +-
...Req.java => PipeTransferTabletBinaryReqV2.java} | 101 +++++---
.../request/PipeTransferTabletInsertNodeReq.java | 12 +-
...java => PipeTransferTabletInsertNodeReqV2.java} | 89 ++++---
.../request/PipeTransferTabletRawReq.java | 5 +-
...RawReq.java => PipeTransferTabletRawReqV2.java} | 83 ++++---
.../request/PipeTransferTsFileSealWithModReq.java | 37 ++-
.../pipeconsensus/PipeConsensusAsyncConnector.java | 5 +-
.../pipeconsensus/PipeConsensusSyncConnector.java | 5 +-
.../PipeConsensusTransferBatchReqBuilder.java | 1 +
.../async/IoTDBDataRegionAsyncConnector.java | 31 ++-
.../async/handler/PipeTransferTsFileHandler.java | 16 +-
.../thrift/sync/IoTDBDataRegionSyncConnector.java | 46 ++--
.../db/pipe/event/common/row/PipeRowCollector.java | 2 +-
.../common/tablet/PipeRawTabletInsertionEvent.java | 2 +-
.../query/TsFileInsertionEventQueryParser.java | 12 +-
.../scan/TsFileInsertionEventScanParser.java | 2 +-
.../pipe/metric/PipeDataNodeReceiverMetrics.java | 84 +++++++
.../pipeconsensus/PipeConsensusReceiver.java | 73 ++++--
.../protocol/thrift/IoTDBDataNodeReceiver.java | 130 +++++++++-
.../plan/statement/crud/LoadTsFileStatement.java | 4 +
.../broker/SubscriptionPrefetchingQueue.java | 152 +++++++-----
.../broker/SubscriptionPrefetchingTabletQueue.java | 148 ++++++-----
.../broker/SubscriptionPrefetchingTsFileQueue.java | 274 +++++++++++----------
.../connector/PipeDataNodeThriftRequestTest.java | 203 +++++++++++++++
.../pipe/event/TsFileInsertionEventParserTest.java | 23 +-
.../payload/thrift/request/PipeRequestType.java | 5 +
.../thrift/request/PipeTransferFileSealReqV2.java | 6 +-
.../commons/pipe/event/PipeInsertionEvent.java | 2 +-
.../org/apache/iotdb/commons/utils/FileUtils.java | 4 +-
.../src/main/thrift/pipeconsensus.thrift | 3 +-
34 files changed, 1415 insertions(+), 448 deletions(-)
(iotdb) 02/02: fix last
Posted by ca...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch last_first_table_model
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0e38315b711dbd90b9872890d7d71977773ea36e
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Oct 16 22:11:48 2024 +0800
fix last
---
.../TableAggregationTableScanOperator.java | 105 +++++++++++++++------
.../relational/aggregation/AvgAccumulator.java | 10 +-
.../relational/aggregation/CountAccumulator.java | 6 +-
.../relational/aggregation/LastAccumulator.java | 18 ++--
.../relational/aggregation/LastByAccumulator.java | 2 +-
.../relational/aggregation/SumAccumulator.java | 8 +-
.../relational/aggregation/TableAccumulator.java | 2 +-
.../relational/aggregation/TableAggregator.java | 8 +-
.../plan/planner/TableOperatorGenerator.java | 102 ++++++++++----------
9 files changed, 156 insertions(+), 105 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 51d2272cef1..734c075f9ff 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
@@ -43,9 +43,11 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.statistics.Statistics;
+import org.apache.tsfile.file.metadata.statistics.StringStatistics;
import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.BinaryColumn;
import org.apache.tsfile.read.common.block.column.LongColumn;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.utils.Binary;
@@ -101,7 +103,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
private final int maxTsBlockLineNum;
// for different aggregations aiming to same column, use this variable to point to same column
- private final int[] layoutArray;
+ private final List<Integer> aggArguments;
private QueryDataSource queryDataSource;
@@ -129,7 +131,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
GroupByTimeParameter groupByTimeParameter,
long maxReturnSize,
boolean canUseStatistics,
- int[] layoutArray) {
+ List<Integer> aggArguments) {
super(
sourceId,
@@ -161,7 +163,7 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
this.measurementColumnTSDataTypes =
measurementSchemas.stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
this.currentDeviceIndex = 0;
- this.layoutArray = layoutArray;
+ this.aggArguments = aggArguments;
this.timeIterator = tableTimeRangeIterator;
if (tableTimeRangeIterator.getType()
== ITableTimeRangeIterator.TimeIteratorType.SINGLE_TIME_ITERATOR) {
@@ -400,29 +402,32 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
TsBlock inputRegion = inputTsBlock.getRegion(0, lastIndexToProcess + 1);
// TODO(beyyes) add optimization: if only agg measurement columns, no need to transfer
- Column[] valueColumns = new Column[tableAggregators.size()];
- for (int i = 0; i < tableAggregators.size(); i++) {
- ColumnSchema columnSchema = columnSchemas.get(layoutArray[i]);
+ Column[] valueColumns = new Column[aggArguments.size()];
+ for (int i = 0; i < aggArguments.size(); i++) {
+ ColumnSchema columnSchema = columnSchemas.get(aggArguments.get(i));
if (Streams.of(
TsTableColumnCategory.ID, TsTableColumnCategory.ATTRIBUTE, TsTableColumnCategory.TIME)
.anyMatch(columnSchema.getColumnCategory()::equals)) {
- valueColumns[i] = inputRegion.getTimeColumn();
+ if (isNullIdOrAttribute(i)) {
+ valueColumns[i] =
+ new RunLengthEncodedColumn(
+ new BinaryColumn(1, Optional.of(new boolean[] {true}), new Binary[] {null}),
+ inputRegion.getTimeColumn().getPositionCount());
+ } else {
+ valueColumns[i] = inputRegion.getTimeColumn();
+ }
} else {
- valueColumns[i] = inputRegion.getColumn(columnsIndexArray[i]);
+ valueColumns[i] = inputRegion.getColumn(columnsIndexArray[aggArguments.get(i)]);
}
}
+
TsBlock tsBlock =
new TsBlock(
inputRegion.getPositionCount(),
new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, inputRegion.getPositionCount()),
valueColumns);
- for (int i = 0; i < aggregators.size(); i++) {
- if (isNullIdOrAttribute(i)) {
- continue;
- }
-
- TableAggregator aggregator = aggregators.get(i);
+ for (TableAggregator aggregator : tableAggregators) {
// current agg method has been calculated
if (aggregator.hasFinalResult()) {
continue;
@@ -441,34 +446,80 @@ public class TableAggregationTableScanOperator extends AbstractSeriesAggregation
/** ID or Attribute is null, skip this aggregation logic */
private boolean isNullIdOrAttribute(int i) {
- if (TsTableColumnCategory.ID == columnSchemas.get(layoutArray[i]).getColumnCategory()
- && deviceEntries.get(currentDeviceIndex).getNthSegment(columnsIndexArray[i] + 1) == null) {
+ if (TsTableColumnCategory.ID == columnSchemas.get(aggArguments.get(i)).getColumnCategory()
+ && deviceEntries
+ .get(currentDeviceIndex)
+ .getNthSegment(columnsIndexArray[aggArguments.get(i)] + 1)
+ == null) {
return true;
}
- return TsTableColumnCategory.ATTRIBUTE == columnSchemas.get(layoutArray[i]).getColumnCategory()
+ return TsTableColumnCategory.ATTRIBUTE
+ == columnSchemas.get(aggArguments.get(i)).getColumnCategory()
&& deviceEntries
.get(currentDeviceIndex)
.getAttributeColumnValues()
- .get(columnsIndexArray[i])
+ .get(columnsIndexArray[aggArguments.get(i)])
== null;
}
protected void calcFromStatistics(Statistics timeStatistics, Statistics[] valueStatistics) {
- for (int i = 0; i < tableAggregators.size(); i++) {
- if (isNullIdOrAttribute(i)) {
- continue;
- }
+ int idx = -1;
+
+ for (TableAggregator aggregator : tableAggregators) {
- TableAggregator aggregator = tableAggregators.get(i);
if (aggregator.hasFinalResult()) {
continue;
}
- aggregator.processStatistics(
- columnSchemas.get(layoutArray[i]).getColumnCategory() == TsTableColumnCategory.MEASUREMENT
- ? valueStatistics[columnsIndexArray[i]]
- : timeStatistics);
+ Statistics[] statisticsArray = new Statistics[aggregator.getChannelCount()];
+
+ for (int i = 0; i < aggregator.getChannelCount(); i++) {
+ idx++;
+
+ TsTableColumnCategory columnSchemaCategory =
+ columnSchemas.get(aggArguments.get(idx)).getColumnCategory();
+ if (columnSchemaCategory == TsTableColumnCategory.TIME) {
+ statisticsArray[i] = timeStatistics;
+ } else if (columnSchemaCategory == TsTableColumnCategory.ATTRIBUTE) {
+ String attr =
+ deviceEntries
+ .get(currentDeviceIndex)
+ .getAttributeColumnValues()
+ .get(columnsIndexArray[aggArguments.get(i)]);
+ if (attr == null) {
+ statisticsArray[i] = null;
+ } else {
+ StringStatistics stringStatics = new StringStatistics();
+ stringStatics.setCount((int) timeStatistics.getCount());
+ Binary v = new Binary(attr.getBytes());
+ stringStatics.initializeStats(v, v, v, v);
+ statisticsArray[i] = stringStatics;
+ }
+ } else if (columnSchemaCategory == TsTableColumnCategory.ID) {
+ // TODO avoid create deviceStatics multi times
+ String id =
+ (String)
+ deviceEntries
+ .get(currentDeviceIndex)
+ .getNthSegment(columnsIndexArray[aggArguments.get(idx)] + 1);
+ if (id == null) {
+ statisticsArray[i] = null;
+ } else {
+ StringStatistics stringStatics = new StringStatistics();
+ stringStatics.setCount((int) timeStatistics.getCount());
+ stringStatics.setStartTime(timeStatistics.getStartTime());
+ stringStatics.setEndTime(timeStatistics.getEndTime());
+ Binary v = new Binary(id.getBytes());
+ stringStatics.initializeStats(v, v, v, v);
+ statisticsArray[i] = stringStatics;
+ }
+ } else {
+ statisticsArray[i] = valueStatistics[columnsIndexArray[aggArguments.get(idx)]];
+ }
+ }
+
+ aggregator.processStatistics(statisticsArray);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
index 283fb598b69..4c33e6ac371 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
@@ -137,16 +137,16 @@ public class AvgAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {
+ public void addStatistics(Statistics[] statistics) {
if (statistics == null) {
return;
}
initResult = true;
- countValue += statistics.getCount();
- if (statistics instanceof IntegerStatistics) {
- sumValue += statistics.getSumLongValue();
+ countValue += statistics[0].getCount();
+ if (statistics[0] instanceof IntegerStatistics) {
+ sumValue += statistics[0].getSumLongValue();
} else {
- sumValue += statistics.getSumDoubleValue();
+ sumValue += statistics[0].getSumDoubleValue();
}
if (countValue == 0) {
initResult = false;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
index b179e76a38c..0d1c8de9e47 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
@@ -81,11 +81,11 @@ public class CountAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {
- if (statistics == null) {
+ public void addStatistics(Statistics[] statistics) {
+ if (statistics[0] == null) {
return;
}
- countState += statistics.getCount();
+ countState += statistics[0].getCount();
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java
index 7bcb4d45f4f..a2100c899d9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java
@@ -197,32 +197,34 @@ public class LastAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {
- if (statistics == null) {
+ public void addStatistics(Statistics[] statistics) {
+ if (statistics[0] == null) {
return;
}
switch (seriesDataType) {
case INT32:
case DATE:
- updateIntLastValue((int) statistics.getLastValue(), statistics.getEndTime());
+ updateIntLastValue((int) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
case INT64:
+ updateLongLastValue((long) statistics[0].getLastValue(), statistics[0].getEndTime());
+ break;
case TIMESTAMP:
- updateLongLastValue((long) statistics.getLastValue(), statistics.getEndTime());
+ updateLongLastValue(statistics[0].getEndTime(), statistics[0].getEndTime());
break;
case FLOAT:
- updateFloatLastValue((float) statistics.getLastValue(), statistics.getEndTime());
+ updateFloatLastValue((float) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
case DOUBLE:
- updateDoubleLastValue((double) statistics.getLastValue(), statistics.getEndTime());
+ updateDoubleLastValue((double) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
case TEXT:
case BLOB:
case STRING:
- updateBinaryLastValue((Binary) statistics.getLastValue(), statistics.getEndTime());
+ updateBinaryLastValue((Binary) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
case BOOLEAN:
- updateBooleanLastValue((boolean) statistics.getLastValue(), statistics.getEndTime());
+ updateBooleanLastValue((boolean) statistics[0].getLastValue(), statistics[0].getEndTime());
break;
default:
throw new UnSupportedDataTypeException(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java
index c594eca1c74..dd475864574 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java
@@ -53,7 +53,7 @@ public class LastByAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {}
+ public void addStatistics(Statistics[] statistics) {}
@Override
public void reset() {}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/SumAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/SumAccumulator.java
index e76ad72d551..ad89c40c5f1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/SumAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/SumAccumulator.java
@@ -113,15 +113,15 @@ public class SumAccumulator implements TableAccumulator {
}
@Override
- public void addStatistics(Statistics statistics) {
+ public void addStatistics(Statistics[] statistics) {
if (statistics == null) {
return;
}
initResult = true;
- if (statistics instanceof IntegerStatistics) {
- sumValue += statistics.getSumLongValue();
+ if (statistics[0] instanceof IntegerStatistics) {
+ sumValue += statistics[0].getSumLongValue();
} else {
- sumValue += statistics.getSumDoubleValue();
+ sumValue += statistics[0].getSumDoubleValue();
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java
index 30cb908b43b..38c14de541a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java
@@ -40,7 +40,7 @@ public interface TableAccumulator {
* This method can only be used in AggTableScan, it will use different statistics based on the
* type of Accumulator.
*/
- void addStatistics(Statistics statistics);
+ void addStatistics(Statistics[] statistics);
void reset();
}
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 fae3dee8de0..d9cba3bbbb6 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
@@ -81,9 +81,7 @@ public class TableAggregator {
}
}
- /** Used for AggregateTableScanOperator. */
- public void processStatistics(Statistics statistics) {
- checkArgument(inputChannels.length == 1, "expected 1 input channel for processStatistics");
+ public void processStatistics(Statistics[] statistics) {
accumulator.addStatistics(statistics);
}
@@ -98,4 +96,8 @@ public class TableAggregator {
public long getEstimatedSize() {
return accumulator.getEstimatedSize();
}
+
+ public int getChannelCount() {
+ return this.inputChannels.length;
+ }
}
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 86f89288fb0..8a0c5176a71 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
@@ -1391,7 +1391,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
Map<Symbol, Integer> columnLayout = new HashMap<>(node.getAggregations().size());
int aggregationsCount = node.getAggregations().size();
- int[] layoutArray = new int[aggregationsCount];
+ List<Integer> aggColumnIndexes = new ArrayList<>();
int channel = 0;
int idx = -1;
int measurementColumnCount = 0;
@@ -1399,64 +1399,59 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
Map<Symbol, Integer> idAndAttributeColumnsIndexMap = node.getIdAndAttributeIndexMap();
Map<Symbol, ColumnSchema> columnSchemaMap = node.getAssignments();
List<ColumnSchema> columnSchemas = new ArrayList<>(aggregationsCount);
- int[] columnsIndexArray = new int[aggregationsCount];
+ int[] columnsIndexArray = new int[aggregationsCount * 2];
List<String> measurementColumnNames = new ArrayList<>();
List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
- // List<Integer>
for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : node.getAggregations().entrySet()) {
- idx++;
- Symbol symbol = Symbol.from(entry.getValue().getArguments().get(0));
- ColumnSchema schema = requireNonNull(columnSchemaMap.get(symbol), symbol + " is null");
-
String funcName = entry.getValue().getResolvedFunction().getSignature().getName();
- if (schema.getType().equals(BlobType.BLOB)
- && (FIRST.equals(funcName) || LAST.equals(funcName))) {
- // first/last aggregation with BLOB type can not use statistics
- canUseStatistic = false;
- }
- switch (schema.getColumnCategory()) {
- case ID:
- case ATTRIBUTE:
- if (columnLayout.containsKey(symbol)) {
- columnsIndexArray[idx] = columnsIndexArray[columnLayout.get(symbol)];
- } else {
- columnsIndexArray[idx] =
- requireNonNull(idAndAttributeColumnsIndexMap.get(symbol), symbol + " is null");
- columnSchemas.add(schema);
- }
- break;
- case MEASUREMENT:
- if (columnLayout.containsKey(symbol)) {
- columnsIndexArray[idx] = columnsIndexArray[columnLayout.get(symbol)];
- } else {
- columnsIndexArray[idx] = measurementColumnCount;
- measurementColumnCount++;
- measurementColumnNames.add(symbol.getName());
- measurementSchemas.add(
- new MeasurementSchema(schema.getName(), getTSDataType(schema.getType())));
- columnSchemas.add(schema);
- }
- break;
- case TIME:
- if (columnLayout.containsKey(symbol)) {
- columnsIndexArray[idx] = -1;
- } else {
- columnsIndexArray[idx] = -1;
- columnSchemas.add(schema);
- }
- break;
- default:
- throw new IllegalArgumentException(
- "Unexpected column category: " + schema.getColumnCategory());
- }
+ for (Expression argument : entry.getValue().getArguments()) {
+ idx++;
+ Symbol symbol = Symbol.from(argument);
+ ColumnSchema schema = requireNonNull(columnSchemaMap.get(symbol), symbol + " is null");
+ if (schema.getType().equals(BlobType.BLOB)
+ && (FIRST.equals(funcName) || LAST.equals(funcName))) {
+ // first/last aggregation with BLOB type can not use statistics
+ canUseStatistic = false;
+ }
- if (!columnLayout.containsKey(symbol)) {
- layoutArray[idx] = channel;
- columnLayout.put(symbol, channel++);
- } else {
- layoutArray[idx] = columnLayout.get(symbol);
+ 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;
+ measurementColumnCount++;
+ measurementColumnNames.add(symbol.getName());
+ measurementSchemas.add(
+ new MeasurementSchema(schema.getName(), getTSDataType(schema.getType())));
+ columnSchemas.add(schema);
+ }
+ break;
+ case TIME:
+ if (!columnLayout.containsKey(symbol)) {
+ columnsIndexArray[channel] = -1;
+ columnSchemas.add(schema);
+ }
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unexpected column category: " + schema.getColumnCategory());
+ }
+
+ if (!columnLayout.containsKey(symbol)) {
+ aggColumnIndexes.add(channel);
+ columnLayout.put(symbol, channel++);
+ } else {
+ aggColumnIndexes.add(columnLayout.get(symbol));
+ }
}
aggregators.add(
@@ -1468,6 +1463,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
context.getTypeProvider()));
}
+ // TODO if this needed?
for (Map.Entry<Symbol, ColumnSchema> entry : node.getAssignments().entrySet()) {
if (!columnLayout.containsKey(entry.getKey())
&& entry.getValue().getColumnCategory() == MEASUREMENT) {
@@ -1566,7 +1562,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
null,
calculateMaxAggregationResultSize(),
canUseStatistic,
- layoutArray);
+ aggColumnIndexes);
((DataDriverContext) context.getDriverContext()).addSourceOperator(aggTableScanOperator);