You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "Wei-hao-Li (via GitHub)" <gi...@apache.org> on 2026/03/12 15:20:22 UTC
[PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Wei-hao-Li opened a new pull request, #17294:
URL: https://github.com/apache/iotdb/pull/17294
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4060600882
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate failed**
Failed conditions
 [B Maintainability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "JackieTien97 (via GitHub)" <gi...@apache.org>.
JackieTien97 commented on code in PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#discussion_r2944024785
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeNonAlignedDeviceViewAggregationScanOperator.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
+
+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.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractDataSourceOperator;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.AbstractTableScanOperator.CURRENT_DEVICE_INDEX_STRING;
+
+public class TreeNonAlignedDeviceViewAggregationScanOperator
Review Comment:
you need to override close method to close `currentDeviceRootOperator` just like `DeviceIteratorScanOperator`.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java:
##########
@@ -2821,8 +2831,8 @@ private GroupedAggregator buildGroupByAggregator(
}
@Override
- public Operator visitAggregationTreeDeviceViewScan(
- AggregationTreeDeviceViewScanNode node, LocalExecutionPlanContext context) {
+ public Operator visitAlignedAggregationTreeDeviceViewScan(
Review Comment:
add `visitAggregationTreeDeviceViewScanNode` method and directly throw UnsupportedException for that, avoiding `AggregationTreeDeviceViewScanNode` visit its super class `AggregationTableScanNode`
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeNonAlignedDeviceViewAggregationScanOperator.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
+
+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.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractDataSourceOperator;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.AbstractTableScanOperator.CURRENT_DEVICE_INDEX_STRING;
+
+public class TreeNonAlignedDeviceViewAggregationScanOperator
+ extends AbstractDefaultAggTableScanOperator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(
+ TreeNonAlignedDeviceViewAggregationScanOperator.class);
+
+ private final IDeviceID.TreeDeviceIdColumnValueExtractor extractor;
+ private final DeviceIteratorScanOperator.DeviceChildOperatorTreeGenerator childOperatorGenerator;
+
+ private Operator child;
+ private List<Operator> dataSourceOperators;
+
+ public TreeNonAlignedDeviceViewAggregationScanOperator(
+ AbstractAggTableScanOperatorParameter parameter,
+ IDeviceID.TreeDeviceIdColumnValueExtractor extractor,
+ DeviceIteratorScanOperator.DeviceChildOperatorTreeGenerator childOperatorGenerator) {
+ super(parameter);
+ this.extractor = extractor;
+ this.childOperatorGenerator = childOperatorGenerator;
+ constructCurrentDeviceOperatorTree();
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex) {
+ return (String) extractor.extract(deviceEntry.getDeviceID(), idColumnIndex);
+ }
+
+ @Override
+ protected Optional<Boolean> calculateAggregationResultForCurrentTimeRange() {
+ try {
+ // Try to calculate from cached data
+ if (calcFromCachedData()) {
+ updateResultTsBlock();
+ checkIfAllAggregatorHasFinalResult();
+ return Optional.of(true);
+ }
+
+ // Read from child operator
+ if (readAndCalcFromChild()) {
+ updateResultTsBlock();
+ checkIfAllAggregatorHasFinalResult();
+ return Optional.of(true);
+ }
+
+ // No more data from child, finish the current device
+ if (!child.hasNext()) {
+ updateResultTsBlock();
+ timeIterator.resetCurTimeRange();
+ nextDevice();
+
+ if (currentDeviceIndex >= deviceCount) {
+ // All devices consumed
+ timeIterator.setFinished();
+ return Optional.of(true);
+ } else {
+ // More devices to process, child should provide next device's data
+ return Optional.of(false);
+ }
+ }
+
+ return Optional.of(false);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while processing aggregation from child operator", e);
+ }
+ }
+
+ /** Read data from child operator and calculate aggregation. */
+ private boolean readAndCalcFromChild() throws Exception {
+ long start = System.nanoTime();
+
+ while (System.nanoTime() - start < leftRuntimeOfOneNextCall && child.hasNext()) {
+ // Get next TsBlock from child
+ TsBlock tsBlock = child.nextWithTimer();
+ if (tsBlock == null || tsBlock.isEmpty()) {
+ continue;
+ }
+
+ // Calculate aggregation from raw data
+ if (calcUsingRawData(tsBlock)) {
+ return true;
+ }
+
+ // If not finished, continue reading from child
+ }
+
+ return false;
+ }
+
+ @Override
+ protected void nextDevice() throws Exception {
+ currentDeviceIndex++;
+ childOperatorGenerator.getCurrentDeviceStartCloseOperator().close();
+ if (currentDeviceIndex >= deviceEntries.size()) {
+ return;
+ }
+ constructCurrentDeviceOperatorTree();
+ queryDataSource.reset();
+ initQueryDataSource(queryDataSource);
+ this.operatorContext.recordSpecifiedInfo(
+ CURRENT_DEVICE_INDEX_STRING, Integer.toString(currentDeviceIndex));
+ }
+
+ private void constructCurrentDeviceOperatorTree() {
+ if (this.deviceEntries.isEmpty()) {
+ return;
+ }
+ if (this.deviceEntries.get(this.currentDeviceIndex) == null) {
+ throw new IllegalStateException(
+ "Device entries of index " + this.currentDeviceIndex + " is empty");
+ }
+ DeviceEntry deviceEntry = this.deviceEntries.get(this.currentDeviceIndex);
+
+ childOperatorGenerator.generateCurrentDeviceOperatorTree(deviceEntry, false);
+ child = childOperatorGenerator.getCurrentDeviceRootOperator();
+ dataSourceOperators = childOperatorGenerator.getCurrentDeviceDataSourceOperators();
+ }
+
+ /** same with {@link DeviceIteratorScanOperator#initQueryDataSource(IQueryDataSource)} */
+ @Override
+ public void initQueryDataSource(IQueryDataSource dataSource) {
+ if (resultTsBlockBuilder == null) {
+ // only need to do this when init firstly
+ this.queryDataSource = (QueryDataSource) dataSource;
+ this.resultTsBlockBuilder = new TsBlockBuilder(getResultDataTypes());
+ }
+
+ if (dataSourceOperators == null || dataSourceOperators.isEmpty()) {
+ return;
+ }
+
+ for (Operator operator : dataSourceOperators) {
+ ((AbstractDataSourceOperator) operator).initQueryDataSource(dataSource);
+ }
+ }
+
+ @Override
+ protected void checkIfAllAggregatorHasFinalResult() throws Exception {
+ if (allAggregatorsHasFinalResult
+ && (timeIterator.getType() == ITableTimeRangeIterator.TimeIteratorType.SINGLE_TIME_ITERATOR
+ || tableAggregators.isEmpty())) {
+ nextDevice();
+ inputTsBlock = null;
+
+ if (currentDeviceIndex >= deviceCount) {
+ // all devices have been consumed
+ timeIterator.setFinished();
+ }
+
+ allAggregatorsHasFinalResult = false;
+ }
+ }
+
+ @Override
+ public long calculateMaxPeekMemory() {
+ return child.calculateMaxPeekMemory();
+ }
+
+ @Override
+ public long calculateMaxReturnSize() {
+ return child.calculateMaxReturnSize();
+ }
+
+ @Override
+ public long calculateRetainedSizeAfterCallingNext() {
+ return child.calculateRetainedSizeAfterCallingNext();
+ }
+
+ @Override
+ public long ramBytesUsed() {
Review Comment:
The estimation includes child and dataSourceOperators but doesn't include the childOperatorGenerator, which may hold references to the operator tree templates. Minor memory accounting concern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4053329942
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate failed**
Failed conditions
 [B Maintainability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4072152870
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate failed**
Failed conditions
 [B Maintainability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4051211762
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate failed**
Failed conditions
 [B Maintainability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4047681576
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate failed**
Failed conditions
 [B Maintainability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4071846788
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate failed**
Failed conditions
 [B Maintainability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4089124855
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate failed**
Failed conditions
 [5.5% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=17294&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "JackieTien97 (via GitHub)" <gi...@apache.org>.
JackieTien97 commented on code in PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#discussion_r2944019300
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeNonAlignedDeviceViewAggregationScanOperator.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
+
+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.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractDataSourceOperator;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.AbstractTableScanOperator.CURRENT_DEVICE_INDEX_STRING;
+
+public class TreeNonAlignedDeviceViewAggregationScanOperator
+ extends AbstractDefaultAggTableScanOperator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(
+ TreeNonAlignedDeviceViewAggregationScanOperator.class);
+
+ private final IDeviceID.TreeDeviceIdColumnValueExtractor extractor;
+ private final DeviceIteratorScanOperator.DeviceChildOperatorTreeGenerator childOperatorGenerator;
+
+ private Operator child;
+ private List<Operator> dataSourceOperators;
+
+ public TreeNonAlignedDeviceViewAggregationScanOperator(
+ AbstractAggTableScanOperatorParameter parameter,
+ IDeviceID.TreeDeviceIdColumnValueExtractor extractor,
+ DeviceIteratorScanOperator.DeviceChildOperatorTreeGenerator childOperatorGenerator) {
+ super(parameter);
+ this.extractor = extractor;
+ this.childOperatorGenerator = childOperatorGenerator;
+ constructCurrentDeviceOperatorTree();
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex) {
+ return (String) extractor.extract(deviceEntry.getDeviceID(), idColumnIndex);
+ }
+
+ @Override
+ protected Optional<Boolean> calculateAggregationResultForCurrentTimeRange() {
+ try {
+ // Try to calculate from cached data
+ if (calcFromCachedData()) {
+ updateResultTsBlock();
+ checkIfAllAggregatorHasFinalResult();
+ return Optional.of(true);
+ }
+
+ // Read from child operator
+ if (readAndCalcFromChild()) {
+ updateResultTsBlock();
+ checkIfAllAggregatorHasFinalResult();
+ return Optional.of(true);
+ }
+
+ // No more data from child, finish the current device
+ if (!child.hasNext()) {
+ updateResultTsBlock();
+ timeIterator.resetCurTimeRange();
+ nextDevice();
+
+ if (currentDeviceIndex >= deviceCount) {
+ // All devices consumed
+ timeIterator.setFinished();
+ return Optional.of(true);
+ } else {
+ // More devices to process, child should provide next device's data
+ return Optional.of(false);
+ }
+ }
+
+ return Optional.of(false);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while processing aggregation from child operator", e);
+ }
+ }
+
+ /** Read data from child operator and calculate aggregation. */
+ private boolean readAndCalcFromChild() throws Exception {
+ long start = System.nanoTime();
+
+ while (System.nanoTime() - start < leftRuntimeOfOneNextCall && child.hasNext()) {
+ // Get next TsBlock from child
+ TsBlock tsBlock = child.nextWithTimer();
+ if (tsBlock == null || tsBlock.isEmpty()) {
+ continue;
+ }
+
+ // Calculate aggregation from raw data
+ if (calcUsingRawData(tsBlock)) {
+ return true;
+ }
+
+ // If not finished, continue reading from child
+ }
+
+ return false;
Review Comment:
```suggestion
if (child.hasNext()) {
// Get next TsBlock from child
TsBlock tsBlock = child.nextWithTimer();
if (tsBlock == null || tsBlock.isEmpty()) {
return false;
}
// Calculate aggregation from raw data
return calcUsingRawData(tsBlock);
}
return false;
```
no need to do flow and time control here, what you need to do is processing one batch and then give the control back to super class where time slice control and result builder full judgement happend.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeNonAlignedDeviceViewAggregationScanOperator.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
+
+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.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractDataSourceOperator;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.AbstractTableScanOperator.CURRENT_DEVICE_INDEX_STRING;
+
+public class TreeNonAlignedDeviceViewAggregationScanOperator
+ extends AbstractDefaultAggTableScanOperator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(
+ TreeNonAlignedDeviceViewAggregationScanOperator.class);
+
+ private final IDeviceID.TreeDeviceIdColumnValueExtractor extractor;
+ private final DeviceIteratorScanOperator.DeviceChildOperatorTreeGenerator childOperatorGenerator;
+
+ private Operator child;
+ private List<Operator> dataSourceOperators;
+
+ public TreeNonAlignedDeviceViewAggregationScanOperator(
+ AbstractAggTableScanOperatorParameter parameter,
+ IDeviceID.TreeDeviceIdColumnValueExtractor extractor,
+ DeviceIteratorScanOperator.DeviceChildOperatorTreeGenerator childOperatorGenerator) {
+ super(parameter);
+ this.extractor = extractor;
+ this.childOperatorGenerator = childOperatorGenerator;
+ constructCurrentDeviceOperatorTree();
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex) {
+ return (String) extractor.extract(deviceEntry.getDeviceID(), idColumnIndex);
+ }
+
+ @Override
+ protected Optional<Boolean> calculateAggregationResultForCurrentTimeRange() {
Review Comment:
```suggestion
protected Optional<Boolean> calculateAggregationResultForCurrentTimeRange() throws Exception {
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeNonAlignedDeviceViewAggregationScanOperator.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
+
+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.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractDataSourceOperator;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.AbstractTableScanOperator.CURRENT_DEVICE_INDEX_STRING;
+
+public class TreeNonAlignedDeviceViewAggregationScanOperator
+ extends AbstractDefaultAggTableScanOperator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(
+ TreeNonAlignedDeviceViewAggregationScanOperator.class);
+
+ private final IDeviceID.TreeDeviceIdColumnValueExtractor extractor;
+ private final DeviceIteratorScanOperator.DeviceChildOperatorTreeGenerator childOperatorGenerator;
+
+ private Operator child;
+ private List<Operator> dataSourceOperators;
+
+ public TreeNonAlignedDeviceViewAggregationScanOperator(
+ AbstractAggTableScanOperatorParameter parameter,
+ IDeviceID.TreeDeviceIdColumnValueExtractor extractor,
+ DeviceIteratorScanOperator.DeviceChildOperatorTreeGenerator childOperatorGenerator) {
+ super(parameter);
+ this.extractor = extractor;
+ this.childOperatorGenerator = childOperatorGenerator;
+ constructCurrentDeviceOperatorTree();
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex) {
+ return (String) extractor.extract(deviceEntry.getDeviceID(), idColumnIndex);
+ }
+
+ @Override
+ protected Optional<Boolean> calculateAggregationResultForCurrentTimeRange() {
+ try {
+ // Try to calculate from cached data
+ if (calcFromCachedData()) {
+ updateResultTsBlock();
+ checkIfAllAggregatorHasFinalResult();
+ return Optional.of(true);
+ }
+
+ // Read from child operator
+ if (readAndCalcFromChild()) {
+ updateResultTsBlock();
+ checkIfAllAggregatorHasFinalResult();
+ return Optional.of(true);
+ }
+
+ // No more data from child, finish the current device
+ if (!child.hasNext()) {
+ updateResultTsBlock();
+ timeIterator.resetCurTimeRange();
+ nextDevice();
+
+ if (currentDeviceIndex >= deviceCount) {
+ // All devices consumed
+ timeIterator.setFinished();
+ return Optional.of(true);
+ } else {
+ // More devices to process, child should provide next device's data
+ return Optional.of(false);
+ }
+ }
+
+ return Optional.of(false);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while processing aggregation from child operator", e);
+ }
Review Comment:
```suggestion
```
Exceptions from child operations are wrapped in RuntimeException. The parent class AbstractDefaultAggTableScanOperator declares this method as throws Exception, so you could propagate the checked exception directly instead of wrapping.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4072337096
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate passed**
Issues
 [34 New issues](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=17294&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=17294&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=17294&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=17294&metric=new_coverage&view=list)
 [4.1% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=17294&metric=new_duplicated_lines_density&view=list)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4048126002
## [Codecov](https://app.codecov.io/gh/apache/iotdb/pull/17294?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `88.02395%` with `20 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 39.77%. Comparing base ([`0012727`](https://app.codecov.io/gh/apache/iotdb/commit/00127279cf1c3c84b1fe57149c730aae61369ca3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`26259ef`](https://app.codecov.io/gh/apache/iotdb/commit/26259ef0663f90ad708e8183128d2bd134d93c57?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 7 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/iotdb/pull/17294?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...nner/distribute/TableDistributedPlanGenerator.java](https://app.codecov.io/gh/apache/iotdb/pull/17294?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fqueryengine%2Fplan%2Frelational%2Fplanner%2Fdistribute%2FTableDistributedPlanGenerator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvcXVlcnllbmdpbmUvcGxhbi9yZWxhdGlvbmFsL3BsYW5uZXIvZGlzdHJpYnV0ZS9UYWJsZURpc3RyaWJ1dGVkUGxhbkdlbmVyYXRvci5qYXZh) | 93.46% | [10 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/17294?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...node/AlignedAggregationTreeDeviceViewScanNode.java](https://app.codecov.io/gh/apache/iotdb/pull/17294?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fqueryengine%2Fplan%2Frelational%2Fplanner%2Fnode%2FAlignedAggregationTreeDeviceViewScanNode.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvcXVlcnllbmdpbmUvcGxhbi9yZWxhdGlvbmFsL3BsYW5uZXIvbm9kZS9BbGlnbmVkQWdncmVnYXRpb25UcmVlRGV2aWNlVmlld1NjYW5Ob2RlLmphdmE=) | 28.57% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/17294?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...e/NonAlignedAggregationTreeDeviceViewScanNode.java](https://app.codecov.io/gh/apache/iotdb/pull/17294?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fqueryengine%2Fplan%2Frelational%2Fplanner%2Fnode%2FNonAlignedAggregationTreeDeviceViewScanNode.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvcXVlcnllbmdpbmUvcGxhbi9yZWxhdGlvbmFsL3BsYW5uZXIvbm9kZS9Ob25BbGlnbmVkQWdncmVnYXRpb25UcmVlRGV2aWNlVmlld1NjYW5Ob2RlLmphdmE=) | 28.57% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/17294?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #17294 +/- ##
============================================
+ Coverage 39.70% 39.77% +0.06%
Complexity 282 282
============================================
Files 5100 5102 +2
Lines 341950 342273 +323
Branches 43559 43610 +51
============================================
+ Hits 135769 136136 +367
+ Misses 206181 206137 -44
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/iotdb/pull/17294?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4088994352
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate failed**
Failed conditions
 [5.5% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=17294&metric=new_duplicated_lines_density&view=list) (required ≤ 5%)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "JackieTien97 (via GitHub)" <gi...@apache.org>.
JackieTien97 merged PR #17294:
URL: https://github.com/apache/iotdb/pull/17294
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#issuecomment-4066866746
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) **Quality Gate failed**
Failed conditions
 [B Maintainability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294) (required ≥ A)
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=17294)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarQube for IDE](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "JackieTien97 (via GitHub)" <gi...@apache.org>.
JackieTien97 commented on code in PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#discussion_r2940100537
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java:
##########
@@ -2852,6 +2862,70 @@ public Operator visitAggregationTreeDeviceViewScan(
return treeAlignedDeviceViewAggregationScanOperator;
}
+ @Override
+ public Operator visitNonAlignedAggregationTreeDeviceViewScan(
+ NonAlignedAggregationTreeDeviceViewScanNode node, LocalExecutionPlanContext context) {
+ QualifiedObjectName qualifiedObjectName = node.getQualifiedObjectName();
+ TsTable tsTable =
+ DataNodeTableCache.getInstance()
+ .getTable(qualifiedObjectName.getDatabaseName(), qualifiedObjectName.getObjectName());
+ IDeviceID.TreeDeviceIdColumnValueExtractor idColumnValueExtractor =
+ createTreeDeviceIdColumnValueExtractor(DataNodeTreeViewSchemaUtils.getPrefixPath(tsTable));
+
+ AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter parameter =
+ constructAbstractAggTableScanOperatorParameter(
+ node,
+ context,
+ TreeAlignedDeviceViewAggregationScanOperator.class.getSimpleName(),
Review Comment:
```suggestion
TreeNonAlignedDeviceViewAggregationScanOperator.class.getSimpleName(),
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java:
##########
@@ -2852,6 +2862,70 @@ public Operator visitAggregationTreeDeviceViewScan(
return treeAlignedDeviceViewAggregationScanOperator;
}
+ @Override
+ public Operator visitNonAlignedAggregationTreeDeviceViewScan(
+ NonAlignedAggregationTreeDeviceViewScanNode node, LocalExecutionPlanContext context) {
+ QualifiedObjectName qualifiedObjectName = node.getQualifiedObjectName();
+ TsTable tsTable =
+ DataNodeTableCache.getInstance()
+ .getTable(qualifiedObjectName.getDatabaseName(), qualifiedObjectName.getObjectName());
+ IDeviceID.TreeDeviceIdColumnValueExtractor idColumnValueExtractor =
+ createTreeDeviceIdColumnValueExtractor(DataNodeTreeViewSchemaUtils.getPrefixPath(tsTable));
+
+ AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter parameter =
+ constructAbstractAggTableScanOperatorParameter(
+ node,
+ context,
+ TreeAlignedDeviceViewAggregationScanOperator.class.getSimpleName(),
+ node.getMeasurementColumnNameMap(),
+ tsTable.getCachedTableTTL());
+
+ // construct source operator (generator)
+ TreeNonAlignedDeviceViewScanNode scanNode =
+ new TreeNonAlignedDeviceViewScanNode(
+ node.getPlanNodeId(),
+ node.getQualifiedObjectName(),
+ // the outputSymbols of AggTableScanNode is not equals with TableScanNode
+ parameter.getOutputSymbols(),
+ node.getAssignments(),
+ node.getDeviceEntries(),
+ node.getTagAndAttributeIndexMap(),
+ node.getScanOrder(),
+ node.getTimePredicate().orElse(null),
+ node.getPushDownPredicate(),
+ node.getPushDownLimit(),
+ node.getPushDownOffset(),
+ node.isPushLimitToEachDevice(),
+ true,
+ node.getTreeDBName(),
+ node.getMeasurementColumnNameMap());
+
+ Operator sourceOperator = visitTreeNonAlignedDeviceViewScan(scanNode, context);
+ if (sourceOperator instanceof DeviceIteratorScanOperator) {
+ // Use deviceChildOperatorTreeGenerator directly, we will control switch of devices in
+ // TreeNonAlignedDeviceViewAggregationScanOperator
+ TreeNonAlignedDeviceViewAggregationScanOperator aggTableScanOperator =
+ new TreeNonAlignedDeviceViewAggregationScanOperator(
+ parameter,
+ idColumnValueExtractor,
+ ((DeviceIteratorScanOperator) sourceOperator).getDeviceChildOperatorTreeGenerator());
+
+ addSource(
+ aggTableScanOperator,
+ context,
+ node,
+ parameter.getMeasurementColumnNames(),
+ parameter.getMeasurementSchemas(),
+ parameter.getAllSensors(),
+ AggregationTableScanNode.class.getSimpleName());
Review Comment:
```suggestion
NonAlignedAggregationTreeDeviceViewScanNode.class.getSimpleName());
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java:
##########
@@ -1188,70 +1190,19 @@ private boolean prefixMatched(OrderingScheme childOrdering, List<Symbol> preGrou
@Override
public List<PlanNode> visitAggregationTableScan(
AggregationTableScanNode node, PlanContext context) {
- String dbName =
- node instanceof AggregationTreeDeviceViewScanNode
- ? ((AggregationTreeDeviceViewScanNode) node).getTreeDBName()
- : node.getQualifiedObjectName().getDatabaseName();
+ String dbName = node.getQualifiedObjectName().getDatabaseName();
DataPartition dataPartition = analysis.getDataPartitionInfo();
if (dbName == null || dataPartition == null) {
node.setRegionReplicaSet(NOT_ASSIGNED);
return Collections.singletonList(node);
}
- boolean needSplit = false;
- List<List<TRegionReplicaSet>> regionReplicaSetsList = new ArrayList<>();
- if (dataPartition != null) {
- Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>> seriesSlotMap =
- dataPartition.getDataPartitionMap().get(dbName);
- if (seriesSlotMap == null) {
- throw new SemanticException(
- String.format("Given queried database: %s is not exist!", dbName));
- }
- Map<Integer, List<TRegionReplicaSet>> cachedSeriesSlotWithRegions = new HashMap<>();
- for (DeviceEntry deviceEntry : node.getDeviceEntries()) {
- List<TRegionReplicaSet> regionReplicaSets =
- getDeviceReplicaSets(
- dataPartition,
- seriesSlotMap,
- deviceEntry.getDeviceID(),
- node.getTimeFilter(),
- cachedSeriesSlotWithRegions);
- if (regionReplicaSets.size() > 1) {
- needSplit = true;
- context.deviceCrossRegion = true;
- queryContext.setNeedUpdateScanNumForLastQuery(node.mayUseLastCache());
- }
- regionReplicaSetsList.add(regionReplicaSets);
- }
- }
-
- if (regionReplicaSetsList.isEmpty()) {
- regionReplicaSetsList = Collections.singletonList(Collections.singletonList(NOT_ASSIGNED));
- }
+ AggregationDistributionInfo distributionInfo =
+ prepareAggregationDistribution(node, dbName, dataPartition, context);
Map<TRegionReplicaSet, AggregationTableScanNode> regionNodeMap = new HashMap<>();
- // Step is SINGLE and device data in more than one region, we need to final aggregate the result
- // from different region here, so split
- // this node into two-stage
- needSplit = needSplit && node.getStep() == SINGLE;
- AggregationNode finalAggregation = null;
- if (needSplit) {
- Pair<AggregationNode, AggregationTableScanNode> splitResult =
- split(node, symbolAllocator, queryId);
- finalAggregation = splitResult.left;
- AggregationTableScanNode partialAggregation = splitResult.right;
-
- // cover case: complete push-down + group by + streamable
- if (!context.hasSortProperty && finalAggregation.isStreamable()) {
- OrderingScheme expectedOrderingSchema =
- constructOrderingSchema(node.getPreGroupedSymbols());
- context.setExpectedOrderingScheme(expectedOrderingSchema);
- }
-
- buildRegionNodeMap(node, regionReplicaSetsList, regionNodeMap, partialAggregation);
Review Comment:
need to update `buildRegionNodeMap` method, there is no need to do `originalAggTableScanNode instanceof AggregationTreeDeviceViewScanNode` judgement in that method.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java:
##########
@@ -2852,6 +2862,70 @@ public Operator visitAggregationTreeDeviceViewScan(
return treeAlignedDeviceViewAggregationScanOperator;
}
+ @Override
+ public Operator visitNonAlignedAggregationTreeDeviceViewScan(
+ NonAlignedAggregationTreeDeviceViewScanNode node, LocalExecutionPlanContext context) {
+ QualifiedObjectName qualifiedObjectName = node.getQualifiedObjectName();
+ TsTable tsTable =
+ DataNodeTableCache.getInstance()
+ .getTable(qualifiedObjectName.getDatabaseName(), qualifiedObjectName.getObjectName());
+ IDeviceID.TreeDeviceIdColumnValueExtractor idColumnValueExtractor =
+ createTreeDeviceIdColumnValueExtractor(DataNodeTreeViewSchemaUtils.getPrefixPath(tsTable));
+
+ AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter parameter =
+ constructAbstractAggTableScanOperatorParameter(
+ node,
+ context,
+ TreeAlignedDeviceViewAggregationScanOperator.class.getSimpleName(),
+ node.getMeasurementColumnNameMap(),
+ tsTable.getCachedTableTTL());
+
+ // construct source operator (generator)
+ TreeNonAlignedDeviceViewScanNode scanNode =
+ new TreeNonAlignedDeviceViewScanNode(
+ node.getPlanNodeId(),
+ node.getQualifiedObjectName(),
+ // the outputSymbols of AggTableScanNode is not equals with TableScanNode
Review Comment:
```suggestion
// the outputSymbols of NonAlignedAggregationTreeDeviceViewScanNode is not equals with TreeNonAlignedDeviceViewScanNode
```
better to use specific sub class name instead of super class name
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java:
##########
@@ -2852,6 +2862,70 @@ public Operator visitAggregationTreeDeviceViewScan(
return treeAlignedDeviceViewAggregationScanOperator;
}
+ @Override
+ public Operator visitNonAlignedAggregationTreeDeviceViewScan(
+ NonAlignedAggregationTreeDeviceViewScanNode node, LocalExecutionPlanContext context) {
+ QualifiedObjectName qualifiedObjectName = node.getQualifiedObjectName();
+ TsTable tsTable =
+ DataNodeTableCache.getInstance()
+ .getTable(qualifiedObjectName.getDatabaseName(), qualifiedObjectName.getObjectName());
+ IDeviceID.TreeDeviceIdColumnValueExtractor idColumnValueExtractor =
+ createTreeDeviceIdColumnValueExtractor(DataNodeTreeViewSchemaUtils.getPrefixPath(tsTable));
+
+ AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter parameter =
+ constructAbstractAggTableScanOperatorParameter(
+ node,
+ context,
+ TreeAlignedDeviceViewAggregationScanOperator.class.getSimpleName(),
+ node.getMeasurementColumnNameMap(),
+ tsTable.getCachedTableTTL());
+
+ // construct source operator (generator)
+ TreeNonAlignedDeviceViewScanNode scanNode =
+ new TreeNonAlignedDeviceViewScanNode(
+ node.getPlanNodeId(),
+ node.getQualifiedObjectName(),
+ // the outputSymbols of AggTableScanNode is not equals with TableScanNode
+ parameter.getOutputSymbols(),
Review Comment:
only include field is enough?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support aggregation push-down optimize for table view of non aligned series [iotdb]
Posted by "Wei-hao-Li (via GitHub)" <gi...@apache.org>.
Wei-hao-Li commented on code in PR #17294:
URL: https://github.com/apache/iotdb/pull/17294#discussion_r2940224107
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java:
##########
@@ -2852,6 +2862,70 @@ public Operator visitAggregationTreeDeviceViewScan(
return treeAlignedDeviceViewAggregationScanOperator;
}
+ @Override
+ public Operator visitNonAlignedAggregationTreeDeviceViewScan(
+ NonAlignedAggregationTreeDeviceViewScanNode node, LocalExecutionPlanContext context) {
+ QualifiedObjectName qualifiedObjectName = node.getQualifiedObjectName();
+ TsTable tsTable =
+ DataNodeTableCache.getInstance()
+ .getTable(qualifiedObjectName.getDatabaseName(), qualifiedObjectName.getObjectName());
+ IDeviceID.TreeDeviceIdColumnValueExtractor idColumnValueExtractor =
+ createTreeDeviceIdColumnValueExtractor(DataNodeTreeViewSchemaUtils.getPrefixPath(tsTable));
+
+ AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter parameter =
+ constructAbstractAggTableScanOperatorParameter(
+ node,
+ context,
+ TreeAlignedDeviceViewAggregationScanOperator.class.getSimpleName(),
+ node.getMeasurementColumnNameMap(),
+ tsTable.getCachedTableTTL());
+
+ // construct source operator (generator)
+ TreeNonAlignedDeviceViewScanNode scanNode =
+ new TreeNonAlignedDeviceViewScanNode(
+ node.getPlanNodeId(),
+ node.getQualifiedObjectName(),
+ // the outputSymbols of AggTableScanNode is not equals with TableScanNode
+ parameter.getOutputSymbols(),
Review Comment:
Yes, the tree operator will always return `time` column. And the `tag` and `attribute` column we can build use deviceEntry in TreeNonAlignedDeviceViewAggregationScanOperator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org