This is an automated email from the ASF dual-hosted git repository. lancelly pushed a commit to branch pipelineMemoryControl in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4d7e3b1095f5bab855c1711405d26016109ad44d Author: lancelly <[email protected]> AuthorDate: Sat Apr 13 17:11:01 2024 +0800 Take all Drivers of one FragmentInstance into account when estimating memory usage of FI --- .../plan/planner/LocalExecutionPlanner.java | 72 +++++++-- .../plan/planner/OperatorTreeGenerator.java | 8 +- .../plan/planner/PipelineDriverFactory.java | 4 + .../db/queryengine/plan/planner/FEPlanUtil.java | 171 +++++++++++++++++++++ .../plan/planner/LocalExecutionPlannerTest.java | 141 +++++++++++++++++ .../plan/planner/PipelineBuilderTest.java | 129 +--------------- 6 files changed, 386 insertions(+), 139 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java index 6acd4e1b716..55fd149c8ef 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java @@ -40,9 +40,9 @@ import java.util.ArrayList; import java.util.List; /** - * Used to plan a fragment instance. Currently, we simply change it from PlanNode to executable - * Operator tree, but in the future, we may split one fragment instance into multiple pipeline to - * run a fragment instance parallel and take full advantage of multi-cores + * Used to plan a fragment instance. One fragment instance could be split into multiple pipelines so + * that a fragment instance could be run in parallel, and thus we can take full advantages of + * multi-cores. */ public class LocalExecutionPlanner { @@ -50,12 +50,17 @@ public class LocalExecutionPlanner { private static final long ALLOCATE_MEMORY_FOR_OPERATORS; private static final long MAX_REST_MEMORY_FOR_LOAD; + private static final long QUERY_THREAD_COUNT; + + private static final long ESTIMATED_FI_NUM = 8; + static { IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); ALLOCATE_MEMORY_FOR_OPERATORS = CONFIG.getAllocateMemoryForOperators(); MAX_REST_MEMORY_FOR_LOAD = (long) ((ALLOCATE_MEMORY_FOR_OPERATORS) * (1.0 - CONFIG.getMaxAllocateMemoryRatioForLoad())); + QUERY_THREAD_COUNT = CONFIG.getQueryThreadCount(); } /** allocated memory for operator execution */ @@ -82,10 +87,11 @@ public class LocalExecutionPlanner { // TODO Replace operator with operatorFactory to build multiple driver for one pipeline Operator root = plan.accept(new OperatorTreeGenerator(), context); - // check whether current free memory is enough to execute current query - long estimatedMemorySize = checkMemory(root, instanceContext.getStateMachine()); + context.addPipelineDriverFactory( + root, context.getDriverContext(), root.calculateMaxPeekMemory()); - context.addPipelineDriverFactory(root, context.getDriverContext(), estimatedMemorySize); + // check whether current free memory is enough to execute current query + checkMemory(context.getPipelineDriverFactories(), instanceContext.getStateMachine()); instanceContext.setSourcePaths(collectSourcePaths(context)); @@ -105,10 +111,11 @@ public class LocalExecutionPlanner { Operator root = plan.accept(new OperatorTreeGenerator(), context); - // check whether current free memory is enough to execute current query - checkMemory(root, instanceContext.getStateMachine()); + context.addPipelineDriverFactory( + root, context.getDriverContext(), root.calculateMaxPeekMemory()); - context.addPipelineDriverFactory(root, context.getDriverContext(), 0); + // check whether current free memory is enough to execute current query + checkMemory(context.getPipelineDriverFactories(), instanceContext.getStateMachine()); // set maxBytes one SourceHandle can reserve after visiting the whole tree context.setMaxBytesOneHandleCanReserve(); @@ -116,16 +123,18 @@ public class LocalExecutionPlanner { return context.getPipelineDriverFactories(); } - private long checkMemory(Operator root, FragmentInstanceStateMachine stateMachine) + private void checkMemory( + List<PipelineDriverFactory> pipelineDriverFactories, + FragmentInstanceStateMachine stateMachine) throws MemoryNotEnoughException { // if it is disabled, just return if (!IoTDBDescriptor.getInstance().getConfig().isEnableQueryMemoryEstimation() && !IoTDBDescriptor.getInstance().getConfig().isQuotaEnable()) { - return 0; + return; } - long estimatedMemorySize = root.calculateMaxPeekMemoryWithCounter(); + long estimatedMemorySize = calculateEstimatedMemorySize(pipelineDriverFactories); QueryRelatedResourceMetricSet.getInstance().updateEstimatedMemory(estimatedMemorySize); synchronized (this) { @@ -164,7 +173,44 @@ public class LocalExecutionPlanner { } } }); - return estimatedMemorySize; + } + + /** + * Calculate the estimated memory size this FI would use. Given that there are + * QUERY_THREAD_COUNT(X) threads, each thread could deal with one DriverTask. Now we have + * pipelineDriverFactories.size()(Y) DriverTasks(We use pipelineDriverFactories.size() instead of + * degreeOfParallelism to estimate). Suppose that there are M FragmentInstances at the same time. + * Then one FragmentInstance could have N Drivers running. + * + * <p>N = (X / ( M * Y)) * Y = X / M + * + * <p>The estimated running memory size this FI would use is: + * + * <p>N * avgMemoryUsedPerDriver = N * totalSizeOfDriver / driverNum + * + * <p>Some operators still retain memory when they are not running, so we need to add the size of + * retained memory of all the Drivers when they are not running. + * + * <p>The total estimated memory size this FI would use is: + * + * <p>retainedSize + N * totalSizeOfDriver / driverNum + */ + private long calculateEstimatedMemorySize( + final List<PipelineDriverFactory> pipelineDriverFactories) { + long retainedSize = + pipelineDriverFactories.stream() + .map( + pipelineDriverFactory -> + pipelineDriverFactory.getOperation().calculateRetainedSizeAfterCallingNext()) + .reduce(0L, Long::sum); + long totalSizeOfDrivers = + pipelineDriverFactories.stream() + .map(PipelineDriverFactory::getEstimatedMemorySize) + .reduce(0L, Long::sum); + long runningMemorySize = + Math.max((QUERY_THREAD_COUNT / ESTIMATED_FI_NUM), 1) + * (totalSizeOfDrivers / pipelineDriverFactories.size()); + return retainedSize + runningMemorySize; } private List<PartialPath> collectSourcePaths(LocalExecutionPlanContext context) { 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 a6d043f5ad2..0c290a81b7f 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 @@ -3098,7 +3098,8 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP // Attention, there is no parent node, use first child node instead subContext.getDriverContext(), childNode.getPlanNodeId().getId()); subContext.setISink(localSinkChannel); - subContext.addPipelineDriverFactory(childOperation, subContext.getDriverContext(), 0); + subContext.addPipelineDriverFactory( + childOperation, subContext.getDriverContext(), childOperation.calculateMaxPeekMemory()); ExchangeOperator sourceOperator = new ExchangeOperator( @@ -3156,7 +3157,10 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP // Attention, there is no parent node, use first child node instead context.getDriverContext(), childNode.getPlanNodeId().getId()); subContext.setISink(localSinkChannel); - subContext.addPipelineDriverFactory(childOperation, subContext.getDriverContext(), 0); + subContext.addPipelineDriverFactory( + childOperation, + subContext.getDriverContext(), + childOperation.calculateMaxPeekMemory()); // OneByOneChild may be divided into more than dop pipelines, but the number of running // actually is dop diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineDriverFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineDriverFactory.java index 1f5ef929fd2..38d19ef1e58 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineDriverFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineDriverFactory.java @@ -45,6 +45,10 @@ public class PipelineDriverFactory { this.estimatedMemorySize = estimatedMemorySize; } + public long getEstimatedMemorySize() { + return estimatedMemorySize; + } + public DriverContext getDriverContext() { return driverContext; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/FEPlanUtil.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/FEPlanUtil.java new file mode 100644 index 00000000000..02409befaaa --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/FEPlanUtil.java @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.queryengine.plan.planner; + +import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.path.AlignedPath; +import org.apache.iotdb.commons.path.MeasurementPath; +import org.apache.iotdb.db.queryengine.common.FragmentInstanceId; +import org.apache.iotdb.db.queryengine.common.PlanFragmentId; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.execution.fragment.DataNodeQueryContext; +import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; +import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine; +import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.DeviceViewNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDeviceViewNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.FullOuterTimeJoinNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.LeftOuterTimeJoinNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.AlignedSeriesScanNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesScanNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.OrderByParameter; +import org.apache.iotdb.db.queryengine.plan.statement.component.OrderByKey; +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.storageengine.dataregion.DataRegion; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; + +import org.mockito.Mockito; + +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.ExecutorService; + +import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext; +import static org.apache.iotdb.db.queryengine.plan.statement.component.OrderByKey.DEVICE; + +public class FEPlanUtil { + + private FEPlanUtil() { + // Util class does not need a Construct Method + } + + protected static LocalExecutionPlanContext createLocalExecutionPlanContext( + final TypeProvider typeProvider) { + ExecutorService instanceNotificationExecutor = + IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification"); + + QueryId queryId = new QueryId("stub_query"); + FragmentInstanceId instanceId = + new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance"); + FragmentInstanceStateMachine stateMachine = + new FragmentInstanceStateMachine(instanceId, instanceNotificationExecutor); + DataRegion dataRegion = Mockito.mock(DataRegion.class); + FragmentInstanceContext fragmentInstanceContext = + createFragmentInstanceContext(instanceId, stateMachine); + fragmentInstanceContext.setDataRegion(dataRegion); + + return new LocalExecutionPlanContext( + typeProvider, fragmentInstanceContext, new DataNodeQueryContext(1)); + } + + /** + * This method will init a FullOuterTimeJoinNode with @childNum seriesScanNode as children. + * + * @param childNum the number of children + * @return a timeJoinNode with @childNum seriesScanNode as children + */ + protected static FullOuterTimeJoinNode initFullOuterTimeJoinNode( + final TypeProvider typeProvider, final int childNum) throws IllegalPathException { + FullOuterTimeJoinNode fullOuterTimeJoinNode = + new FullOuterTimeJoinNode(new PlanNodeId("TimeJoinNode"), Ordering.ASC); + for (int i = 0; i < childNum; i++) { + SeriesScanNode seriesScanNode = + new SeriesScanNode( + new PlanNodeId(String.format("SeriesScanNode%d", i)), + new MeasurementPath(String.format("root.sg.d%d.s1", i), TSDataType.INT32)); + typeProvider.setType(seriesScanNode.getSeriesPath().toString(), TSDataType.INT32); + fullOuterTimeJoinNode.addChild(seriesScanNode); + } + return fullOuterTimeJoinNode; + } + + /** + * This method will init a DeviceViewNode with @childNum alignedSeriesScanNode as children. + * + * @param childNum the number of children + * @return a DeviceViewNode with @childNum alignedSeriesScanNode as children + */ + protected static DeviceViewNode initDeviceViewNode( + final TypeProvider typeProvider, final int childNum) throws IllegalPathException { + DeviceViewNode deviceViewNode = + new DeviceViewNode( + new PlanNodeId("DeviceViewNode"), null, Collections.EMPTY_LIST, Collections.EMPTY_MAP); + for (int i = 0; i < childNum; i++) { + AlignedSeriesScanNode alignedSeriesScanNode = + new AlignedSeriesScanNode( + new PlanNodeId(String.format("AlignedSeriesScanNode%d", i)), + new AlignedPath(String.format("root.sg.d%d", i), "s1")); + deviceViewNode.addChild(alignedSeriesScanNode); + } + return deviceViewNode; + } + + protected static LeftOuterTimeJoinNode initLeftOuterTimeJoinNode(TypeProvider typeProvider) + throws IllegalPathException { + LeftOuterTimeJoinNode leftOuterTimeJoinNode = + new LeftOuterTimeJoinNode(new PlanNodeId("TimeJoinNode"), Ordering.ASC); + for (int i = 0; i < 2; i++) { + SeriesScanNode seriesScanNode = + new SeriesScanNode( + new PlanNodeId(String.format("SeriesScanNode%d", i)), + new MeasurementPath(String.format("root.sg.d%d.s1", i), TSDataType.INT32)); + typeProvider.setType(seriesScanNode.getSeriesPath().toString(), TSDataType.INT32); + leftOuterTimeJoinNode.addChild(seriesScanNode); + } + return leftOuterTimeJoinNode; + } + + protected static TopKNode initTopKNode(TypeProvider typeProvider, int childNum) + throws IllegalPathException { + TopKNode topKNode = + new TopKNode( + new PlanNodeId("TopKNode"), + 10, + new OrderByParameter( + Arrays.asList( + new SortItem(OrderByKey.TIME, Ordering.ASC), + new SortItem(DEVICE, Ordering.ASC))), + Arrays.asList("Time", "Device", "s1")); + for (int i = 0; i < childNum; i++) { + SingleDeviceViewNode singleDeviceViewNode = + new SingleDeviceViewNode( + new PlanNodeId(String.format("SingleDeviceViewNode%d", i)), + Arrays.asList("Time", "Device", "s1"), + "root.sg.d" + i, + Arrays.asList(0, 1, 2)); + singleDeviceViewNode.setCacheOutputColumnNames(true); + SeriesScanNode seriesScanNode = + new SeriesScanNode( + new PlanNodeId(String.format("SeriesScanNode%d", i)), + new MeasurementPath(String.format("root.sg.d%d.s1", i), TSDataType.INT32)); + typeProvider.setType(seriesScanNode.getSeriesPath().toString(), TSDataType.INT32); + singleDeviceViewNode.addChild(seriesScanNode); + typeProvider.setType("Time", TSDataType.INT64); + typeProvider.setType("Device", TSDataType.TEXT); + typeProvider.setType("s1", TSDataType.INT32); + topKNode.addChild(singleDeviceViewNode); + } + return topKNode; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlannerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlannerTest.java new file mode 100644 index 00000000000..a0f528a32a2 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlannerTest.java @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.queryengine.plan.planner; + +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.db.queryengine.execution.operator.Operator; +import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.DeviceViewNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.FullOuterTimeJoinNode; +import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.List; + +import static org.apache.iotdb.db.queryengine.plan.planner.FEPlanUtil.createLocalExecutionPlanContext; +import static org.apache.iotdb.db.queryengine.plan.planner.FEPlanUtil.initDeviceViewNode; +import static org.apache.iotdb.db.queryengine.plan.planner.FEPlanUtil.initFullOuterTimeJoinNode; + +public class LocalExecutionPlannerTest { + + private static final long ESTIMATED_FI_NUM = 8; + + private static final long QUERY_THREAD_COUNT = 8; + + private static final long ALIGNED_MAX_SIZE = + TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes(); + + /** + * This test will test dop = 2. Expected result is five pipelines with dependency: + * + * <p>The first is: DeviceView - [ExchangeOperator, ExchangeOperator, ExchangeOperator, + * ExchangeOperator]; + * + * <p>The second is: ExchangeOperator - SeriesScan0. + * + * <p>The third is: ExchangeOperator - SeriesScan1, which has dependency second pipeline. + * + * <p>The forth is: ExchangeOperator - SeriesScan2, which has dependency third pipeline. + * + * <p>The fifth is: ExchangeOperator - SeriesScan3, which has dependency forth pipeline. + */ + @Test + public void testCheckMemoryWithDeviceView() { + try { + TypeProvider typeProvider = new TypeProvider(); + DeviceViewNode deviceViewNode = initDeviceViewNode(typeProvider, 4); + LocalExecutionPlanContext context = createLocalExecutionPlanContext(typeProvider); + context.setDegreeOfParallelism(2); + Operator root = deviceViewNode.accept(new OperatorTreeGenerator(), context); + + // all the four children pipelines are ScanOperator + context + .getPipelineDriverFactories() + .forEach( + pipelineDriverFactory -> + Assert.assertEquals( + ALIGNED_MAX_SIZE, pipelineDriverFactory.getEstimatedMemorySize())); + + context.addPipelineDriverFactory( + root, context.getDriverContext(), root.calculateMaxPeekMemory()); + Assert.assertEquals( + ALIGNED_MAX_SIZE, calculateEstimatedMemorySize(context.getPipelineDriverFactories())); + } catch (Exception e) { + Assert.fail(); + } + } + + /** + * This test will test dop = 2. Expected result is two pipelines: + * + * <p>The first is: TimeJoin1 - [SeriesScan1, SeriesScan0, ExchangeOperator]; + * + * <p>The second is: ExchangeOperator - TimeJoin1-1[SeriesScan2, SeriesScan3]. + */ + @Test + public void testCheckMemoryWithTimeJoin() throws IllegalPathException { + TypeProvider typeProvider = new TypeProvider(); + FullOuterTimeJoinNode timeJoinNode = initFullOuterTimeJoinNode(typeProvider, 4); + LocalExecutionPlanContext context = createLocalExecutionPlanContext(typeProvider); + context.setDegreeOfParallelism(2); + + Operator root = timeJoinNode.accept(new OperatorTreeGenerator(), context); + + // The second pipeline is a RowBasedTimeJoinOperator + Assert.assertEquals( + 2 * ALIGNED_MAX_SIZE, context.getPipelineDriverFactories().get(0).getEstimatedMemorySize()); + + context.addPipelineDriverFactory( + root, context.getDriverContext(), root.calculateMaxPeekMemory()); + + // ALIGNED_MAX_SIZE * 5 / 2 is calculated by directly applying the algorithm on the Operator + // Tree. + Assert.assertEquals( + ALIGNED_MAX_SIZE * 5 / 2 + + context.getPipelineDriverFactories().stream() + .map( + pipelineDriverFactory -> + pipelineDriverFactory + .getOperation() + .calculateRetainedSizeAfterCallingNext()) + .reduce(0L, Long::sum), + calculateEstimatedMemorySize(context.getPipelineDriverFactories())); + } + + private long calculateEstimatedMemorySize( + final List<PipelineDriverFactory> pipelineDriverFactories) { + long retainedSize = + pipelineDriverFactories.stream() + .map( + pipelineDriverFactory -> + pipelineDriverFactory.getOperation().calculateRetainedSizeAfterCallingNext()) + .reduce(0L, Long::sum); + long totalSizeOfDrivers = + pipelineDriverFactories.stream() + .map(PipelineDriverFactory::getEstimatedMemorySize) + .reduce(0L, Long::sum); + long runningMemorySize = + Math.max((QUERY_THREAD_COUNT / ESTIMATED_FI_NUM), 1) + * (totalSizeOfDrivers / pipelineDriverFactories.size()); + return retainedSize + runningMemorySize; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineBuilderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineBuilderTest.java index 2ec572ede00..c82eaf13d41 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineBuilderTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/PipelineBuilderTest.java @@ -21,17 +21,11 @@ package org.apache.iotdb.db.queryengine.plan.planner; import org.apache.iotdb.common.rpc.thrift.TAggregationType; import org.apache.iotdb.common.rpc.thrift.TEndPoint; -import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.exception.IllegalPathException; -import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.queryengine.common.FragmentInstanceId; import org.apache.iotdb.db.queryengine.common.PlanFragmentId; -import org.apache.iotdb.db.queryengine.common.QueryId; -import org.apache.iotdb.db.queryengine.execution.fragment.DataNodeQueryContext; -import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; -import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine; import org.apache.iotdb.db.queryengine.execution.operator.Operator; import org.apache.iotdb.db.queryengine.execution.operator.process.SingleDeviceViewOperator; import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanOperator; @@ -44,33 +38,27 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.AggregationNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.DeviceViewNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDeviceViewNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.FullOuterTimeJoinNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.LeftOuterTimeJoinNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.AlignedSeriesScanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesAggregationScanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesScanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.AggregationDescriptor; import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.AggregationStep; -import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.OrderByParameter; -import org.apache.iotdb.db.queryengine.plan.statement.component.OrderByKey; 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.storageengine.dataregion.DataRegion; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.junit.Test; -import org.mockito.Mockito; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.List; -import java.util.concurrent.ExecutorService; -import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext; -import static org.apache.iotdb.db.queryengine.plan.statement.component.OrderByKey.DEVICE; +import static org.apache.iotdb.db.queryengine.plan.planner.FEPlanUtil.createLocalExecutionPlanContext; +import static org.apache.iotdb.db.queryengine.plan.planner.FEPlanUtil.initDeviceViewNode; +import static org.apache.iotdb.db.queryengine.plan.planner.FEPlanUtil.initFullOuterTimeJoinNode; +import static org.apache.iotdb.db.queryengine.plan.planner.FEPlanUtil.initLeftOuterTimeJoinNode; +import static org.apache.iotdb.db.queryengine.plan.planner.FEPlanUtil.initTopKNode; import static org.junit.Assert.assertEquals; public class PipelineBuilderTest { @@ -1038,45 +1026,6 @@ public class PipelineBuilderTest { assertEquals(3, context.getExchangeSumNum()); } - private LocalExecutionPlanContext createLocalExecutionPlanContext(TypeProvider typeProvider) { - ExecutorService instanceNotificationExecutor = - IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification"); - - QueryId queryId = new QueryId("stub_query"); - FragmentInstanceId instanceId = - new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance"); - FragmentInstanceStateMachine stateMachine = - new FragmentInstanceStateMachine(instanceId, instanceNotificationExecutor); - DataRegion dataRegion = Mockito.mock(DataRegion.class); - FragmentInstanceContext fragmentInstanceContext = - createFragmentInstanceContext(instanceId, stateMachine); - fragmentInstanceContext.setDataRegion(dataRegion); - - return new LocalExecutionPlanContext( - typeProvider, fragmentInstanceContext, new DataNodeQueryContext(1)); - } - - /** - * This method will init a timeJoinNode with @childNum seriesScanNode as children. - * - * @param childNum the number of children - * @return a timeJoinNode with @childNum seriesScanNode as children - */ - private FullOuterTimeJoinNode initFullOuterTimeJoinNode(TypeProvider typeProvider, int childNum) - throws IllegalPathException { - FullOuterTimeJoinNode fullOuterTimeJoinNode = - new FullOuterTimeJoinNode(new PlanNodeId("TimeJoinNode"), Ordering.ASC); - for (int i = 0; i < childNum; i++) { - SeriesScanNode seriesScanNode = - new SeriesScanNode( - new PlanNodeId(String.format("SeriesScanNode%d", i)), - new MeasurementPath(String.format("root.sg.d%d.s1", i), TSDataType.INT32)); - typeProvider.setType(seriesScanNode.getSeriesPath().toString(), TSDataType.INT32); - fullOuterTimeJoinNode.addChild(seriesScanNode); - } - return fullOuterTimeJoinNode; - } - private FullOuterTimeJoinNode initFullOuterTimeJoinNodeWithExchangeNode( TypeProvider typeProvider, int exchangeNum, int scanNum) throws IllegalPathException { FullOuterTimeJoinNode fullOuterTimeJoinNode = @@ -1100,72 +1049,4 @@ public class PipelineBuilderTest { } return fullOuterTimeJoinNode; } - - private LeftOuterTimeJoinNode initLeftOuterTimeJoinNode(TypeProvider typeProvider) - throws IllegalPathException { - LeftOuterTimeJoinNode leftOuterTimeJoinNode = - new LeftOuterTimeJoinNode(new PlanNodeId("TimeJoinNode"), Ordering.ASC); - for (int i = 0; i < 2; i++) { - SeriesScanNode seriesScanNode = - new SeriesScanNode( - new PlanNodeId(String.format("SeriesScanNode%d", i)), - new MeasurementPath(String.format("root.sg.d%d.s1", i), TSDataType.INT32)); - typeProvider.setType(seriesScanNode.getSeriesPath().toString(), TSDataType.INT32); - leftOuterTimeJoinNode.addChild(seriesScanNode); - } - return leftOuterTimeJoinNode; - } - - /** - * This method will init a DeviceViewNode with @childNum alignedSeriesScanNode as children. - * - * @param childNum the number of children - * @return a DeviceViewNode with @childNum alignedSeriesScanNode as children - */ - private DeviceViewNode initDeviceViewNode(TypeProvider typeProvider, int childNum) - throws IllegalPathException { - DeviceViewNode deviceViewNode = - new DeviceViewNode(new PlanNodeId("DeviceViewNode"), null, null, null); - for (int i = 0; i < childNum; i++) { - AlignedSeriesScanNode alignedSeriesScanNode = - new AlignedSeriesScanNode( - new PlanNodeId(String.format("AlignedSeriesScanNode%d", i)), - new AlignedPath(String.format("root.sg.d%d", i), "s1")); - deviceViewNode.addChild(alignedSeriesScanNode); - } - return deviceViewNode; - } - - private TopKNode initTopKNode(TypeProvider typeProvider, int childNum) - throws IllegalPathException { - TopKNode topKNode = - new TopKNode( - new PlanNodeId("TopKNode"), - 10, - new OrderByParameter( - Arrays.asList( - new SortItem(OrderByKey.TIME, Ordering.ASC), - new SortItem(DEVICE, Ordering.ASC))), - Arrays.asList("Time", "Device", "s1")); - for (int i = 0; i < childNum; i++) { - SingleDeviceViewNode singleDeviceViewNode = - new SingleDeviceViewNode( - new PlanNodeId(String.format("SingleDeviceViewNode%d", i)), - Arrays.asList("Time", "Device", "s1"), - "root.sg.d" + i, - Arrays.asList(0, 1, 2)); - singleDeviceViewNode.setCacheOutputColumnNames(true); - SeriesScanNode seriesScanNode = - new SeriesScanNode( - new PlanNodeId(String.format("SeriesScanNode%d", i)), - new MeasurementPath(String.format("root.sg.d%d.s1", i), TSDataType.INT32)); - typeProvider.setType(seriesScanNode.getSeriesPath().toString(), TSDataType.INT32); - singleDeviceViewNode.addChild(seriesScanNode); - typeProvider.setType("Time", TSDataType.INT64); - typeProvider.setType("Device", TSDataType.TEXT); - typeProvider.setType("s1", TSDataType.INT32); - topKNode.addChild(singleDeviceViewNode); - } - return topKNode; - } }
