This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch fix-external-tsfile-grouping-pushdown in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4638d60989bcfc0edef84d3697cf968da65f5db5 Author: shuwenwei <[email protected]> AuthorDate: Tue Jul 14 18:19:12 2026 +0800 Fix external TsFile grouping optimization --- .../plan/planner/plan/node/PlanVisitor.java | 2 +- .../optimizations/ParallelizeGroupingTest.java | 112 +++++++++++++++++++++ 2 files changed, 113 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java index 81596f2d38c..a5b886ff4b0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java @@ -652,7 +652,7 @@ public interface PlanVisitor<R, C> extends ICoreQueryPlanVisitor<R, C> { } default R visitExternalTsFileScan(ExternalTsFileScanNode node, C context) { - return visitTableScan(node, context); + return visitDeviceTableScan(node, context); } default R visitExternalTsFileAggregationScan(ExternalTsFileAggregationScanNode node, C context) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGroupingTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGroupingTest.java new file mode 100644 index 00000000000..5ac1858e384 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGroupingTest.java @@ -0,0 +1,112 @@ +/* + * 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.plan.relational.planner.optimizations; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.OrderingScheme; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.GroupNode; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.execution.warnings.WarningCollector; +import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis; +import org.apache.iotdb.db.queryengine.plan.relational.execution.querystats.PlanOptimizersStatsCollector; +import org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile.ExternalTsFileQueryResource; +import org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile.ReadTsFileTableFunction; +import org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExternalTsFileScanNode; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.junit.Test; + +import java.nio.file.Path; +import java.util.Collections; +import java.util.Map; + +import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TAG; +import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestUtils.SESSION_INFO; +import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestUtils.TEST_MATADATA; +import static org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder.ASC_NULLS_LAST; +import static org.apache.tsfile.read.common.type.StringType.STRING; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class ParallelizeGroupingTest { + + @Test + public void testExternalTsFileScanKeepsGroupNodeWhenPartitionedByAllTags() { + Symbol region = new Symbol("region"); + Symbol site = new Symbol("site"); + Map<Symbol, ColumnSchema> assignments = + ImmutableMap.of( + region, new ColumnSchema("region", STRING, false, TAG), + site, new ColumnSchema("site", STRING, false, TAG)); + + MPPQueryContext queryContext = new MPPQueryContext(new QueryId("parallelize_grouping")); + ExternalTsFileQueryResource resource = + new ExternalTsFileQueryResource( + queryContext, + Path.of("target", "parallelize-grouping"), + "metrics", + Collections.emptyList(), + assignments, + ReadTsFileTableFunction.DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES); + try { + ExternalTsFileScanNode scanNode = + new ExternalTsFileScanNode( + new PlanNodeId("scan"), + QualifiedObjectName.valueOf("external.metrics"), + ImmutableList.of(region, site), + assignments, + ImmutableMap.of(region, 0, site, 1), + resource); + OrderingScheme orderingScheme = + new OrderingScheme( + ImmutableList.of(region, site), + ImmutableMap.of(region, ASC_NULLS_LAST, site, ASC_NULLS_LAST)); + GroupNode groupNode = new GroupNode(new PlanNodeId("group"), scanNode, orderingScheme, 2); + + Analysis analysis = new Analysis(null, Collections.emptyMap()); + analysis.setQuery(true); + PlanNode optimizedPlan = + new ParallelizeGrouping() + .optimize( + groupNode, + new PlanOptimizer.Context( + SESSION_INFO, + analysis, + TEST_MATADATA, + queryContext, + new SymbolAllocator(), + new QueryId("parallelize_grouping"), + WarningCollector.NOOP, + PlanOptimizersStatsCollector.createPlanOptimizersStatsCollector())); + + assertTrue(optimizedPlan instanceof GroupNode); + assertSame(scanNode, ((GroupNode) optimizedPlan).getChild()); + } finally { + resource.closeByQueryExecution(); + } + } +}
