This is an automated email from the ASF dual-hosted git repository. chenyz pushed a commit to branch builtin-udtf in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c3c3e39e322694fd0f39186bc474662f77e24ecd Merge: 1496caa6ac8 8630cb6fabd Author: Chen YZ <[email protected]> AuthorDate: Wed Mar 19 18:18:53 2025 +0800 save .../it/cluster/IoTDBClusterNodeGetterIT.java | 2 +- .../iotdb/db/it/schema/IoTDBExtendTemplateIT.java | 7 + .../db/it/IoTDBMultiTAGsWithAttributesTableIT.java | 16 +- .../IoTDBCorrelatedExistsSubqueryIT.java | 24 +- ...TDBConsumer2With1TopicShareProcessTsfileIT.java | 14 +- .../IoTDBPathLooseDeviceTsfilePushConsumerIT.java | 10 +- .../IoTDBTimeLooseTsfilePushConsumerIT.java | 10 +- .../IoTDBTSPatternTsfilePushConsumerIT.java | 2 +- .../time/IoTDBRealTimeDBTsfilePushConsumerIT.java | 8 +- .../time/IoTDBTimeRangeDBTsfilePushConsumerIT.java | 22 +- .../java/org/apache/iotdb/cli/AbstractCli.java | 2 +- .../confignode/client/CnToCnNodeRequestType.java | 2 +- .../client/async/CnToDnAsyncRequestType.java | 1 + .../CnToDnInternalServiceAsyncRequestManager.java | 6 + .../rpc/DataNodeAsyncRequestRPCHandler.java | 1 + .../client/sync/SyncConfigNodeClientPool.java | 6 +- .../write/pipe/payload/PipeDeleteDevicesPlan.java | 44 +-- .../request/write/table/AbstractTablePlan.java | 15 +- .../iotdb/confignode/manager/ProcedureManager.java | 2 +- .../extractor/ConfigRegionListeningFilter.java | 3 +- ...ConfigPhysicalPlanTablePatternParseVisitor.java | 62 ++-- ...nfigPhysicalPlanTablePrivilegeParseVisitor.java | 82 +++--- .../receiver/protocol/IoTDBConfigNodeReceiver.java | 37 +-- .../procedure/env/ConfigNodeProcedureEnv.java | 20 +- .../impl/node/RemoveConfigNodeProcedure.java | 8 +- .../impl/region/CreateRegionGroupsProcedure.java | 35 ++- .../region/NotifyRegionMigrationProcedure.java | 137 +++++++++ .../impl/region/RegionMigrateProcedure.java | 2 + ...eState.java => NotifyRegionMigrationState.java} | 6 +- .../procedure/state/RemoveConfigNodeState.java | 2 +- .../procedure/store/ProcedureFactory.java | 7 + .../confignode/procedure/store/ProcedureType.java | 1 + .../thrift/ConfigNodeRPCServiceProcessor.java | 10 +- ...igPhysicalPlanTablePatternParseVisitorTest.java | 9 + .../protocol/opcda/OpcDaServerHandle.java | 4 +- .../connector/protocol/opcua/OpcUaNameSpace.java | 4 +- .../async/IoTDBDataRegionAsyncConnector.java | 21 -- .../db/pipe/event/common/PipeInsertionEvent.java | 13 - .../tablet/PipeInsertNodeTabletInsertionEvent.java | 9 +- .../common/tablet/PipeRawTabletInsertionEvent.java | 4 +- .../tablet/parser/TabletInsertionEventParser.java | 15 +- .../common/tsfile/PipeTsFileInsertionEvent.java | 70 +++-- .../parser/TsFileInsertionEventParserProvider.java | 19 +- .../PipeRealtimeDataRegionHybridExtractor.java | 12 +- .../pipeconsensus/PipeConsensusReceiver.java | 17 +- .../protocol/thrift/IoTDBDataNodeReceiver.java | 52 +++- .../resource/memory/InsertNodeMemoryEstimator.java | 319 ++++----------------- .../pipe/resource/tsfile/PipeTsFileResource.java | 2 +- .../iotdb/db/protocol/client/ConfigNodeClient.java | 4 +- .../impl/DataNodeInternalRPCServiceImpl.java | 22 +- .../fragment/FragmentInstanceContext.java | 38 ++- .../schema/source/DeviceAttributeUpdater.java | 10 +- .../schema/source/DeviceBlackListConstructor.java | 11 +- .../schema/source/DevicePredicateFilter.java | 13 +- .../schema/source/DevicePredicateHandler.java | 22 +- .../operator/schema/source/DeviceUpdater.java | 13 +- .../schema/source/SchemaSourceFactory.java | 4 +- .../schema/source/TableDeviceQuerySource.java | 21 +- .../plan/analyze/schema/TemplateSchemaFetcher.java | 11 +- .../config/executor/ClusterConfigTaskExecutor.java | 8 +- .../plan/planner/TableOperatorGenerator.java | 37 ++- .../plan/planner/plan/node/PlanGraphPrinter.java | 15 +- .../plan/relational/analyzer/Analysis.java | 17 -- .../metadata/fetcher/TableDeviceSchemaFetcher.java | 5 + .../relational/planner/TableLogicalPlanner.java | 2 +- .../distribute/TableDistributedPlanGenerator.java | 228 +++++++-------- .../plan/relational/planner/node/GroupNode.java | 47 +-- .../planner/optimizations/ParallelizeGrouping.java | 130 +++++---- .../optimizations/PushPredicateIntoTableScan.java | 3 +- .../optimizations/UnaliasSymbolReferences.java | 3 - .../sql/ast/AbstractQueryDeviceWithCache.java | 33 ++- .../relational/sql/ast/AbstractTraverseDevice.java | 9 +- .../plan/relational/sql/ast/DeleteDevice.java | 19 +- .../plan/relational/sql/ast/ShowDevice.java | 10 +- .../plan/scheduler/load/LoadTsFileScheduler.java | 10 + .../plan/statement/crud/InsertBaseStatement.java | 46 ++- .../crud/InsertMultiTabletsStatement.java | 16 ++ .../plan/statement/crud/InsertRowStatement.java | 13 + .../crud/InsertRowsOfOneDeviceStatement.java | 16 ++ .../plan/statement/crud/InsertRowsStatement.java | 16 ++ .../plan/statement/crud/InsertTabletStatement.java | 18 ++ .../attribute/update/UpdateDetailContainer.java | 2 +- .../schemaregion/impl/SchemaRegionMemoryImpl.java | 29 +- .../iotdb/db/service/RegionMigrateService.java | 18 ++ .../buffer/TimeSeriesMetadataCache.java | 3 +- .../db/storageengine/dataregion/DataRegion.java | 7 +- .../dataregion/VirtualDataRegion.java | 2 +- .../writer/AbstractCrossCompactionWriter.java | 25 +- .../utils/writer/FastCrossCompactionWriter.java | 12 + .../dataregion/tsfile/TsFileResource.java | 50 ++-- .../tsfile/timeindex/ArrayDeviceTimeIndex.java | 12 +- .../plan/relational/analyzer/AnalyzerTest.java | 10 +- .../relational/analyzer/TableFunctionTest.java | 1 - .../FastCrossCompactionPerformerTest.java | 78 +++++ .../apache/iotdb/commons/conf/CommonConfig.java | 12 +- .../iotdb/commons/conf/CommonDescriptor.java | 5 + .../consensus/index/impl/HybridProgressIndex.java | 4 +- .../iotdb/commons/pipe/config/PipeConfig.java | 7 + .../commons/schema/view/LogicalViewSchema.java | 13 +- .../multi/FunctionViewExpression.java | 14 + .../viewExpression/unary/LikeViewExpression.java | 39 ++- .../relational/tvf/CapacityTableFunction.java | 7 +- .../relational/tvf/SessionTableFunction.java | 3 +- .../relational/tvf/VariationTableFunction.java | 6 +- .../iotdb/commons/utils/TimePartitionUtils.java | 7 +- .../schema/LikeViewExpreesionSerDeTest.java | 48 ++++ .../src/main/thrift/confignode.thrift | 4 +- .../src/main/thrift/datanode.thrift | 10 + 108 files changed, 1439 insertions(+), 1035 deletions(-) diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java index bbd94c59614,40a204aa9f3..348d262d90c --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java @@@ -142,6 -143,6 +143,7 @@@ import org.apache.iotdb.db.queryengine. import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExplainAnalyzeNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode; ++import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GroupNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.InformationSchemaTableScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.JoinNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LimitNode; @@@ -1360,6 -1362,6 +1363,19 @@@ public class TableOperatorGenerator ext TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber()); } ++ @Override ++ public Operator visitGroup(GroupNode node, LocalExecutionPlanContext context) { ++ StreamSortNode streamSortNode = ++ new StreamSortNode( ++ node.getPlanNodeId(), ++ node.getChild(), ++ node.getOrderingScheme(), ++ false, ++ false, ++ node.getPartitionKeyCount() - 1); ++ return visitStreamSort(streamSortNode, context); ++ } ++ @Override public Operator visitJoin(JoinNode node, LocalExecutionPlanContext context) { List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider()); diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java index 18b3f757d24,18b3f757d24..f6d90ce4a6b --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java @@@ -932,9 -932,9 +932,18 @@@ public class PlanGraphPrinter extends P public List<String> visitGroup(GroupNode node, GraphContext context) { List<String> boxValue = new ArrayList<>(); boxValue.add(String.format("GroupNode-%s", node.getPlanNodeId().getId())); -- boxValue.add(String.format("EnableParalleled: %s", node.isEnableParalleled())); -- boxValue.add(String.format("PartitionKeyCount: %s", node.getPartitionKeyCount())); -- boxValue.add(String.format("OrderingScheme: %s", node.getOrderingScheme())); ++ boxValue.add( ++ String.format( ++ "PartitionKey: %s", ++ node.getOrderingScheme().getOrderBy().subList(0, node.getPartitionKeyCount()))); ++ List<String> orderKey = new ArrayList<>(); ++ for (int i = node.getPartitionKeyCount(); ++ i < node.getOrderingScheme().getOrderBy().size(); ++ i++) { ++ Symbol symbol = node.getOrderingScheme().getOrderBy().get(i); ++ orderKey.add(symbol + " " + node.getOrderingScheme().getOrdering(symbol)); ++ } ++ boxValue.add(String.format("OrderKey: %s", orderKey)); return render(node, boxValue, context); } diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java index 5ddeb38bad8,cb279807d5a..3140f56f4ec --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java @@@ -291,32 -290,32 +290,26 @@@ public class TableDistributedPlanGenera public List<PlanNode> visitGroup(GroupNode node, PlanContext context) { boolean pushDown = context.isPushDownGrouping(); try { -- context.setPushDownGrouping(node.isEnableParalleled()); - // if (node.isEnableParalleled()) { - if (node.isEnableParalleled()) { -- List<PlanNode> result = new ArrayList<>(); -- context.setExpectedOrderingScheme(node.getOrderingScheme()); -- List<PlanNode> childrenNodes = node.getChild().accept(this, context); -- for (PlanNode child : childrenNodes) { -- if (canSortEliminated( -- node.getOrderingScheme(), nodeOrderingMap.get(child.getPlanNodeId()))) { -- result.add(child); -- } else { -- StreamSortNode subSortNode = -- new StreamSortNode( -- queryId.genPlanNodeId(), -- child, -- node.getOrderingScheme(), -- false, -- false, -- node.getPartitionKeyCount() - 1); -- result.add(subSortNode); - // nodeOrderingMap.put(subSortNode.getPlanNodeId(), subSortNode.getOrderingScheme()); - // should not set nodeOrderingMap here -- } ++ context.setPushDownGrouping(true); ++ List<PlanNode> result = new ArrayList<>(); ++ context.setExpectedOrderingScheme(node.getOrderingScheme()); ++ List<PlanNode> childrenNodes = node.getChild().accept(this, context); ++ for (PlanNode child : childrenNodes) { ++ if (canSortEliminated( ++ node.getOrderingScheme(), nodeOrderingMap.get(child.getPlanNodeId()))) { ++ result.add(child); ++ } else { ++ GroupNode subGroupNode = ++ new GroupNode( ++ queryId.genPlanNodeId(), ++ child, ++ node.getOrderingScheme(), ++ node.getPartitionKeyCount()); ++ result.add(subGroupNode); ++ // should not set nodeOrderingMap here } -- return result; - // } else { - // return visitSort(node, context); - // } - } else { - return visitSort(node, context); + } ++ return result; } finally { context.setPushDownGrouping(pushDown); } @@@ -1075,10 -1064,10 +1058,7 @@@ if (node.getChildren().isEmpty()) { return Collections.singletonList(node); } -- boolean canSplitPushDown = -- node.isRowSemantic() -- || (node.getChild() instanceof GroupNode) -- && ((GroupNode) node.getChild()).isEnableParalleled(); ++ boolean canSplitPushDown = node.isRowSemantic() || (node.getChild() instanceof GroupNode); List<PlanNode> childrenNodes = node.getChild().accept(this, context); if (childrenNodes.size() == 1) { node.setChild(childrenNodes.get(0)); diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/GroupNode.java index 4e9cb2b17bc,4e9cb2b17bc..80ea7657fff --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/GroupNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/GroupNode.java @@@ -51,45 -51,45 +51,15 @@@ public class GroupNode extends SortNod */ private int partitionKeyCount; -- /** GroupNode can be pushed down for paralleled execution. */ -- private boolean enableParalleled = false; -- public GroupNode(PlanNodeId id, PlanNode child, OrderingScheme scheme, int partitionKeyCount) { super(id, child, scheme, false, false); this.partitionKeyCount = partitionKeyCount; } -- public GroupNode( -- PlanNodeId id, -- PlanNode child, -- OrderingScheme scheme, -- boolean partial, -- boolean orderByAllIdsAndTime, -- boolean enableParalleled, -- int partitionKeyCount) { -- super(id, child, scheme, partial, orderByAllIdsAndTime); -- this.enableParalleled = enableParalleled; -- this.partitionKeyCount = partitionKeyCount; -- } -- @Override public PlanNode replaceChildren(List<PlanNode> newChildren) { return new GroupNode( -- id, -- Iterables.getOnlyElement(newChildren), -- orderingScheme, -- partial, -- orderByAllIdsAndTime, -- enableParalleled, -- partitionKeyCount); -- } -- -- public boolean isEnableParalleled() { -- return enableParalleled; -- } -- -- public void setEnableParalleled(boolean enableParalleled) { -- this.enableParalleled = enableParalleled; ++ id, Iterables.getOnlyElement(newChildren), orderingScheme, partitionKeyCount); } public int getPartitionKeyCount() { @@@ -103,21 -103,21 +73,13 @@@ @Override public PlanNode clone() { -- return new GroupNode( -- id, -- null, -- orderingScheme, -- partial, -- orderByAllIdsAndTime, -- enableParalleled, -- partitionKeyCount); ++ return new GroupNode(id, null, orderingScheme, partitionKeyCount); } @Override protected void serializeAttributes(ByteBuffer byteBuffer) { PlanNodeType.TABLE_GROUP_NODE.serialize(byteBuffer); orderingScheme.serialize(byteBuffer); -- ReadWriteIOUtils.write(enableParalleled, byteBuffer); ReadWriteIOUtils.write(partitionKeyCount, byteBuffer); } @@@ -125,17 -125,17 +87,14 @@@ protected void serializeAttributes(DataOutputStream stream) throws IOException { PlanNodeType.TABLE_GROUP_NODE.serialize(stream); orderingScheme.serialize(stream); -- ReadWriteIOUtils.write(enableParalleled, stream); ReadWriteIOUtils.write(partitionKeyCount, stream); } public static GroupNode deserialize(ByteBuffer byteBuffer) { OrderingScheme orderingScheme = OrderingScheme.deserialize(byteBuffer); -- boolean enableParalleled = ReadWriteIOUtils.readBoolean(byteBuffer); int partitionColumnCount = ReadWriteIOUtils.readInt(byteBuffer); PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer); -- return new GroupNode( -- planNodeId, null, orderingScheme, false, false, enableParalleled, partitionColumnCount); ++ return new GroupNode(planNodeId, null, orderingScheme, partitionColumnCount); } @Override diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGrouping.java index 79ef9d06140,79ef9d06140..c9c4af970c6 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGrouping.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGrouping.java @@@ -45,12 -45,12 +45,13 @@@ import java.util.Set import java.util.stream.Collectors; import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TAG; --import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.ParallelizeGrouping.CanParalleled.ENABLE; --import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.ParallelizeGrouping.CanParalleled.PENDING; --import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.ParallelizeGrouping.CanParalleled.UNABLE; ++import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.ParallelizeGrouping.CanOptimized.ELIMINATE; ++import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.ParallelizeGrouping.CanOptimized.KEEP_GROUP; ++import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.ParallelizeGrouping.CanOptimized.PENDING; ++import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.ParallelizeGrouping.CanOptimized.TO_SORT; /** -- * This rule is used to determine whether the GroupNode can be optimized during logical plan. ++ * This rule is used to determine whether the GroupNode can be executed parallel. * * <p>Optimization phase: Logical plan planning. * @@@ -68,8 -68,8 +69,9 @@@ * </ul> * * <ul> -- * The GroupNode can be transformed into a StreamSortNode if the lasted offspring that guarantees -- * the data is grouped by PartitionKey but not ordered by OrderKey. For example: ++ * The GroupNode should be kept and implemented as a StreamSortOperator later if the lasted ++ * offspring that guarantees the data is grouped by PartitionKey but not ordered by OrderKey. For ++ * example: * <ul> * <li>GroupNode[PK={device_id}, OK={s1}] -> ... -> TableDeviceScanNode * <li>GroupNode[PK={device_id,attr}, OK={s1}] -> ... -> TableDeviceScanNode @@@ -79,7 -79,7 +81,7 @@@ * </ul> * </ul> * -- * <p>Otherwise, the GroupNode cannot be optimized. It will be transformed into a SortNode. ++ * <p>Otherwise, the GroupNode should be transformed into a SortNode. */ public class ParallelizeGrouping implements PlanOptimizer { @Override @@@ -107,12 -107,12 +109,12 @@@ } /** We need to make sure: context#partitionKey can match the prefix of childOrderSchema */ -- private void checkPrefixMatch(Context context, List<Symbol> childOrder) { ++ private void checkForPartitionBasedNode(Context context, List<Symbol> childOrder) { if (context.canSkip()) { return; } if (context.partitionKeyCount > childOrder.size()) { -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; return; } OrderingScheme prefix = context.sortKey; @@@ -120,58 -120,58 +122,97 @@@ Symbol lhs = prefix.getOrderBy().get(i); Symbol rhs = childOrder.get(i); if (!lhs.equals(rhs)) { -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; return; } } -- context.canParalleled = ENABLE; ++ if (context.partitionKeyCount == prefix.getOrderBy().size()) { ++ context.canOptimized = ELIMINATE; ++ } else { ++ context.canOptimized = KEEP_GROUP; ++ } ++ } ++ ++ private void checkForSortBasedNode(Context context, OrderingScheme childOrder) { ++ if (context.canSkip()) { ++ return; ++ } ++ if (context.partitionKeyCount > childOrder.getOrderBy().size()) { ++ context.canOptimized = TO_SORT; ++ return; ++ } ++ OrderingScheme prefix = context.sortKey; ++ for (int i = 0; i < context.partitionKeyCount; i++) { ++ Symbol lhs = prefix.getOrderBy().get(i); ++ Symbol rhs = childOrder.getOrderBy().get(i); ++ if (!lhs.equals(rhs)) { ++ context.canOptimized = TO_SORT; ++ return; ++ } ++ } ++ for (int i = context.partitionKeyCount; i < prefix.getOrderBy().size(); i++) { ++ Symbol lhs = prefix.getOrderBy().get(i); ++ Symbol rhs = childOrder.getOrderBy().get(i); ++ if (!lhs.equals(rhs) || !prefix.getOrdering(lhs).equals(childOrder.getOrdering(lhs))) { ++ context.canOptimized = KEEP_GROUP; ++ return; ++ } ++ } ++ context.canOptimized = ELIMINATE; } @Override public PlanNode visitGroup(GroupNode node, Context context) { -- checkPrefixMatch(context, node.getOrderingScheme().getOrderBy()); ++ checkForPartitionBasedNode( ++ context, node.getOrderingScheme().getOrderBy().subList(0, node.getPartitionKeyCount())); Context newContext = new Context(node.getOrderingScheme(), node.getPartitionKeyCount()); -- GroupNode newNode = (GroupNode) node.clone(); -- newNode.addChild(node.getChild().accept(this, newContext)); -- if (newContext.canParalleled.equals(ENABLE)) { -- newNode.setEnableParalleled(true); ++ PlanNode child = node.getChild().accept(this, newContext); ++ switch (newContext.canOptimized) { ++ case ELIMINATE: ++ return child; ++ case KEEP_GROUP: ++ GroupNode newNode = (GroupNode) node.clone(); ++ newNode.addChild(child); ++ return newNode; ++ case TO_SORT: ++ default: ++ return new SortNode(node.getPlanNodeId(), child, node.getOrderingScheme(), false, false); } -- return newNode; } @Override public PlanNode visitSort(SortNode node, Context context) { -- checkPrefixMatch(context, node.getOrderingScheme().getOrderBy()); ++ checkForSortBasedNode(context, node.getOrderingScheme()); return visitPlan(node, context); } @Override public PlanNode visitStreamSort(StreamSortNode node, Context context) { -- checkPrefixMatch(context, node.getOrderingScheme().getOrderBy()); ++ checkForSortBasedNode(context, node.getOrderingScheme()); return visitPlan(node, context); } @Override public PlanNode visitTopK(TopKNode node, Context context) { -- checkPrefixMatch(context, node.getOrderingScheme().getOrderBy()); ++ checkForSortBasedNode(context, node.getOrderingScheme()); return visitPlan(node, context); } @Override public PlanNode visitJoin(JoinNode node, Context context) { -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; return visitPlan(node, context); } @Override public PlanNode visitCorrelatedJoin(CorrelatedJoinNode node, Context context) { -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; return visitPlan(node, context); } @Override public PlanNode visitSemiJoin(SemiJoinNode node, Context context) { -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; return visitPlan(node, context); } @@@ -180,15 -180,15 +221,15 @@@ if (!context.canSkip()) { if (node.getChildren().isEmpty()) { // leaf node -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; return node; } Optional<DataOrganizationSpecification> dataOrganizationSpecification = node.getDataOrganizationSpecification(); if (!dataOrganizationSpecification.isPresent()) { -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; } else { -- checkPrefixMatch(context, dataOrganizationSpecification.get().getPartitionBy()); ++ checkForPartitionBasedNode(context, dataOrganizationSpecification.get().getPartitionBy()); } } return visitPlan(node, context); @@@ -200,19 -200,19 +241,7 @@@ OrderingScheme sortKey = context.sortKey; Map<Symbol, ColumnSchema> tableColumnSchema = analysis.getTableColumnSchema(node.getQualifiedObjectName()); -- // // 1. It is possible for the last sort key to be a time column -- // if (sortKey.getOrderBy().size() > context.partitionKeyCount + 1) { -- // context.canParalleled = UNABLE; -- // return node; -- // } else if (sortKey.getOrderBy().size() == context.partitionKeyCount + 1) { -- // Symbol lastSymbol = sortKey.getOrderBy().get(context.partitionKeyCount); -- // if (!tableColumnSchema.containsKey(lastSymbol) -- // || tableColumnSchema.get(lastSymbol).getColumnCategory() != TIME) { -- // context.canParalleled = UNABLE; -- // return node; -- // } -- // } -- // 2. check there are no field in sortKey and all tags in sortKey ++ // check there are no field in sortKey and all tags in sortKey Set<Symbol> tagSymbols = tableColumnSchema.entrySet().stream() .filter(entry -> entry.getValue().getColumnCategory() == TAG) @@@ -221,7 -221,7 +250,7 @@@ for (int i = 0; i < context.partitionKeyCount; i++) { Symbol symbol = sortKey.getOrderBy().get(i); if (!tableColumnSchema.containsKey(symbol)) { -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; return node; } switch (tableColumnSchema.get(symbol).getColumnCategory()) { @@@ -232,28 -232,28 +261,28 @@@ // If all tags in partition key, attributes must be the same in one partition. break; default: -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; return node; } } if (!tagSymbols.isEmpty()) { -- context.canParalleled = UNABLE; ++ context.canOptimized = TO_SORT; return node; } -- context.canParalleled = ENABLE; ++ context.canOptimized = KEEP_GROUP; } return node; } @Override public PlanNode visitAggregation(AggregationNode node, Context context) { -- checkPrefixMatch(context, node.getGroupingKeys()); ++ checkForPartitionBasedNode(context, node.getGroupingKeys()); return visitPlan(node, context); } @Override public PlanNode visitAggregationTableScan(AggregationTableScanNode node, Context context) { -- checkPrefixMatch(context, node.getGroupingKeys()); ++ checkForPartitionBasedNode(context, node.getGroupingKeys()); return node; } } @@@ -261,7 -261,7 +290,7 @@@ private static class Context { private final OrderingScheme sortKey; private final int partitionKeyCount; -- private CanParalleled canParalleled = PENDING; ++ private CanOptimized canOptimized = PENDING; private Context(OrderingScheme sortKey, int partitionKeyCount) { this.sortKey = sortKey; @@@ -269,13 -269,13 +298,14 @@@ } private boolean canSkip() { -- return sortKey == null || canParalleled != PENDING; ++ return sortKey == null || canOptimized != PENDING; } } -- protected enum CanParalleled { -- ENABLE, -- UNABLE, ++ protected enum CanOptimized { ++ KEEP_GROUP, ++ ELIMINATE, ++ TO_SORT, PENDING } } diff --cc iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java index c449f26f23b,c449f26f23b..7cedc9d6281 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java @@@ -438,9 -438,9 +438,6 @@@ public class UnaliasSymbolReferences im node.getPlanNodeId(), rewrittenSource.getRoot(), newOrderingScheme, -- node.isPartial(), -- node.isOrderByAllIdsAndTime(), -- node.isEnableParalleled(), node.getPartitionKeyCount()), mapping); } diff --cc iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java index 7fca8e59fa0,7fca8e59fa0..71854d8b5be --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java @@@ -212,7 -212,7 +212,6 @@@ public class TableFunctionTest * │ └──TableScan * └──ExchangeNode * └──TableFunctionProcessor -- * └──streamSort * └──MergeSortNode * ├──ExchangeNode * │ └──TableScan diff --cc iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/CapacityTableFunction.java index d33188be8db,10cc00588b4..7fb40b37ce5 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/CapacityTableFunction.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/CapacityTableFunction.java @@@ -56,18 -85,24 +56,15 @@@ public class CapacityTableFunction impl } @Override - public TableFunctionAnalysis analyze(Map<String, Argument> arguments) { - TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME); - String expectedFieldName = - (String) ((ScalarArgument) arguments.get(TIMECOL_PARAMETER_NAME)).getValue(); - int requiredIndex = findTimeColumnIndex(tableArgument, expectedFieldName); - if (requiredIndex == -1) { - throw new UDFException("The required field is not found in the input table"); + public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws UDFException { + long size = (long) ((ScalarArgument) arguments.get(SIZE_PARAMETER_NAME)).getValue(); + if (size <= 0) { + throw new UDFException("Size must be greater than 0"); } - DescribedSchema properColumnSchema = - new DescribedSchema.Builder() - .addField("window_start", Type.TIMESTAMP) - .addField("window_end", Type.TIMESTAMP) - .build(); - - // outputColumnSchema return TableFunctionAnalysis.builder() - .properColumnSchema(properColumnSchema) - .requiredColumns(DATA_PARAMETER_NAME, Collections.singletonList(requiredIndex)) + .properColumnSchema( - new DescribedSchema.Builder() - .addField("window_index", Type.INT64) - .addField("count", Type.INT64) - .build()) ++ new DescribedSchema.Builder().addField("window_index", Type.INT64).build()) + .requiredColumns(DATA_PARAMETER_NAME, Collections.singletonList(0)) .build(); } @@@ -98,30 -137,17 +95,28 @@@ Record input, List<ColumnBuilder> properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { - // find the first windows that satisfy the condition: start + n*slide <= time < start + - // n*slide + size - long timeValue = input.getLong(0); - long window_start = (timeValue - start - size + slide) / slide * slide; - while (window_start <= timeValue && window_start + size > timeValue) { - properColumnBuilders.get(0).writeLong(window_start); - properColumnBuilders.get(1).writeLong(window_start + size - 1); - passThroughIndexBuilder.writeLong(curIndex); - window_start += slide; + if (currentRowIndexes.size() >= size) { + outputWindow(properColumnBuilders, passThroughIndexBuilder); } + currentRowIndexes.add(curIndex); curIndex++; } + + @Override + public void finish(List<ColumnBuilder> columnBuilders, ColumnBuilder passThroughIndexBuilder) { + if (!currentRowIndexes.isEmpty()) { + outputWindow(columnBuilders, passThroughIndexBuilder); + } + } + + private void outputWindow( + List<ColumnBuilder> properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { - int sz = currentRowIndexes.size(); + for (Long currentRowIndex : currentRowIndexes) { + properColumnBuilders.get(0).writeLong(windowIndex); - properColumnBuilders.get(1).writeLong(sz); + passThroughIndexBuilder.writeLong(currentRowIndex); + } + windowIndex++; + currentRowIndexes.clear(); + } } } diff --cc iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/SessionTableFunction.java index bd6b3186ce3,4e1afb7e0bd..570cec65c72 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/SessionTableFunction.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/SessionTableFunction.java @@@ -112,33 -136,22 +112,32 @@@ public class SessionTableFunction imple List<ColumnBuilder> properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { long timeValue = input.getLong(0); - if (curTime == Long.MIN_VALUE) { - curTime = timeValue; - } - if (curTime + size <= timeValue) { - // jump to appropriate window - long move = (timeValue - curTime - size) / slide + 1; - curTime += move * slide; + if (!currentRowIndexes.isEmpty() && timeValue > windowEnd) { + outputWindow(properColumnBuilders, passThroughIndexBuilder); - } - long slideTime = curTime; - while (slideTime <= timeValue && slideTime + size > timeValue) { - properColumnBuilders.get(0).writeLong(slideTime); - properColumnBuilders.get(1).writeLong(slideTime + size); - passThroughIndexBuilder.writeLong(curIndex); - slideTime += slide; + if (currentRowIndexes.isEmpty()) { + windowStart = timeValue; } + currentRowIndexes.add(curIndex); + windowEnd = timeValue + gap; curIndex++; } + + @Override + public void finish(List<ColumnBuilder> columnBuilders, ColumnBuilder passThroughIndexBuilder) { + if (!currentRowIndexes.isEmpty()) { + outputWindow(columnBuilders, passThroughIndexBuilder); + } + } + + private void outputWindow( + List<ColumnBuilder> properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { + for (Long currentRowIndex : currentRowIndexes) { + properColumnBuilders.get(0).writeLong(windowStart); - properColumnBuilders.get(1).writeLong(windowEnd - 1); ++ properColumnBuilders.get(1).writeLong(windowEnd - gap); + passThroughIndexBuilder.writeLong(currentRowIndex); + } + currentRowIndexes.clear(); + } } } diff --cc iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/VariationTableFunction.java index 06491b4e0ba,10cc00588b4..44db77235a0 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/VariationTableFunction.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/VariationTableFunction.java @@@ -64,21 -72,33 +64,18 @@@ public class VariationTableFunction imp .build()); } - private int findTimeColumnIndex(TableArgument tableArgument, String expectedFieldName) { - int requiredIndex = -1; - for (int i = 0; i < tableArgument.getFieldTypes().size(); i++) { - Optional<String> fieldName = tableArgument.getFieldNames().get(i); - if (fieldName.isPresent() && expectedFieldName.equalsIgnoreCase(fieldName.get())) { - requiredIndex = i; - break; - } - } - return requiredIndex; - } - @Override - public TableFunctionAnalysis analyze(Map<String, Argument> arguments) { + public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws UDFException { TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME); String expectedFieldName = - (String) ((ScalarArgument) arguments.get(TIMECOL_PARAMETER_NAME)).getValue(); - int requiredIndex = findTimeColumnIndex(tableArgument, expectedFieldName); - if (requiredIndex == -1) { - throw new UDFException("The required field is not found in the input table"); - } + (String) ((ScalarArgument) arguments.get(COL_PARAMETER_NAME)).getValue(); + int requiredIndex = + findColumnIndex( + tableArgument, + expectedFieldName, + ImmutableSet.of(Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE)); DescribedSchema properColumnSchema = -- new DescribedSchema.Builder() - .addField("window_index", Type.INT64) - .addField("base_value", Type.DOUBLE) - .addField("window_start", Type.TIMESTAMP) - .addField("window_end", Type.TIMESTAMP) -- .build(); - ++ new DescribedSchema.Builder().addField("window_index", Type.INT64).build(); // outputColumnSchema return TableFunctionAnalysis.builder() .properColumnSchema(properColumnSchema) @@@ -114,34 -137,17 +111,33 @@@ Record input, List<ColumnBuilder> properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { - // find the first windows that satisfy the condition: start + n*slide <= time < start + - // n*slide + size - long timeValue = input.getLong(0); - long window_start = (timeValue - start - size + slide) / slide * slide; - while (window_start <= timeValue && window_start + size > timeValue) { - properColumnBuilders.get(0).writeLong(window_start); - properColumnBuilders.get(1).writeLong(window_start + size - 1); - passThroughIndexBuilder.writeLong(curIndex); - window_start += slide; + double value = input.getDouble(0); + if (!currentRowIndexes.isEmpty() && Math.abs(value - baseValue) > gap) { + outputWindow(properColumnBuilders, passThroughIndexBuilder); } + if (currentRowIndexes.isEmpty()) { + // use the first value in the window as the base value + baseValue = value; + } + currentRowIndexes.add(curIndex); curIndex++; } + + @Override + public void finish(List<ColumnBuilder> columnBuilders, ColumnBuilder passThroughIndexBuilder) { + if (!currentRowIndexes.isEmpty()) { + outputWindow(columnBuilders, passThroughIndexBuilder); + } + } + + private void outputWindow( + List<ColumnBuilder> properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { + for (Long currentRowIndex : currentRowIndexes) { + properColumnBuilders.get(0).writeLong(windowIndex); - properColumnBuilders.get(1).writeDouble(baseValue); + passThroughIndexBuilder.writeLong(currentRowIndex); + } + currentRowIndexes.clear(); + windowIndex++; + } } }
