From b127a215305094b4cabbb2163ba343ff18aaeeb5 Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Tue, 4 Nov 2025 19:15:44 +0800 Subject: [PATCH 01/10] feat(sql): show timeseries order by timeseries --- ...oTDBShowTimeseriesOrderByTimeseriesIT.java | 154 ++++++++++++++++++ .../apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 9 +- .../queryengine/plan/parser/ASTVisitor.java | 16 ++ .../plan/planner/LogicalPlanVisitor.java | 11 +- .../metadata/ShowTimeSeriesStatement.java | 19 +++ 5 files changed, 206 insertions(+), 3 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java new file mode 100644 index 0000000000000..9a5f12c97d8c5 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java @@ -0,0 +1,154 @@ +/* + * 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.it.schema; + +import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; +import org.apache.iotdb.util.AbstractSchemaIT; + +import org.junit.After; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runners.Parameterized; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +@Category({LocalStandaloneIT.class, ClusterIT.class}) +public class IoTDBShowTimeseriesOrderByTimeseriesIT extends AbstractSchemaIT { + + public IoTDBShowTimeseriesOrderByTimeseriesIT(SchemaTestMode schemaTestMode) { + super(schemaTestMode); + } + + @Parameterized.BeforeParam + public static void before() throws Exception { + setUpEnvironment(); + EnvFactory.getEnv().initClusterEnvironment(); + } + + @Parameterized.AfterParam + public static void after() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + tearDownEnvironment(); + } + + @After + public void tearDown() throws Exception { + clearSchema(); + } + + private void prepareSimpleSchema() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE root.ln"); + statement.execute("CREATE DATABASE root.sg"); + statement.execute( + "create timeseries root.ln.d0.s0 with datatype=INT32, encoding=RLE, compression=SNAPPY"); + statement.execute( + "create timeseries root.sg.d0.s2 with datatype=INT32, encoding=RLE, compression=SNAPPY"); + statement.execute( + "create timeseries root.sg.d0.s1 with datatype=INT32, encoding=RLE, compression=SNAPPY"); + } + } + + @Test + public void testOrderByTimeseriesAsc() throws Exception { + prepareSimpleSchema(); + + List expected = + new ArrayList<>(Arrays.asList("root.ln.d0.s0", "root.sg.d0.s1", "root.sg.d0.s2")); + + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery("show timeseries order by timeseries")) { + List actual = new ArrayList<>(); + while (resultSet.next()) { + actual.add(resultSet.getString(ColumnHeaderConstant.TIMESERIES)); + } + assertEquals(expected, actual); + } + } + + @Test + public void testOrderByTimeseriesDescWithLimit() throws Exception { + prepareSimpleSchema(); + + List all = + new ArrayList<>(Arrays.asList("root.ln.d0.s0", "root.sg.d0.s1", "root.sg.d0.s2")); + Collections.sort(all); + Collections.reverse(all); + List expectedTop2 = all.subList(0, 2); + + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement(); + ResultSet resultSet = + statement.executeQuery("show timeseries order by timeseries desc limit 2")) { + List actual = new ArrayList<>(); + while (resultSet.next()) { + actual.add(resultSet.getString(ColumnHeaderConstant.TIMESERIES)); + } + assertEquals(expectedTop2, actual); + } + } + + @Test + public void testConflictWithLatest() throws Exception { + prepareSimpleSchema(); + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + try (ResultSet ignored = + statement.executeQuery("show latest timeseries order by timeseries")) { + fail("Expected exception for conflict between LATEST and ORDER BY TIMESERIES"); + } catch (SQLException e) { + assertTrue( + e.getMessage().toLowerCase().contains("latest") + && e.getMessage().toLowerCase().contains("order by timeseries")); + } + } + } + + @Test + public void testConflictWithTimeCondition() throws Exception { + prepareSimpleSchema(); + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + try (ResultSet ignored = + statement.executeQuery("show timeseries where time > 0 order by timeseries")) { + fail("Expected exception for conflict between TIME condition and ORDER BY TIMESERIES"); + } catch (SQLException e) { + assertTrue( + e.getMessage().toLowerCase().contains("time condition") + && e.getMessage().toLowerCase().contains("order by timeseries")); + } + } + } +} diff --git a/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 b/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 index efe661e05430c..2308d3b81b7d2 100644 --- a/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 +++ b/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 @@ -200,7 +200,12 @@ showDevices // ---- Show Timeseries showTimeseries - : SHOW LATEST? TIMESERIES prefixPath? timeseriesWhereClause? timeConditionClause? rowPaginationClause? + : SHOW LATEST? TIMESERIES prefixPath? timeseriesWhereClause? timeConditionClause? orderByTimeseriesClause? rowPaginationClause? + ; + +// order by timeseries for SHOW TIMESERIES +orderByTimeseriesClause + : ORDER BY TIMESERIES (ASC | DESC)? ; // ---- Show Child Paths @@ -1586,4 +1591,4 @@ subStringExpression signedIntegerLiteral : (PLUS|MINUS)?INTEGER_LITERAL - ; \ No newline at end of file + ; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java index 70325257b085e..4c477ae28768f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java @@ -789,6 +789,22 @@ public Statement visitShowTimeseries(IoTDBSqlParser.ShowTimeseriesContext ctx) { showTimeSeriesStatement.setTimeCondition( parseWhereClause(ctx.timeConditionClause().whereClause())); } + + // ORDER BY TIMESERIES [ASC|DESC] + if (ctx.orderByTimeseriesClause() != null) { + if (orderByHeat) { + throw new SemanticException( + "LATEST and ORDER BY TIMESERIES cannot be used at the same time."); + } + if (ctx.timeConditionClause() != null) { + throw new SemanticException("ORDER BY TIMESERIES does not support TIME condition."); + } + Ordering ordering = Ordering.ASC; + if (ctx.orderByTimeseriesClause().DESC() != null) { + ordering = Ordering.DESC; + } + showTimeSeriesStatement.setOrderByTimeseries(true, ordering); + } if (ctx.rowPaginationClause() != null) { if (ctx.rowPaginationClause().limitClause() != null) { showTimeSeriesStatement.setLimit(parseLimitClause(ctx.rowPaginationClause().limitClause())); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java index 2274869c35543..f434286168b63 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java @@ -54,6 +54,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.AggregationStep; import org.apache.iotdb.db.queryengine.plan.statement.StatementNode; import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor; +import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem; import org.apache.iotdb.db.queryengine.plan.statement.crud.DeleteDataStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertMultiTabletsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; @@ -572,7 +573,8 @@ public PlanNode visitShowTimeSeries( boolean canPushDownOffsetLimit = analysis.getSchemaPartitionInfo() != null && analysis.getSchemaPartitionInfo().getDistributionInfo().size() == 1 - && !showTimeSeriesStatement.isOrderByHeat(); + && !showTimeSeriesStatement.isOrderByHeat() + && !showTimeSeriesStatement.isOrderByTimeseries(); if (showTimeSeriesStatement.isOrderByHeat()) { limit = 0; @@ -594,6 +596,13 @@ public PlanNode visitShowTimeSeries( showTimeSeriesStatement.getAuthorityScope()) .planSchemaQueryMerge(showTimeSeriesStatement.isOrderByHeat()); + // order by timeseries name + if (showTimeSeriesStatement.isOrderByTimeseries()) { + SortItem sortItem = + new SortItem(ColumnHeaderConstant.TIMESERIES, showTimeSeriesStatement.getNameOrdering()); + planBuilder = planBuilder.planOrderBy(java.util.Collections.singletonList(sortItem)); + } + // show latest timeseries if (showTimeSeriesStatement.isOrderByHeat() && null != analysis.getDataPartitionInfo() diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowTimeSeriesStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowTimeSeriesStatement.java index 82a389d622499..bb07ca6a141ef 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowTimeSeriesStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowTimeSeriesStatement.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.schema.filter.SchemaFilter; import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.queryengine.plan.statement.component.WhereCondition; import java.util.Collections; @@ -42,6 +43,9 @@ public class ShowTimeSeriesStatement extends ShowStatement { // if is true, the result will be sorted according to the inserting frequency of the time series private final boolean orderByHeat; private WhereCondition timeCondition; + // order by timeseries name + private boolean orderByTimeseries; + private Ordering nameOrdering = Ordering.ASC; public ShowTimeSeriesStatement(PartialPath pathPattern, boolean orderByHeat) { super(); @@ -65,6 +69,21 @@ public boolean isOrderByHeat() { return orderByHeat; } + public boolean isOrderByTimeseries() { + return orderByTimeseries; + } + + public Ordering getNameOrdering() { + return nameOrdering; + } + + public void setOrderByTimeseries(boolean orderByTimeseries, Ordering ordering) { + this.orderByTimeseries = orderByTimeseries; + if (ordering != null) { + this.nameOrdering = ordering; + } + } + public void setTimeCondition(WhereCondition timeCondition) { this.timeCondition = timeCondition; } From d4fd695deba771dcf7f50dc5b06a2eb76ac86457 Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Tue, 4 Nov 2025 20:48:37 +0800 Subject: [PATCH 02/10] feat(Temp): show timeseries order by timeseries --- .../db/queryengine/plan/planner/LogicalPlanVisitor.java | 9 +++++++++ .../distribution/SimpleFragmentParallelPlanner.java | 3 ++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java index f434286168b63..73aad8e407773 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java @@ -557,6 +557,15 @@ public PlanNode visitShowTimeSeries( ShowTimeSeriesStatement showTimeSeriesStatement, MPPQueryContext context) { LogicalPlanBuilder planBuilder = new LogicalPlanBuilder(analysis, context); + // Ensure TypeProvider has schema query column types for later SortNode + org.apache.iotdb.commons.schema.column.ColumnHeaderConstant + .showTimeSeriesColumnHeaders.forEach( + columnHeader -> + context + .getTypeProvider() + .setTreeModelType( + columnHeader.getColumnName(), columnHeader.getColumnType())); + long limit = showTimeSeriesStatement.getLimit(); long offset = showTimeSeriesStatement.getOffset(); if (showTimeSeriesStatement.hasTimeCondition()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SimpleFragmentParallelPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SimpleFragmentParallelPlanner.java index 334d1973f5345..aaf5549eb4a0b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SimpleFragmentParallelPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SimpleFragmentParallelPlanner.java @@ -157,7 +157,8 @@ private void produceFragmentInstance(PlanFragment fragment) { || analysis.getTreeStatement() instanceof ExplainAnalyzeStatement || analysis.getTreeStatement() instanceof ShowQueriesStatement || (analysis.getTreeStatement() instanceof ShowTimeSeriesStatement - && ((ShowTimeSeriesStatement) analysis.getTreeStatement()).isOrderByHeat())) { + && (((ShowTimeSeriesStatement) analysis.getTreeStatement()).isOrderByHeat() + || ((ShowTimeSeriesStatement) analysis.getTreeStatement()).isOrderByTimeseries()))) { fragmentInstance.getFragment().generateTypeProvider(queryContext.getTypeProvider()); } instanceMap.putIfAbsent(fragment.getId(), fragmentInstance); From 94e783d7314ec6f5de13eec1cc77f971118d112c Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Fri, 28 Nov 2025 10:03:48 +0800 Subject: [PATCH 03/10] feat: show timeseries order by timeseries (pushDown for single region) --- ...oTDBShowTimeseriesOrderByTimeseriesIT.java | 6 +- .../source/LogicalViewSchemaSource.java | 4 +- .../schema/source/SchemaSourceFactory.java | 17 +++++- .../schema/source/TimeSeriesSchemaSource.java | 12 +++- .../plan/planner/LogicalPlanBuilder.java | 8 ++- .../plan/planner/LogicalPlanVisitor.java | 32 ++++++----- .../plan/planner/OperatorTreeGenerator.java | 4 +- .../SimpleFragmentParallelPlanner.java | 3 +- .../read/TimeSeriesSchemaScanNode.java | 57 +++++++++++++++++-- .../impl/mem/MTreeBelowSGMemoryImpl.java | 30 ++++++++++ .../read/req/IShowTimeSeriesPlan.java | 6 ++ .../read/req/SchemaRegionReadPlanFactory.java | 15 ++++- .../read/req/impl/ShowTimeSeriesPlanImpl.java | 27 ++++++++- .../schemaRegion/SchemaRegionTestUtil.java | 37 ++++++++++-- .../schema/SchemaQueryScanOperatorTest.java | 4 +- 15 files changed, 221 insertions(+), 41 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java index 9a5f12c97d8c5..1bc9e838bc716 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java @@ -84,12 +84,12 @@ private void prepareSimpleSchema() throws Exception { public void testOrderByTimeseriesAsc() throws Exception { prepareSimpleSchema(); - List expected = - new ArrayList<>(Arrays.asList("root.ln.d0.s0", "root.sg.d0.s1", "root.sg.d0.s2")); + List expected = new ArrayList<>(Arrays.asList("root.ln.d0.s0", "root.sg.d0.s1")); try (Connection connection = EnvFactory.getEnv().getConnection(); Statement statement = connection.createStatement(); - ResultSet resultSet = statement.executeQuery("show timeseries order by timeseries")) { + ResultSet resultSet = + statement.executeQuery("show timeseries order by timeseries offset 2 limit 2")) { List actual = new ArrayList<>(); while (resultSet.next()) { actual.add(resultSet.getString(ColumnHeaderConstant.TIMESERIES)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/LogicalViewSchemaSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/LogicalViewSchemaSource.java index f1eaaebbd1ddc..cd5dd52e8046d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/LogicalViewSchemaSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/LogicalViewSchemaSource.java @@ -79,7 +79,9 @@ public ISchemaReader getSchemaReader(ISchemaRegion schema SchemaFilterFactory.and( schemaFilter, SchemaFilterFactory.createViewTypeFilter(ViewType.VIEW)), true, - scope)); + scope, + false, + false)); } catch (MetadataException e) { throw new SchemaExecutionException(e.getMessage(), e); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/SchemaSourceFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/SchemaSourceFactory.java index 2ef0ab9e18a78..4cb22e3578200 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/SchemaSourceFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/SchemaSourceFactory.java @@ -47,7 +47,7 @@ public static ISchemaSource getTimeSeriesSchemaCountSourc Map templateMap, PathPatternTree scope) { return new TimeSeriesSchemaSource( - pathPattern, isPrefixMatch, 0, 0, schemaFilter, templateMap, false, scope); + pathPattern, isPrefixMatch, 0, 0, schemaFilter, templateMap, false, scope, false, false); } // show time series @@ -58,9 +58,20 @@ public static ISchemaSource getTimeSeriesSchemaScanSource long offset, SchemaFilter schemaFilter, Map templateMap, - PathPatternTree scope) { + PathPatternTree scope, + boolean orderByTimeseries, + boolean orderByTimeseriesDesc) { return new TimeSeriesSchemaSource( - pathPattern, isPrefixMatch, limit, offset, schemaFilter, templateMap, true, scope); + pathPattern, + isPrefixMatch, + limit, + offset, + schemaFilter, + templateMap, + true, + scope, + orderByTimeseries, + orderByTimeseriesDesc); } // count device diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java index 40799c548eeba..59aaa0a1964e3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/schema/source/TimeSeriesSchemaSource.java @@ -54,6 +54,8 @@ public class TimeSeriesSchemaSource implements ISchemaSource templateMap; private final boolean needViewDetail; + private final boolean orderByTimeseries; + private final boolean orderByTimeseriesDesc; TimeSeriesSchemaSource( PartialPath pathPattern, @@ -63,7 +65,9 @@ public class TimeSeriesSchemaSource implements ISchemaSource templateMap, boolean needViewDetail, - PathPatternTree scope) { + PathPatternTree scope, + boolean orderByTimeseries, + boolean orderByTimeseriesDesc) { this.pathPattern = pathPattern; this.isPrefixMatch = isPrefixMatch; this.limit = limit; @@ -72,6 +76,8 @@ public class TimeSeriesSchemaSource implements ISchemaSource getSchemaReader(ISchemaRegion schema isPrefixMatch, schemaFilter, needViewDetail, - scope)); + scope, + orderByTimeseries, + orderByTimeseriesDesc)); } catch (MetadataException e) { throw new SchemaExecutionException(e.getMessage(), e); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java index 5414b5ac02003..5b0109c09e47e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java @@ -995,7 +995,9 @@ public LogicalPlanBuilder planTimeSeriesSchemaSource( boolean orderByHeat, boolean prefixPath, Map templateMap, - PathPatternTree scope) { + PathPatternTree scope, + boolean orderByTimeseries, + boolean orderByTimeseriesDesc) { this.root = new TimeSeriesSchemaScanNode( context.getQueryId().genPlanNodeId(), @@ -1006,7 +1008,9 @@ public LogicalPlanBuilder planTimeSeriesSchemaSource( orderByHeat, prefixPath, templateMap, - scope); + scope, + orderByTimeseries, + orderByTimeseriesDesc); return this; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java index 73aad8e407773..4cc3567ad572c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java @@ -54,6 +54,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.AggregationStep; import org.apache.iotdb.db.queryengine.plan.statement.StatementNode; import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem; import org.apache.iotdb.db.queryengine.plan.statement.crud.DeleteDataStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertMultiTabletsStatement; @@ -558,13 +559,11 @@ public PlanNode visitShowTimeSeries( LogicalPlanBuilder planBuilder = new LogicalPlanBuilder(analysis, context); // Ensure TypeProvider has schema query column types for later SortNode - org.apache.iotdb.commons.schema.column.ColumnHeaderConstant - .showTimeSeriesColumnHeaders.forEach( - columnHeader -> - context - .getTypeProvider() - .setTreeModelType( - columnHeader.getColumnName(), columnHeader.getColumnType())); + ColumnHeaderConstant.showTimeSeriesColumnHeaders.forEach( + columnHeader -> + context + .getTypeProvider() + .setTreeModelType(columnHeader.getColumnName(), columnHeader.getColumnType())); long limit = showTimeSeriesStatement.getLimit(); long offset = showTimeSeriesStatement.getOffset(); @@ -579,11 +578,10 @@ public PlanNode visitShowTimeSeries( // If there is only one region, we can push down the offset and limit operation to // source operator. - boolean canPushDownOffsetLimit = + boolean singleSchemaRegion = analysis.getSchemaPartitionInfo() != null - && analysis.getSchemaPartitionInfo().getDistributionInfo().size() == 1 - && !showTimeSeriesStatement.isOrderByHeat() - && !showTimeSeriesStatement.isOrderByTimeseries(); + && analysis.getSchemaPartitionInfo().getDistributionInfo().size() == 1; + boolean canPushDownOffsetLimit = singleSchemaRegion && !showTimeSeriesStatement.isOrderByHeat(); if (showTimeSeriesStatement.isOrderByHeat()) { limit = 0; @@ -592,6 +590,10 @@ public PlanNode visitShowTimeSeries( limit = showTimeSeriesStatement.getLimit() + showTimeSeriesStatement.getOffset(); offset = 0; } + boolean orderByTimeseries = showTimeSeriesStatement.isOrderByTimeseries() && singleSchemaRegion; + boolean orderByTimeseriesDesc = + orderByTimeseries && showTimeSeriesStatement.getNameOrdering() == Ordering.DESC; + planBuilder = planBuilder .planTimeSeriesSchemaSource( @@ -602,11 +604,13 @@ public PlanNode visitShowTimeSeries( showTimeSeriesStatement.isOrderByHeat(), showTimeSeriesStatement.isPrefixPath(), analysis.getRelatedTemplateInfo(), - showTimeSeriesStatement.getAuthorityScope()) + showTimeSeriesStatement.getAuthorityScope(), + orderByTimeseries, + orderByTimeseriesDesc) .planSchemaQueryMerge(showTimeSeriesStatement.isOrderByHeat()); - // order by timeseries name - if (showTimeSeriesStatement.isOrderByTimeseries()) { + // order by timeseries name in multi-region case: still need global SortNode + if (showTimeSeriesStatement.isOrderByTimeseries() && !singleSchemaRegion) { SortItem sortItem = new SortItem(ColumnHeaderConstant.TIMESERIES, showTimeSeriesStatement.getNameOrdering()); planBuilder = planBuilder.planOrderBy(java.util.Collections.singletonList(sortItem)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java index f869ce59d470f..261cad23a2c0f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java @@ -913,7 +913,9 @@ public Operator visitTimeSeriesSchemaScan( node.getOffset(), node.getSchemaFilter(), node.getTemplateMap(), - node.getScope())); + node.getScope(), + node.isOrderByTimeseries(), + node.isOrderByTimeseriesDesc())); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SimpleFragmentParallelPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SimpleFragmentParallelPlanner.java index aaf5549eb4a0b..25f51f88d9eae 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SimpleFragmentParallelPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SimpleFragmentParallelPlanner.java @@ -158,7 +158,8 @@ private void produceFragmentInstance(PlanFragment fragment) { || analysis.getTreeStatement() instanceof ShowQueriesStatement || (analysis.getTreeStatement() instanceof ShowTimeSeriesStatement && (((ShowTimeSeriesStatement) analysis.getTreeStatement()).isOrderByHeat() - || ((ShowTimeSeriesStatement) analysis.getTreeStatement()).isOrderByTimeseries()))) { + || ((ShowTimeSeriesStatement) analysis.getTreeStatement()) + .isOrderByTimeseries()))) { fragmentInstance.getFragment().generateTypeProvider(queryContext.getTypeProvider()); } instanceMap.putIfAbsent(fragment.getId(), fragmentInstance); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/read/TimeSeriesSchemaScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/read/TimeSeriesSchemaScanNode.java index 1997a3caefe84..ddf36b9bb78c6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/read/TimeSeriesSchemaScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/read/TimeSeriesSchemaScanNode.java @@ -48,6 +48,11 @@ public class TimeSeriesSchemaScanNode extends SchemaQueryScanNode { // if is true, the result will be sorted according to the inserting frequency of the timeseries private final boolean orderByHeat; + // Whether to order result by timeseries full path in this region. + private final boolean orderByTimeseries; + // When orderByTimeseries is true, whether the ordering is descending. + private final boolean orderByTimeseriesDesc; + private final SchemaFilter schemaFilter; private final Map templateMap; @@ -66,6 +71,28 @@ public TimeSeriesSchemaScanNode( this.schemaFilter = schemaFilter; this.orderByHeat = orderByHeat; this.templateMap = templateMap; + this.orderByTimeseries = false; + this.orderByTimeseriesDesc = false; + } + + public TimeSeriesSchemaScanNode( + PlanNodeId id, + PartialPath partialPath, + SchemaFilter schemaFilter, + long limit, + long offset, + boolean orderByHeat, + boolean isPrefixPath, + @NotNull Map templateMap, + @NotNull PathPatternTree scope, + boolean orderByTimeseries, + boolean orderByTimeseriesDesc) { + super(id, partialPath, limit, offset, isPrefixPath, scope); + this.schemaFilter = schemaFilter; + this.orderByHeat = orderByHeat; + this.templateMap = templateMap; + this.orderByTimeseries = orderByTimeseries; + this.orderByTimeseriesDesc = orderByTimeseriesDesc; } public SchemaFilter getSchemaFilter() { @@ -82,6 +109,8 @@ protected void serializeAttributes(ByteBuffer byteBuffer) { ReadWriteIOUtils.write(offset, byteBuffer); ReadWriteIOUtils.write(orderByHeat, byteBuffer); ReadWriteIOUtils.write(isPrefixPath, byteBuffer); + ReadWriteIOUtils.write(orderByTimeseries, byteBuffer); + ReadWriteIOUtils.write(orderByTimeseriesDesc, byteBuffer); ReadWriteIOUtils.write(templateMap.size(), byteBuffer); for (Template template : templateMap.values()) { @@ -99,6 +128,8 @@ protected void serializeAttributes(DataOutputStream stream) throws IOException { ReadWriteIOUtils.write(offset, stream); ReadWriteIOUtils.write(orderByHeat, stream); ReadWriteIOUtils.write(isPrefixPath, stream); + ReadWriteIOUtils.write(orderByTimeseries, stream); + ReadWriteIOUtils.write(orderByTimeseriesDesc, stream); ReadWriteIOUtils.write(templateMap.size(), stream); for (Template template : templateMap.values()) { @@ -120,6 +151,8 @@ public static TimeSeriesSchemaScanNode deserialize(ByteBuffer byteBuffer) { long offset = ReadWriteIOUtils.readLong(byteBuffer); boolean oderByHeat = ReadWriteIOUtils.readBool(byteBuffer); boolean isPrefixPath = ReadWriteIOUtils.readBool(byteBuffer); + boolean orderByTimeseries = ReadWriteIOUtils.readBool(byteBuffer); + boolean orderByTimeseriesDesc = ReadWriteIOUtils.readBool(byteBuffer); int templateNum = ReadWriteIOUtils.readInt(byteBuffer); Map templateMap = new HashMap<>(); @@ -141,13 +174,23 @@ public static TimeSeriesSchemaScanNode deserialize(ByteBuffer byteBuffer) { oderByHeat, isPrefixPath, templateMap, - scope); + scope, + orderByTimeseries, + orderByTimeseriesDesc); } public boolean isOrderByHeat() { return orderByHeat; } + public boolean isOrderByTimeseries() { + return orderByTimeseries; + } + + public boolean isOrderByTimeseriesDesc() { + return orderByTimeseriesDesc; + } + public Map getTemplateMap() { return templateMap; } @@ -168,7 +211,9 @@ public PlanNode clone() { orderByHeat, isPrefixPath, templateMap, - scope); + scope, + orderByTimeseries, + orderByTimeseriesDesc); } @Override @@ -190,12 +235,16 @@ public boolean equals(Object o) { return false; } TimeSeriesSchemaScanNode that = (TimeSeriesSchemaScanNode) o; - return orderByHeat == that.orderByHeat && Objects.equals(schemaFilter, that.schemaFilter); + return orderByHeat == that.orderByHeat + && orderByTimeseries == that.orderByTimeseries + && orderByTimeseriesDesc == that.orderByTimeseriesDesc + && Objects.equals(schemaFilter, that.schemaFilter); } @Override public int hashCode() { - return Objects.hash(super.hashCode(), schemaFilter, orderByHeat); + return Objects.hash( + super.hashCode(), schemaFilter, orderByHeat, orderByTimeseries, orderByTimeseriesDesc); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java index 9aec147ace1fe..1878f3e54ec7b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java @@ -1495,6 +1495,36 @@ public ITimeSeriesSchemaInfo snapshot() { } }; } + + @Override + protected Iterator getChildrenIterator(final IMemMNode parent) + throws MetadataException { + Iterator baseIterator = super.getChildrenIterator(parent); + + if (!showTimeSeriesPlan.isOrderByTimeseries()) { + return baseIterator; + } + + List children = new ArrayList<>(); + if (baseIterator instanceof IMNodeIterator) { + IMNodeIterator it = (IMNodeIterator) baseIterator; + while (it.hasNext()) { + children.add(it.next()); + } + it.close(); + } else { + while (baseIterator.hasNext()) { + children.add(baseIterator.next()); + } + } + + children.sort( + (a, b) -> { + int cmp = a.getName().compareTo(b.getName()); + return showTimeSeriesPlan.isOrderByTimeseriesDesc() ? -cmp : cmp; + }); + return children.iterator(); + } }; collector.setTemplateMap(showTimeSeriesPlan.getRelatedTemplate(), nodeFactory); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/IShowTimeSeriesPlan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/IShowTimeSeriesPlan.java index e05ee54f253cc..0226e0dca9c7c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/IShowTimeSeriesPlan.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/IShowTimeSeriesPlan.java @@ -32,4 +32,10 @@ public interface IShowTimeSeriesPlan extends IShowSchemaPlan { SchemaFilter getSchemaFilter(); Map getRelatedTemplate(); + + /** Whether to order result by timeseries full path in this region. */ + boolean isOrderByTimeseries(); + + /** Whether the timeseries ordering is descending when {@link #isOrderByTimeseries()} is true. */ + boolean isOrderByTimeseriesDesc(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/SchemaRegionReadPlanFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/SchemaRegionReadPlanFactory.java index 84a2ca9cdf7cc..3707fe6f43edd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/SchemaRegionReadPlanFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/SchemaRegionReadPlanFactory.java @@ -61,9 +61,20 @@ public static IShowTimeSeriesPlan getShowTimeSeriesPlan( boolean isPrefixMatch, SchemaFilter schemaFilter, boolean needViewDetail, - PathPatternTree scope) { + PathPatternTree scope, + boolean orderByTimeseries, + boolean orderByTimeseriesDesc) { return new ShowTimeSeriesPlanImpl( - path, relatedTemplate, limit, offset, isPrefixMatch, schemaFilter, needViewDetail, scope); + path, + relatedTemplate, + limit, + offset, + isPrefixMatch, + schemaFilter, + needViewDetail, + scope, + orderByTimeseries, + orderByTimeseriesDesc); } public static IShowNodesPlan getShowNodesPlan(PartialPath path, PathPatternTree scope) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/ShowTimeSeriesPlanImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/ShowTimeSeriesPlanImpl.java index 8aeb0e837bedd..ea22e22d8d61b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/ShowTimeSeriesPlanImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/ShowTimeSeriesPlanImpl.java @@ -37,6 +37,10 @@ public class ShowTimeSeriesPlanImpl extends AbstractShowSchemaPlanImpl private final SchemaFilter schemaFilter; private final boolean needViewDetail; + // order-by-timeseries pushdown flags inside a single SchemaRegion + private final boolean orderByTimeseries; + private final boolean orderByTimeseriesDesc; + public ShowTimeSeriesPlanImpl( PartialPath path, Map relatedTemplate, @@ -45,11 +49,15 @@ public ShowTimeSeriesPlanImpl( boolean isPrefixMatch, SchemaFilter schemaFilter, boolean needViewDetail, - PathPatternTree scope) { + PathPatternTree scope, + boolean orderByTimeseries, + boolean orderByTimeseriesDesc) { super(path, limit, offset, isPrefixMatch, scope); this.relatedTemplate = relatedTemplate; this.schemaFilter = schemaFilter; this.needViewDetail = needViewDetail; + this.orderByTimeseries = orderByTimeseries; + this.orderByTimeseriesDesc = orderByTimeseriesDesc; } @Override @@ -67,18 +75,31 @@ public Map getRelatedTemplate() { return relatedTemplate; } + @Override + public boolean isOrderByTimeseries() { + return orderByTimeseries; + } + + @Override + public boolean isOrderByTimeseriesDesc() { + return orderByTimeseriesDesc; + } + @Override public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; if (!super.equals(o)) return false; ShowTimeSeriesPlanImpl that = (ShowTimeSeriesPlanImpl) o; - return Objects.equals(relatedTemplate, that.relatedTemplate) + return orderByTimeseries == that.orderByTimeseries + && orderByTimeseriesDesc == that.orderByTimeseriesDesc + && Objects.equals(relatedTemplate, that.relatedTemplate) && Objects.equals(schemaFilter, that.schemaFilter); } @Override public int hashCode() { - return Objects.hash(super.hashCode(), relatedTemplate, schemaFilter); + return Objects.hash( + super.hashCode(), relatedTemplate, schemaFilter, orderByTimeseries, orderByTimeseriesDesc); } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionTestUtil.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionTestUtil.java index 42800c4e58630..45b4c9137881e 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionTestUtil.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaRegionTestUtil.java @@ -175,7 +175,16 @@ public static long getAllTimeSeriesCount( try (ISchemaReader timeSeriesReader = schemaRegion.getTimeSeriesReader( SchemaRegionReadPlanFactory.getShowTimeSeriesPlan( - pathPattern, templateMap, 0, 0, isPrefixMatch, null, false, ALL_MATCH_SCOPE))) { + pathPattern, + templateMap, + 0, + 0, + isPrefixMatch, + null, + false, + ALL_MATCH_SCOPE, + false, + false))) { long count = 0; while (timeSeriesReader.hasNext()) { timeSeriesReader.next(); @@ -200,7 +209,16 @@ public static void checkSingleTimeSeries( try (final ISchemaReader timeSeriesReader = schemaRegion.getTimeSeriesReader( SchemaRegionReadPlanFactory.getShowTimeSeriesPlan( - pathPattern, Collections.emptyMap(), 0, 0, false, null, false, ALL_MATCH_SCOPE))) { + pathPattern, + Collections.emptyMap(), + 0, + 0, + false, + null, + false, + ALL_MATCH_SCOPE, + false, + false))) { Assert.assertTrue(timeSeriesReader.hasNext()); final ITimeSeriesSchemaInfo info = timeSeriesReader.next(); Assert.assertEquals(isAligned, info.isUnderAlignedDevice()); @@ -237,7 +255,16 @@ public static Map getMeasurementCountGroupByLevel( try (ISchemaReader timeSeriesReader = schemaRegion.getTimeSeriesReader( SchemaRegionReadPlanFactory.getShowTimeSeriesPlan( - pathPattern, null, 0, 0, isPrefixMatch, null, false, ALL_MATCH_SCOPE))) { + pathPattern, + null, + 0, + 0, + isPrefixMatch, + null, + false, + ALL_MATCH_SCOPE, + false, + false))) { Map countMap = new HashMap<>(); while (timeSeriesReader.hasNext()) { ITimeSeriesSchemaInfo timeSeriesSchemaInfo = timeSeriesReader.next(); @@ -356,7 +383,9 @@ public static List showTimeseries( isPrefixMatch, schemaFilter, needViewDetail, - ALL_MATCH_SCOPE))) { + ALL_MATCH_SCOPE, + false, + false))) { while (reader.hasNext()) { timeSeriesSchemaInfo = reader.next(); result.add( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaQueryScanOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaQueryScanOperatorTest.java index a5d4a6e6acef6..aab01b785c799 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaQueryScanOperatorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/schema/SchemaQueryScanOperatorTest.java @@ -215,7 +215,9 @@ public void testTimeSeriesSchemaScan() throws Exception { 0, null, Collections.emptyMap(), - SchemaConstant.ALL_MATCH_SCOPE); + SchemaConstant.ALL_MATCH_SCOPE, + false, + false); SchemaOperatorTestUtil.mockGetSchemaReader( timeSeriesSchemaSource, showTimeSeriesResults.iterator(), schemaRegion, true); From 1fff1ba12e322696683978c0b85041b7fc28c4b6 Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Fri, 26 Dec 2025 13:25:24 +0800 Subject: [PATCH 04/10] fix: ShowStatement limit default value should be -1 not 0. --- .../queryengine/plan/planner/LogicalPlanBuilder.java | 2 +- .../queryengine/plan/planner/LogicalPlanVisitor.java | 10 +++++----- .../plan/statement/metadata/ShowStatement.java | 9 ++++++++- 3 files changed, 14 insertions(+), 7 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java index 5b0109c09e47e..e30e325f26b36 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java @@ -905,7 +905,7 @@ public LogicalPlanBuilder planFill(FillDescriptor fillDescriptor, Ordering scanO } public LogicalPlanBuilder planLimit(long rowLimit) { - if (rowLimit == 0) { + if (rowLimit < 0) { return this; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java index 4cc3567ad572c..373039578b0cd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java @@ -584,13 +584,13 @@ public PlanNode visitShowTimeSeries( boolean canPushDownOffsetLimit = singleSchemaRegion && !showTimeSeriesStatement.isOrderByHeat(); if (showTimeSeriesStatement.isOrderByHeat()) { - limit = 0; + limit = -1; offset = 0; } else if (!canPushDownOffsetLimit) { - limit = showTimeSeriesStatement.getLimit() + showTimeSeriesStatement.getOffset(); + limit = showTimeSeriesStatement.getLimitWithOffset(); offset = 0; } - boolean orderByTimeseries = showTimeSeriesStatement.isOrderByTimeseries() && singleSchemaRegion; + boolean orderByTimeseries = showTimeSeriesStatement.isOrderByTimeseries(); boolean orderByTimeseriesDesc = orderByTimeseries && showTimeSeriesStatement.getNameOrdering() == Ordering.DESC; @@ -658,7 +658,7 @@ public PlanNode visitShowDevices( long limit = showDevicesStatement.getLimit(); long offset = showDevicesStatement.getOffset(); if (!canPushDownOffsetLimit) { - limit = showDevicesStatement.getLimit() + showDevicesStatement.getOffset(); + limit = showDevicesStatement.getLimitWithOffset(); offset = 0; } @@ -1030,7 +1030,7 @@ public PlanNode visitShowLogicalView( long limit = showLogicalViewStatement.getLimit(); long offset = showLogicalViewStatement.getOffset(); if (!canPushDownOffsetLimit) { - limit = showLogicalViewStatement.getLimit() + showLogicalViewStatement.getOffset(); + limit = showLogicalViewStatement.getLimitWithOffset(); offset = 0; } planBuilder = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowStatement.java index 971fae6698be6..a283204d80e57 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowStatement.java @@ -28,7 +28,7 @@ public class ShowStatement extends AuthorityInformationStatement { - long limit = 0; + long limit = -1; long offset = 0; protected boolean isPrefixPath; @@ -59,6 +59,13 @@ public void setOffset(long offset) { this.offset = offset; } + public long getLimitWithOffset() { + if (limit < 0) { + return limit; + } + return limit + offset; + } + public boolean isPrefixPath() { return isPrefixPath; } From 66833ce01bd67e4e10cb0ed9e4e3104fdfff7a64 Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Fri, 16 Jan 2026 09:54:32 +0800 Subject: [PATCH 05/10] feat: (show timeseries order by timeseries) limit/offset pushdown to subtree of metadata tree. initial version. # Conflicts: # iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java --- ...oTDBShowTimeseriesOrderByTimeseriesIT.java | 140 +++++++++++++----- .../impl/mem/MTreeBelowSGMemoryImpl.java | 115 ++++++++++++-- .../mtree/impl/mem/mnode/IMemMNode.java | 11 +- .../impl/mem/mnode/basic/BasicMNode.java | 16 +- .../mem/mnode/impl/AboveDatabaseMNode.java | 10 ++ .../impl/mem/mnode/impl/DatabaseMNode.java | 10 ++ .../impl/mem/mnode/impl/MeasurementMNode.java | 10 ++ .../req/impl/AbstractShowSchemaPlanImpl.java | 2 +- .../impl/SchemaReaderLimitOffsetWrapper.java | 2 +- 9 files changed, 260 insertions(+), 56 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java index 1bc9e838bc716..ca53cdb19446c 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java @@ -45,6 +45,19 @@ @Category({LocalStandaloneIT.class, ClusterIT.class}) public class IoTDBShowTimeseriesOrderByTimeseriesIT extends AbstractSchemaIT { + private static final List BASE_TIMESERIES = + Arrays.asList( + "root.db1.devA.m1", + "root.db1.devA.m2", + "root.db1.devB.m1", + "root.db1.devB.x", + "root.db2.devA.m1", + "root.db2.devC.m0", + "root.db2.devC.m3", + "root.db3.z.m1", + "root.db3.z.m10", + "root.db3.z.m2"); + public IoTDBShowTimeseriesOrderByTimeseriesIT(SchemaTestMode schemaTestMode) { super(schemaTestMode); } @@ -66,63 +79,116 @@ public void tearDown() throws Exception { clearSchema(); } - private void prepareSimpleSchema() throws Exception { + private void prepareComplexSchema() throws Exception { try (Connection connection = EnvFactory.getEnv().getConnection(); Statement statement = connection.createStatement()) { - statement.execute("CREATE DATABASE root.ln"); - statement.execute("CREATE DATABASE root.sg"); - statement.execute( - "create timeseries root.ln.d0.s0 with datatype=INT32, encoding=RLE, compression=SNAPPY"); - statement.execute( - "create timeseries root.sg.d0.s2 with datatype=INT32, encoding=RLE, compression=SNAPPY"); - statement.execute( - "create timeseries root.sg.d0.s1 with datatype=INT32, encoding=RLE, compression=SNAPPY"); + statement.execute("CREATE DATABASE root.db1"); + statement.execute("CREATE DATABASE root.db2"); + statement.execute("CREATE DATABASE root.db3"); + + for (String ts : BASE_TIMESERIES) { + statement.execute( + String.format( + "create timeseries %s with datatype=INT32, encoding=RLE, compression=SNAPPY", ts)); + } } } - @Test - public void testOrderByTimeseriesAsc() throws Exception { - prepareSimpleSchema(); - - List expected = new ArrayList<>(Arrays.asList("root.ln.d0.s0", "root.sg.d0.s1")); - + private List queryTimeseries(final String sql) throws Exception { try (Connection connection = EnvFactory.getEnv().getConnection(); Statement statement = connection.createStatement(); - ResultSet resultSet = - statement.executeQuery("show timeseries order by timeseries offset 2 limit 2")) { - List actual = new ArrayList<>(); + ResultSet resultSet = statement.executeQuery(sql)) { + List result = new ArrayList<>(); while (resultSet.next()) { - actual.add(resultSet.getString(ColumnHeaderConstant.TIMESERIES)); + result.add(resultSet.getString(ColumnHeaderConstant.TIMESERIES)); } - assertEquals(expected, actual); + return result; } } @Test - public void testOrderByTimeseriesDescWithLimit() throws Exception { - prepareSimpleSchema(); + public void testOrderAscWithoutLimit() throws Exception { + prepareComplexSchema(); + List expected = new ArrayList<>(BASE_TIMESERIES); + Collections.sort(expected); - List all = - new ArrayList<>(Arrays.asList("root.ln.d0.s0", "root.sg.d0.s1", "root.sg.d0.s2")); - Collections.sort(all); - Collections.reverse(all); - List expectedTop2 = all.subList(0, 2); + List actual = queryTimeseries("show timeseries root.db*.** order by timeseries"); + assertEquals(expected, actual); + } + @Test + public void testOrderDescWithOffsetLimit() throws Exception { + prepareComplexSchema(); + List expected = new ArrayList<>(BASE_TIMESERIES); + Collections.sort(expected); + Collections.reverse(expected); + expected = expected.subList(2, 6); // offset 2 limit 4 + + List actual = + queryTimeseries("show timeseries root.db*.** order by timeseries desc offset 2 limit 4"); + assertEquals(expected, actual); + } + + @Test + public void testInsertThenQueryOrder() throws Exception { + prepareComplexSchema(); try (Connection connection = EnvFactory.getEnv().getConnection(); - Statement statement = connection.createStatement(); - ResultSet resultSet = - statement.executeQuery("show timeseries order by timeseries desc limit 2")) { - List actual = new ArrayList<>(); - while (resultSet.next()) { - actual.add(resultSet.getString(ColumnHeaderConstant.TIMESERIES)); - } - assertEquals(expectedTop2, actual); + Statement statement = connection.createStatement()) { + statement.execute( + "create timeseries root.db0.devX.a with datatype=INT32, encoding=RLE, compression=SNAPPY"); + } + + List expected = new ArrayList<>(BASE_TIMESERIES); + expected.add("root.db0.devX.a"); + Collections.sort(expected); + + List actual = queryTimeseries("show timeseries root.db*.** order by timeseries"); + assertEquals(expected, actual); + } + + @Test + public void testDeleteSubtreeThenQueryOrder() throws Exception { + prepareComplexSchema(); + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("delete timeseries root.db2.devC.**"); } + + List expected = new ArrayList<>(BASE_TIMESERIES); + expected.remove("root.db2.devC.m0"); + expected.remove("root.db2.devC.m3"); + Collections.sort(expected); + + List actual = queryTimeseries("show timeseries root.db*.** order by timeseries"); + assertEquals(expected, actual); + } + + @Test + public void testOffsetLimitAfterDeletesAndAdds() throws Exception { + prepareComplexSchema(); + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("delete timeseries root.db1.devB.x"); + statement.execute( + "create timeseries root.db1.devC.m0 with datatype=INT32, encoding=RLE, compression=SNAPPY"); + statement.execute( + "create timeseries root.db4.devZ.z with datatype=INT32, encoding=RLE, compression=SNAPPY"); + } + + List expected = new ArrayList<>(BASE_TIMESERIES); + expected.remove("root.db1.devB.x"); + expected.add("root.db1.devC.m0"); + expected.add("root.db4.devZ.z"); + Collections.sort(expected); + expected = expected.subList(5, 10); // offset 5 limit 5 + + List actual = queryTimeseries("show timeseries root.db*.** order by timeseries offset 5 limit 5"); + assertEquals(expected, actual); } @Test public void testConflictWithLatest() throws Exception { - prepareSimpleSchema(); + prepareComplexSchema(); try (Connection connection = EnvFactory.getEnv().getConnection(); Statement statement = connection.createStatement()) { try (ResultSet ignored = @@ -138,7 +204,7 @@ public void testConflictWithLatest() throws Exception { @Test public void testConflictWithTimeCondition() throws Exception { - prepareSimpleSchema(); + prepareComplexSchema(); try (Connection connection = EnvFactory.getEnv().getConnection(); Statement statement = connection.createStatement()) { try (ResultSet ignored = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java index 1878f3e54ec7b..dd325f367b929 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java @@ -73,6 +73,8 @@ import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.reader.impl.TimeseriesReaderWithViewFetch; import org.apache.iotdb.db.schemaengine.schemaregion.utils.MetaFormatUtils; import org.apache.iotdb.db.schemaengine.schemaregion.utils.filter.DeviceFilterVisitor; +import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager; +import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeSpaceQuotaManager; import org.apache.iotdb.db.utils.SchemaUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -192,6 +194,22 @@ public synchronized boolean createSnapshot(final File snapshotDir) { return store.createSnapshot(snapshotDir); } + private void applySubtreeMeasurementDelta(IMemMNode startNode, final long delta) { + if (delta == 0 || startNode == null) { + return; + } + IMemMNode current = startNode; + while (current != null) { + current.setSubtreeMeasurementCount(current.getSubtreeMeasurementCount() + delta); + current = current.getParent(); + } + } + + private long getTemplateMeasurementCount(final int templateId) { + final Template template = ClusterTemplateManager.getInstance().getTemplate(templateId); + return template == null ? 0L : template.getMeasurementNumber(); + } + public static MTreeBelowSGMemoryImpl loadFromSnapshot( final File snapshotDir, final String databaseFullPath, @@ -203,18 +221,21 @@ public static MTreeBelowSGMemoryImpl loadFromSnapshot( final Function, Map> tagGetter, final Function, Map> attributeGetter) throws IOException, IllegalPathException { - return new MTreeBelowSGMemoryImpl( - PartialPath.getQualifiedDatabasePartialPath(databaseFullPath), - MemMTreeStore.loadFromSnapshot( - snapshotDir, - measurementProcess, - deviceProcess, - tableDeviceProcess, - regionStatistics, - metric), - tagGetter, - attributeGetter, - regionStatistics); + final MTreeBelowSGMemoryImpl mtree = + new MTreeBelowSGMemoryImpl( + PartialPath.getQualifiedDatabasePartialPath(databaseFullPath), + MemMTreeStore.loadFromSnapshot( + snapshotDir, + measurementProcess, + deviceProcess, + tableDeviceProcess, + regionStatistics, + metric), + tagGetter, + attributeGetter, + regionStatistics); + mtree.rebuildSubtreeMeasurementCount(); + return mtree; } // endregion @@ -314,6 +335,7 @@ public IMeasurementMNode createTimeSeries( entityMNode.addAlias(alias, measurementMNode); } + applySubtreeMeasurementDelta(measurementMNode.getAsMNode(), 1L); return measurementMNode; } } @@ -411,6 +433,7 @@ public List> createAlignedTimeSeries( if (aliasList != null && aliasList.get(i) != null) { entityMNode.addAlias(aliasList.get(i), measurementMNode); } + applySubtreeMeasurementDelta(measurementMNode.getAsMNode(), 1L); measurementMNodeList.add(measurementMNode); } return measurementMNodeList; @@ -619,6 +642,7 @@ public IMeasurementMNode deleteTimeSeries(final PartialPath path) if (deletedNode.getAlias() != null) { parent.getAsDeviceMNode().deleteAliasChild(deletedNode.getAlias()); } + applySubtreeMeasurementDelta(parent, -1L); } deleteEmptyInternalMNode(parent.getAsDeviceMNode()); return deletedNode; @@ -1016,6 +1040,7 @@ public void activateTemplate(final PartialPath activatePath, final Template temp entityMNode.setUseTemplate(true); entityMNode.setSchemaTemplateId(template.getId()); regionStatistics.activateTemplate(template.getId()); + applySubtreeMeasurementDelta(entityMNode.getAsMNode(), (long) template.getMeasurementNumber()); } public Map> constructSchemaBlackListWithTemplate( @@ -1080,6 +1105,8 @@ protected void updateEntity(final IDeviceMNode node) { resultTemplateSetInfo.put( node.getPartialPath(), Collections.singletonList(node.getSchemaTemplateId())); regionStatistics.deactivateTemplate(node.getSchemaTemplateId()); + applySubtreeMeasurementDelta( + node.getAsMNode(), -getTemplateMeasurementCount(node.getSchemaTemplateId())); node.deactivateTemplate(); deleteEmptyInternalMNode(node); } @@ -1112,6 +1139,7 @@ public void activateTemplateWithoutCheck( entityMNode.setUseTemplate(true); entityMNode.setSchemaTemplateId(templateId); regionStatistics.activateTemplate(templateId); + applySubtreeMeasurementDelta(entityMNode.getAsMNode(), getTemplateMeasurementCount(templateId)); } public long countPathsUsingTemplate(final PartialPath pathPattern, final int templateId) @@ -1123,6 +1151,23 @@ public long countPathsUsingTemplate(final PartialPath pathPattern, final int tem } } + public void rebuildSubtreeMeasurementCount() { + rebuildSubtreeMeasurementCountFromNode(rootNode); + } + + private long rebuildSubtreeMeasurementCountFromNode(final IMemMNode node) { + long count = node.isMeasurement() ? 1L : 0L; + final IMNodeIterator iterator = store.getChildrenIterator(node); + while (iterator.hasNext()) { + count += rebuildSubtreeMeasurementCountFromNode(iterator.next()); + } + if (node.isDevice() && node.getAsDeviceMNode().isUseTemplate()) { + count += getTemplateMeasurementCount(node.getAsDeviceMNode().getSchemaTemplateId()); + } + node.setSubtreeMeasurementCount(count); + return count; + } + // endregion // region Interfaces for schema reader @@ -1435,6 +1480,46 @@ public ISchemaReader getTimeSeriesReader( showTimeSeriesPlan.isPrefixMatch(), showTimeSeriesPlan.getScope()) { + private long remainingOffset = Math.max(0, showTimeSeriesPlan.getOffset()); + + private boolean shouldPruneSubtree(final IMemMNode node) { + if (remainingOffset <= 0) { + return false; + } + final long subtreeCount = node.getSubtreeMeasurementCount(); + if (subtreeCount <= remainingOffset) { + remainingOffset -= subtreeCount; + return true; + } + return false; + } + + @Override + protected boolean acceptFullMatchedNode(final IMemMNode node) { + if (!node.isMeasurement()) { + return false; + } + if (remainingOffset > 0) { + // skip this measurement + remainingOffset--; + return false; + } + return true; + } + + @Override + protected boolean shouldVisitSubtreeOfInternalMatchedNode(final IMemMNode node) { + if (shouldPruneSubtree(node)) { + return false; + } + return !node.isMeasurement(); + } + + @Override + protected boolean shouldVisitSubtreeOfFullMatchedNode(final IMemMNode node) { + return !node.isMeasurement() && !shouldPruneSubtree(node); + } + @Override protected ITimeSeriesSchemaInfo collectMeasurement( final IMeasurementMNode node) { @@ -1531,9 +1616,8 @@ protected Iterator getChildrenIterator(final IMemMNode parent) final ISchemaReader reader = new TimeseriesReaderWithViewFetch( collector, showTimeSeriesPlan.getSchemaFilter(), showTimeSeriesPlan.needViewDetail()); - if (showTimeSeriesPlan.getLimit() > 0 || showTimeSeriesPlan.getOffset() > 0) { - return new SchemaReaderLimitOffsetWrapper<>( - reader, showTimeSeriesPlan.getLimit(), showTimeSeriesPlan.getOffset()); + if (showTimeSeriesPlan.getLimit() >= 0) { + return new SchemaReaderLimitOffsetWrapper<>(reader, showTimeSeriesPlan.getLimit(), 0); } else { return reader; } @@ -1645,6 +1729,7 @@ public IMeasurementMNode createLogicalView( measurementMNode.setParent(entityMNode.getAsMNode()); store.addChild(entityMNode.getAsMNode(), leafName, measurementMNode.getAsMNode()); + applySubtreeMeasurementDelta(measurementMNode.getAsMNode(), 1L); return measurementMNode; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/IMemMNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/IMemMNode.java index d3d055928b14a..6cd14800f3e3f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/IMemMNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/IMemMNode.java @@ -20,4 +20,13 @@ import org.apache.iotdb.commons.schema.node.IMNode; -public interface IMemMNode extends IMNode {} +public interface IMemMNode extends IMNode { + + /** + * The count of measurement nodes contained in the subtree rooted at this node. The counter is + * maintained in memory only. + */ + long getSubtreeMeasurementCount(); + + void setSubtreeMeasurementCount(long subtreeMeasurementCount); +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/basic/BasicMNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/basic/BasicMNode.java index a033c56c67e0e..2860bea6e9445 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/basic/BasicMNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/basic/BasicMNode.java @@ -46,6 +46,9 @@ public class BasicMNode implements IMemMNode { private IMemMNode parent; private final BasicMNodeInfo basicMNodeInfo; + /** Cached count of measurements in this node's subtree, restored on restart. */ + private long subtreeMeasurementCount = 0L; + /** from root to this node, only be set when used once for InternalMNode */ private String fullPath; @@ -99,6 +102,16 @@ public void setFullPath(final String fullPath) { this.fullPath = fullPath; } + @Override + public long getSubtreeMeasurementCount() { + return subtreeMeasurementCount; + } + + @Override + public void setSubtreeMeasurementCount(final long subtreeMeasurementCount) { + this.subtreeMeasurementCount = subtreeMeasurementCount; + } + @Override public PartialPath getPartialPath() { final List detachedPath = new ArrayList<>(); @@ -225,6 +238,7 @@ public R accept(final MNodeVisitor visitor, final C context) { *
  • basicMNodeInfo reference, 8B *
  • parent reference, 8B *
  • fullPath reference, 8B + *
  • subtreeMeasurementCount, 8B * *
  • MapEntry in parent *
      @@ -236,7 +250,7 @@ public R accept(final MNodeVisitor visitor, final C context) { */ @Override public int estimateSize() { - return 8 + 8 + 8 + 8 + 8 + 8 + 28 + basicMNodeInfo.estimateSize(); + return 8 + 8 + 8 + 8 + 8 + 8 + 8 + 28 + basicMNodeInfo.estimateSize(); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/AboveDatabaseMNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/AboveDatabaseMNode.java index cff30d8b8c462..87144d4954a5d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/AboveDatabaseMNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/AboveDatabaseMNode.java @@ -33,4 +33,14 @@ public AboveDatabaseMNode(IMemMNode parent, String name) { public IMemMNode getAsMNode() { return this; } + + @Override + public long getSubtreeMeasurementCount() { + return basicMNode.getSubtreeMeasurementCount(); + } + + @Override + public void setSubtreeMeasurementCount(long subtreeMeasurementCount) { + basicMNode.setSubtreeMeasurementCount(subtreeMeasurementCount); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/DatabaseMNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/DatabaseMNode.java index c6b2f5e2427bc..290cf4273601b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/DatabaseMNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/DatabaseMNode.java @@ -45,4 +45,14 @@ public IDeviceInfo getDeviceInfo() { public void setDeviceInfo(IDeviceInfo deviceInfo) { basicMNode.setDeviceInfo(deviceInfo); } + + @Override + public long getSubtreeMeasurementCount() { + return basicMNode.getSubtreeMeasurementCount(); + } + + @Override + public void setSubtreeMeasurementCount(long subtreeMeasurementCount) { + basicMNode.setSubtreeMeasurementCount(subtreeMeasurementCount); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/MeasurementMNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/MeasurementMNode.java index a40cbe6bc0fc9..d2a2cbd80c90c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/MeasurementMNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/impl/MeasurementMNode.java @@ -52,4 +52,14 @@ public IMNodeContainer getChildren() { public final boolean isLogicalView() { return false; } + + @Override + public long getSubtreeMeasurementCount() { + return basicMNode.getSubtreeMeasurementCount(); + } + + @Override + public void setSubtreeMeasurementCount(long subtreeMeasurementCount) { + basicMNode.setSubtreeMeasurementCount(subtreeMeasurementCount); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/AbstractShowSchemaPlanImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/AbstractShowSchemaPlanImpl.java index 8694d8c09ec09..558ce8b188343 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/AbstractShowSchemaPlanImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/AbstractShowSchemaPlanImpl.java @@ -37,7 +37,7 @@ public abstract class AbstractShowSchemaPlanImpl implements IShowSchemaPlan { protected AbstractShowSchemaPlanImpl(PartialPath path) { this.path = path; this.scope = SchemaConstant.ALL_MATCH_SCOPE; - this.limit = 0; + this.limit = -1; this.offset = 0; this.isPrefixMatch = false; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java index 3d9d6abc54efb..f972f8930fb5d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java @@ -45,7 +45,7 @@ public SchemaReaderLimitOffsetWrapper(ISchemaReader schemaReader, long limit, this.schemaReader = schemaReader; this.limit = limit; this.offset = offset; - this.hasLimit = limit > 0 || offset > 0; + this.hasLimit = limit >= 0 || offset > 0; } @Override From 7dba6909a52004af6c0a3e2a440791c0d7009b65 Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Thu, 22 Jan 2026 20:18:54 +0800 Subject: [PATCH 06/10] code-style: spotless apply --- .../db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java index ca53cdb19446c..e4edb54eb4659 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java @@ -182,7 +182,8 @@ public void testOffsetLimitAfterDeletesAndAdds() throws Exception { Collections.sort(expected); expected = expected.subList(5, 10); // offset 5 limit 5 - List actual = queryTimeseries("show timeseries root.db*.** order by timeseries offset 5 limit 5"); + List actual = + queryTimeseries("show timeseries root.db*.** order by timeseries offset 5 limit 5"); assertEquals(expected, actual); } From e6c29811b75f22f0bfbdf6e05bd3301965baace6 Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Thu, 22 Jan 2026 22:20:33 +0800 Subject: [PATCH 07/10] fix: limit default value reset to 0 due to project convention. --- .../db/queryengine/plan/planner/LogicalPlanBuilder.java | 2 +- .../db/queryengine/plan/planner/LogicalPlanVisitor.java | 5 +++-- .../queryengine/plan/statement/metadata/ShowStatement.java | 4 ++-- .../schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java | 3 +-- .../read/req/impl/AbstractShowSchemaPlanImpl.java | 2 +- .../resp/reader/impl/SchemaReaderLimitOffsetWrapper.java | 2 +- 6 files changed, 9 insertions(+), 9 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java index e30e325f26b36..66c55a54c0f62 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java @@ -905,7 +905,7 @@ public LogicalPlanBuilder planFill(FillDescriptor fillDescriptor, Ordering scanO } public LogicalPlanBuilder planLimit(long rowLimit) { - if (rowLimit < 0) { + if (rowLimit <= 0) { return this; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java index 373039578b0cd..263cc717c46cf 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java @@ -97,6 +97,7 @@ import org.apache.tsfile.write.schema.MeasurementSchema; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; @@ -584,7 +585,7 @@ public PlanNode visitShowTimeSeries( boolean canPushDownOffsetLimit = singleSchemaRegion && !showTimeSeriesStatement.isOrderByHeat(); if (showTimeSeriesStatement.isOrderByHeat()) { - limit = -1; + limit = 0; offset = 0; } else if (!canPushDownOffsetLimit) { limit = showTimeSeriesStatement.getLimitWithOffset(); @@ -613,7 +614,7 @@ public PlanNode visitShowTimeSeries( if (showTimeSeriesStatement.isOrderByTimeseries() && !singleSchemaRegion) { SortItem sortItem = new SortItem(ColumnHeaderConstant.TIMESERIES, showTimeSeriesStatement.getNameOrdering()); - planBuilder = planBuilder.planOrderBy(java.util.Collections.singletonList(sortItem)); + planBuilder = planBuilder.planOrderBy(Collections.singletonList(sortItem)); } // show latest timeseries diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowStatement.java index a283204d80e57..b839791821767 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/ShowStatement.java @@ -28,7 +28,7 @@ public class ShowStatement extends AuthorityInformationStatement { - long limit = -1; + long limit = 0; long offset = 0; protected boolean isPrefixPath; @@ -60,7 +60,7 @@ public void setOffset(long offset) { } public long getLimitWithOffset() { - if (limit < 0) { + if (limit <= 0) { return limit; } return limit + offset; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java index dd325f367b929..868a47d6143cd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java @@ -74,7 +74,6 @@ import org.apache.iotdb.db.schemaengine.schemaregion.utils.MetaFormatUtils; import org.apache.iotdb.db.schemaengine.schemaregion.utils.filter.DeviceFilterVisitor; import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager; -import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeSpaceQuotaManager; import org.apache.iotdb.db.utils.SchemaUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -1616,7 +1615,7 @@ protected Iterator getChildrenIterator(final IMemMNode parent) final ISchemaReader reader = new TimeseriesReaderWithViewFetch( collector, showTimeSeriesPlan.getSchemaFilter(), showTimeSeriesPlan.needViewDetail()); - if (showTimeSeriesPlan.getLimit() >= 0) { + if (showTimeSeriesPlan.getLimit() > 0) { return new SchemaReaderLimitOffsetWrapper<>(reader, showTimeSeriesPlan.getLimit(), 0); } else { return reader; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/AbstractShowSchemaPlanImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/AbstractShowSchemaPlanImpl.java index 558ce8b188343..8694d8c09ec09 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/AbstractShowSchemaPlanImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/req/impl/AbstractShowSchemaPlanImpl.java @@ -37,7 +37,7 @@ public abstract class AbstractShowSchemaPlanImpl implements IShowSchemaPlan { protected AbstractShowSchemaPlanImpl(PartialPath path) { this.path = path; this.scope = SchemaConstant.ALL_MATCH_SCOPE; - this.limit = -1; + this.limit = 0; this.offset = 0; this.isPrefixMatch = false; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java index f972f8930fb5d..3d9d6abc54efb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java @@ -45,7 +45,7 @@ public SchemaReaderLimitOffsetWrapper(ISchemaReader schemaReader, long limit, this.schemaReader = schemaReader; this.limit = limit; this.offset = offset; - this.hasLimit = limit >= 0 || offset > 0; + this.hasLimit = limit > 0 || offset > 0; } @Override From 0177c6699654a66f38fb92809c28c65551a3419a Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Tue, 27 Jan 2026 19:40:41 +0800 Subject: [PATCH 08/10] fix: disable the offset-pruning pushdown when showTimeSeriesPlan.getSchemaFilter() is non-null and when subtree may be not completely included in pathPattern. --- .../plan/planner/LogicalPlanVisitor.java | 5 ++- .../impl/mem/MTreeBelowSGMemoryImpl.java | 30 ++++++++++++++++ .../impl/SchemaReaderLimitOffsetWrapper.java | 36 +++++++++---------- 3 files changed, 50 insertions(+), 21 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java index 263cc717c46cf..c4779fe3e23a0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java @@ -582,7 +582,10 @@ public PlanNode visitShowTimeSeries( boolean singleSchemaRegion = analysis.getSchemaPartitionInfo() != null && analysis.getSchemaPartitionInfo().getDistributionInfo().size() == 1; - boolean canPushDownOffsetLimit = singleSchemaRegion && !showTimeSeriesStatement.isOrderByHeat(); + boolean canPushDownOffsetLimit = + singleSchemaRegion + && !showTimeSeriesStatement.isOrderByHeat() + && showTimeSeriesStatement.getSchemaFilter() == null; if (showTimeSeriesStatement.isOrderByHeat()) { limit = 0; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java index 868a47d6143cd..03b3277ab36ca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PathPatternTree; +import org.apache.iotdb.commons.path.PathPatternUtil; import org.apache.iotdb.commons.schema.SchemaConstant; import org.apache.iotdb.commons.schema.node.role.IDeviceMNode; import org.apache.iotdb.commons.schema.node.role.IMeasurementMNode; @@ -1480,11 +1481,40 @@ public ISchemaReader getTimeSeriesReader( showTimeSeriesPlan.getScope()) { private long remainingOffset = Math.max(0, showTimeSeriesPlan.getOffset()); + private final String[] prunePrefixNodes = + getSafePrunePrefixNodes(showTimeSeriesPlan.getPath()); + + private String[] getSafePrunePrefixNodes(final PartialPath pattern) { + if (pattern == null || !pattern.endWithMultiLevelWildcard()) { + return null; + } + final String[] nodes = pattern.getNodes(); + return Arrays.copyOf(nodes, nodes.length - 1); + } + + private boolean isUnderPrunePrefix(final IMemMNode node) { + if (prunePrefixNodes == null) { + return false; + } + final String[] nodePath = getPartialPathFromRootToNode(node).getNodes(); + if (nodePath.length < prunePrefixNodes.length) { + return false; + } + for (int i = 0; i < prunePrefixNodes.length; i++) { + if (!PathPatternUtil.isNodeMatch(prunePrefixNodes[i], nodePath[i])) { + return false; + } + } + return true; + } private boolean shouldPruneSubtree(final IMemMNode node) { if (remainingOffset <= 0) { return false; } + if (!isUnderPrunePrefix(node)) { + return false; + } final long subtreeCount = node.getSubtreeMeasurementCount(); if (subtreeCount <= remainingOffset) { remainingOffset -= subtreeCount; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java index 3d9d6abc54efb..04f4ca8b6f934 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java @@ -45,7 +45,7 @@ public SchemaReaderLimitOffsetWrapper(ISchemaReader schemaReader, long limit, this.schemaReader = schemaReader; this.limit = limit; this.offset = offset; - this.hasLimit = limit > 0 || offset > 0; + this.hasLimit = limit > 0; } @Override @@ -73,24 +73,20 @@ public ListenableFuture isBlocked() { } private ListenableFuture tryGetNext() { - if (hasLimit) { - if (curOffset < offset) { - // first time - return Futures.submit( - () -> { - while (curOffset < offset && schemaReader.hasNext()) { - schemaReader.next(); - curOffset++; - } - return schemaReader.hasNext(); - }, - directExecutor()); - } - if (count >= limit) { - return NOT_BLOCKED; - } else { - return schemaReader.isBlocked(); - } + if (curOffset < offset) { + // first time + return Futures.submit( + () -> { + while (curOffset < offset && schemaReader.hasNext()) { + schemaReader.next(); + curOffset++; + } + return schemaReader.hasNext(); + }, + directExecutor()); + } + if (hasLimit && count >= limit) { + return NOT_BLOCKED; } else { return schemaReader.isBlocked(); } @@ -101,7 +97,7 @@ private ListenableFuture tryGetNext() { public boolean hasNext() { try { isBlocked().get(); - return schemaReader.hasNext() && count < limit; + return schemaReader.hasNext() && (!hasLimit || count < limit); } catch (Exception e) { throw new RuntimeException(e); } From 4027f9724ad417771b64eff82a53c90180a47297 Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Tue, 27 Jan 2026 19:44:25 +0800 Subject: [PATCH 09/10] fix: update subtreeMeasurementCount for all affected paths when a template is extended --- .../impl/DataNodeInternalRPCServiceImpl.java | 14 ++++++++++ .../iotdb/db/schemaengine/SchemaEngine.java | 21 ++++++++++++++ .../impl/SchemaRegionMemoryImpl.java | 8 ++++++ .../impl/mem/MTreeBelowSGMemoryImpl.java | 28 +++++++++++++++++++ 4 files changed, 71 insertions(+) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java index 46d377be0a1bd..a2ea91192e7a6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java @@ -77,6 +77,7 @@ import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.commons.schema.table.TsTableInternalRPCType; import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil; +import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.commons.schema.view.ViewType; import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression; import org.apache.iotdb.commons.service.metric.MetricService; @@ -186,6 +187,7 @@ import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager; import org.apache.iotdb.db.schemaengine.template.TemplateInternalRPCUpdateType; +import org.apache.iotdb.db.schemaengine.template.TemplateInternalRPCUtil; import org.apache.iotdb.db.service.DataNode; import org.apache.iotdb.db.service.RegionMigrateService; import org.apache.iotdb.db.service.externalservice.ExternalServiceManagementService; @@ -2571,7 +2573,19 @@ public TSStatus updateTemplate(final TUpdateTemplateReq req) { ClusterTemplateManager.getInstance().commitTemplatePreSetInfo(req.getTemplateInfo()); break; case UPDATE_TEMPLATE_INFO: + Template newTemplate = + TemplateInternalRPCUtil.parseUpdateTemplateInfoBytes( + ByteBuffer.wrap(req.getTemplateInfo())); + Template oldTemplate = + ClusterTemplateManager.getInstance().getTemplate(newTemplate.getId()); ClusterTemplateManager.getInstance().updateTemplateInfo(req.getTemplateInfo()); + long delta = + newTemplate.getMeasurementNumber() + - (oldTemplate == null ? 0 : oldTemplate.getMeasurementNumber()); + if (delta != 0) { + SchemaEngine.getInstance() + .updateSubtreeMeasurementCountForTemplate(newTemplate.getId(), delta); + } break; default: LOGGER.warn("Unsupported type {} when updating template", req.type); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java index 3f59f9edf68ed..9334e6a85627b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java @@ -43,6 +43,7 @@ import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegionParams; import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionLoader; import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionParams; +import org.apache.iotdb.db.schemaengine.schemaregion.impl.SchemaRegionMemoryImpl; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager; import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatReq; @@ -253,6 +254,26 @@ public List getAllSchemaRegionIds() { return new ArrayList<>(schemaRegionMap.keySet()); } + public void updateSubtreeMeasurementCountForTemplate(final int templateId, final long delta) { + if (delta == 0) { + return; + } + for (final ISchemaRegion schemaRegion : schemaRegionMap.values()) { + if (schemaRegion instanceof SchemaRegionMemoryImpl) { + try { + ((SchemaRegionMemoryImpl) schemaRegion) + .updateSubtreeMeasurementCountForTemplate(templateId, delta); + } catch (MetadataException e) { + logger.warn( + "Failed to update subtree measurement count for template {} in schemaRegion {}", + templateId, + schemaRegion.getSchemaRegionId(), + e); + } + } + } + } + public synchronized void createSchemaRegion( final String storageGroup, final SchemaRegionId schemaRegionId) throws MetadataException { if (this.schemaRegionMap == null) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java index 230ed8330ca16..0719835fe7a61 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java @@ -1450,6 +1450,14 @@ public long countPathsUsingTemplate(final int templateId, final PathPatternTree return result; } + public void updateSubtreeMeasurementCountForTemplate(final int templateId, final long delta) + throws MetadataException { + if (delta == 0 || mTree == null) { + return; + } + mTree.updateSubtreeMeasurementCountForTemplate(templateId, delta); + } + @Override public int fillLastQueryMap( final PartialPath pattern, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java index 03b3277ab36ca..07ffe05d298c8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java @@ -1142,6 +1142,34 @@ public void activateTemplateWithoutCheck( applySubtreeMeasurementDelta(entityMNode.getAsMNode(), getTemplateMeasurementCount(templateId)); } + public void updateSubtreeMeasurementCountForTemplate(final int templateId, final long delta) + throws MetadataException { + if (delta == 0) { + return; + } + final PartialPath pattern = + new PartialPath( + databaseMNode.getFullPath() + + IoTDBConstant.PATH_SEPARATOR + + IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD); + try (final EntityUpdater updater = + new EntityUpdater( + rootNode, pattern, store, false, SchemaConstant.ALL_MATCH_SCOPE) { + @Override + protected void updateEntity(final IDeviceMNode node) { + if (!node.isUseTemplate() || node.getSchemaTemplateId() != templateId) { + return; + } + synchronized (MTreeBelowSGMemoryImpl.this) { + applySubtreeMeasurementDelta(node.getAsMNode(), delta); + } + } + }) { + updater.setSchemaTemplateFilter(templateId); + updater.update(); + } + } + public long countPathsUsingTemplate(final PartialPath pathPattern, final int templateId) throws MetadataException { try (final EntityCounter counter = From c56e155754074d42eed1396b4304929b1bb0bce5 Mon Sep 17 00:00:00 2001 From: xiangmy21 <2248278431@qq.com> Date: Tue, 27 Jan 2026 22:16:48 +0800 Subject: [PATCH 10/10] fix: add test for where clause and template change. correct pushdown in PBTree mode. --- ...oTDBShowTimeseriesOrderByTimeseriesIT.java | 95 +++++++++++++------ .../plan/planner/LogicalPlanVisitor.java | 29 ++++-- .../impl/mem/mnode/basic/BasicMNode.java | 2 +- 3 files changed, 90 insertions(+), 36 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java index e4edb54eb4659..5d82ea3539962 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBShowTimeseriesOrderByTimeseriesIT.java @@ -37,6 +37,8 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -45,18 +47,13 @@ @Category({LocalStandaloneIT.class, ClusterIT.class}) public class IoTDBShowTimeseriesOrderByTimeseriesIT extends AbstractSchemaIT { - private static final List BASE_TIMESERIES = - Arrays.asList( - "root.db1.devA.m1", - "root.db1.devA.m2", - "root.db1.devB.m1", - "root.db1.devB.x", - "root.db2.devA.m1", - "root.db2.devC.m0", - "root.db2.devC.m3", - "root.db3.z.m1", - "root.db3.z.m10", - "root.db3.z.m2"); + private static final List BASE_TIMESERIES_DB1 = + Arrays.asList("root.db1.devA.m1", "root.db1.devB.m1", "root.db1.devA.m2", "root.db1.devB.x"); + private static final List BASE_TIMESERIES_DB2 = + Arrays.asList("root.db2.devA.m1", "root.db2.devC.m3", "root.db2.devC.m0"); + private static final List BASE_TIMESERIES = // combine db1 and db2 + Stream.concat(BASE_TIMESERIES_DB1.stream(), BASE_TIMESERIES_DB2.stream()) + .collect(Collectors.toList()); public IoTDBShowTimeseriesOrderByTimeseriesIT(SchemaTestMode schemaTestMode) { super(schemaTestMode); @@ -84,7 +81,6 @@ private void prepareComplexSchema() throws Exception { Statement statement = connection.createStatement()) { statement.execute("CREATE DATABASE root.db1"); statement.execute("CREATE DATABASE root.db2"); - statement.execute("CREATE DATABASE root.db3"); for (String ts : BASE_TIMESERIES) { statement.execute( @@ -119,13 +115,13 @@ public void testOrderAscWithoutLimit() throws Exception { @Test public void testOrderDescWithOffsetLimit() throws Exception { prepareComplexSchema(); - List expected = new ArrayList<>(BASE_TIMESERIES); + List expected = new ArrayList<>(BASE_TIMESERIES_DB1); Collections.sort(expected); Collections.reverse(expected); - expected = expected.subList(2, 6); // offset 2 limit 4 + expected = expected.subList(1, 3); // offset 1 limit 2 List actual = - queryTimeseries("show timeseries root.db*.** order by timeseries desc offset 2 limit 4"); + queryTimeseries("show timeseries root.db1.** order by timeseries desc offset 1 limit 2"); assertEquals(expected, actual); } @@ -135,14 +131,14 @@ public void testInsertThenQueryOrder() throws Exception { try (Connection connection = EnvFactory.getEnv().getConnection(); Statement statement = connection.createStatement()) { statement.execute( - "create timeseries root.db0.devX.a with datatype=INT32, encoding=RLE, compression=SNAPPY"); + "create timeseries root.db1.devX.a with datatype=INT32, encoding=RLE, compression=SNAPPY"); } - List expected = new ArrayList<>(BASE_TIMESERIES); - expected.add("root.db0.devX.a"); + List expected = new ArrayList<>(BASE_TIMESERIES_DB1); + expected.add("root.db1.devX.a"); Collections.sort(expected); - List actual = queryTimeseries("show timeseries root.db*.** order by timeseries"); + List actual = queryTimeseries("show timeseries root.db1.** order by timeseries"); assertEquals(expected, actual); } @@ -154,12 +150,12 @@ public void testDeleteSubtreeThenQueryOrder() throws Exception { statement.execute("delete timeseries root.db2.devC.**"); } - List expected = new ArrayList<>(BASE_TIMESERIES); + List expected = new ArrayList<>(BASE_TIMESERIES_DB2); expected.remove("root.db2.devC.m0"); expected.remove("root.db2.devC.m3"); Collections.sort(expected); - List actual = queryTimeseries("show timeseries root.db*.** order by timeseries"); + List actual = queryTimeseries("show timeseries root.db2.** order by timeseries"); assertEquals(expected, actual); } @@ -172,18 +168,18 @@ public void testOffsetLimitAfterDeletesAndAdds() throws Exception { statement.execute( "create timeseries root.db1.devC.m0 with datatype=INT32, encoding=RLE, compression=SNAPPY"); statement.execute( - "create timeseries root.db4.devZ.z with datatype=INT32, encoding=RLE, compression=SNAPPY"); + "create timeseries root.db1.devZ.z with datatype=INT32, encoding=RLE, compression=SNAPPY"); } - List expected = new ArrayList<>(BASE_TIMESERIES); + List expected = new ArrayList<>(BASE_TIMESERIES_DB1); expected.remove("root.db1.devB.x"); expected.add("root.db1.devC.m0"); - expected.add("root.db4.devZ.z"); + expected.add("root.db1.devZ.z"); Collections.sort(expected); - expected = expected.subList(5, 10); // offset 5 limit 5 + expected = expected.subList(2, 4); // offset 2 limit 2 List actual = - queryTimeseries("show timeseries root.db*.** order by timeseries offset 5 limit 5"); + queryTimeseries("show timeseries root.db1.** order by timeseries offset 2 limit 2"); assertEquals(expected, actual); } @@ -218,4 +214,49 @@ public void testConflictWithTimeCondition() throws Exception { } } } + + @Test + public void testWhereClauseOffsetAppliedAfterFilter() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE root.ln"); + statement.execute( + "create timeseries root.ln.wf01.wt01.status with datatype=INT32, encoding=RLE, compression=SNAPPY"); + statement.execute( + "create timeseries root.ln.wf02.wt01.status with datatype=INT32, encoding=RLE, compression=SNAPPY"); + statement.execute( + "create timeseries root.ln.wf02.wt02.status with datatype=INT32, encoding=RLE, compression=SNAPPY"); + } + + List actual = + queryTimeseries( + "show timeseries root.ln.** where timeseries contains 'wf02.wt' order by timeseries offset 1 limit 1"); + assertEquals(Collections.singletonList("root.ln.wf02.wt02.status"), actual); + } + + @Test + public void testAlterTemplateUpdatesOffsetOrder() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE root.sg1"); + statement.execute("create device template t1 (s1 INT32)"); + statement.execute("set device template t1 to root.sg1.d1"); + statement.execute("create timeseries using device template on root.sg1.d1"); + statement.execute("set device template t1 to root.sg1.d2"); + statement.execute("create timeseries using device template on root.sg1.d2"); + } + + List before = + queryTimeseries("show timeseries root.sg1.** order by timeseries desc offset 1 limit 1"); + assertEquals(Arrays.asList("root.sg1.d1.s1"), before); + + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("alter device template t1 add (s2 INT32)"); + } + + List after = + queryTimeseries("show timeseries root.sg1.** order by timeseries desc offset 2 limit 2"); + assertEquals(Arrays.asList("root.sg1.d1.s2", "root.sg1.d1.s1"), after); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java index c4779fe3e23a0..76def25d1c3a4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java @@ -18,6 +18,7 @@ */ package org.apache.iotdb.db.queryengine.plan.planner; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; import org.apache.iotdb.commons.schema.template.Template; @@ -90,6 +91,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.pipe.PipeEnrichedStatement; import org.apache.iotdb.db.queryengine.plan.statement.sys.ExplainAnalyzeStatement; import org.apache.iotdb.db.queryengine.plan.statement.sys.ShowQueriesStatement; +import org.apache.iotdb.db.schemaengine.SchemaEngineMode; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; @@ -582,17 +584,27 @@ public PlanNode visitShowTimeSeries( boolean singleSchemaRegion = analysis.getSchemaPartitionInfo() != null && analysis.getSchemaPartitionInfo().getDistributionInfo().size() == 1; + boolean isMemorySchemaEngine = + CommonDescriptor.getInstance() + .getConfig() + .getSchemaEngineMode() + .equals(SchemaEngineMode.Memory.toString()); + boolean isNeedSortDescInPBTree = + !isMemorySchemaEngine && showTimeSeriesStatement.getNameOrdering() == Ordering.DESC; boolean canPushDownOffsetLimit = singleSchemaRegion + && !isNeedSortDescInPBTree && !showTimeSeriesStatement.isOrderByHeat() && showTimeSeriesStatement.getSchemaFilter() == null; - if (showTimeSeriesStatement.isOrderByHeat()) { - limit = 0; - offset = 0; - } else if (!canPushDownOffsetLimit) { - limit = showTimeSeriesStatement.getLimitWithOffset(); - offset = 0; + if (!canPushDownOffsetLimit) { + if (showTimeSeriesStatement.isOrderByHeat() || isNeedSortDescInPBTree) { + limit = 0; + offset = 0; + } else { + limit = showTimeSeriesStatement.getLimitWithOffset(); + offset = 0; + } } boolean orderByTimeseries = showTimeSeriesStatement.isOrderByTimeseries(); boolean orderByTimeseriesDesc = @@ -613,8 +625,9 @@ public PlanNode visitShowTimeSeries( orderByTimeseriesDesc) .planSchemaQueryMerge(showTimeSeriesStatement.isOrderByHeat()); - // order by timeseries name in multi-region case: still need global SortNode - if (showTimeSeriesStatement.isOrderByTimeseries() && !singleSchemaRegion) { + // order by timeseries name in multi-region or PBTree-DESC case: still need global SortNode + if (showTimeSeriesStatement.isOrderByTimeseries() + && (!singleSchemaRegion || isNeedSortDescInPBTree)) { SortItem sortItem = new SortItem(ColumnHeaderConstant.TIMESERIES, showTimeSeriesStatement.getNameOrdering()); planBuilder = planBuilder.planOrderBy(Collections.singletonList(sortItem)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/basic/BasicMNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/basic/BasicMNode.java index 2860bea6e9445..2eab69749fd79 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/basic/BasicMNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/mnode/basic/BasicMNode.java @@ -46,7 +46,7 @@ public class BasicMNode implements IMemMNode { private IMemMNode parent; private final BasicMNodeInfo basicMNodeInfo; - /** Cached count of measurements in this node's subtree, restored on restart. */ + /** Cached count of measurements in this node's subtree, rebuilt on restart. */ private long subtreeMeasurementCount = 0L; /** from root to this node, only be set when used once for InternalMNode */