Implement all fragment instance sharing dop in one dataNode
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceContext.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceContext.java index f810bc1..b299dd9 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceContext.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceContext.java
@@ -20,6 +20,7 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.engine.querycontext.QueryDataSource; import org.apache.iotdb.db.engine.storagegroup.IDataRegionForQuery; import org.apache.iotdb.db.engine.storagegroup.TsFileResource; @@ -27,6 +28,7 @@ import org.apache.iotdb.db.metadata.idtable.IDTable; import org.apache.iotdb.db.mpp.common.FragmentInstanceId; import org.apache.iotdb.db.mpp.common.SessionInfo; +import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance; import org.apache.iotdb.db.query.context.QueryContext; import org.apache.iotdb.db.query.control.FileReaderManager; import org.apache.iotdb.tsfile.read.filter.basic.Filter; @@ -75,6 +77,8 @@ // session info private SessionInfo sessionInfo; + private int degreeOfParallelism = 1; + // private final GcMonitor gcMonitor; // private final AtomicLong startNanos = new AtomicLong(); // private final AtomicLong startFullGcCount = new AtomicLong(-1); @@ -84,9 +88,12 @@ // private final AtomicLong endFullGcTimeNanos = new AtomicLong(-1); public static FragmentInstanceContext createFragmentInstanceContext( - FragmentInstanceId id, FragmentInstanceStateMachine stateMachine, SessionInfo sessionInfo) { + FragmentInstanceId id, + FragmentInstanceStateMachine stateMachine, + FragmentInstance fragmentInstance) { FragmentInstanceContext instanceContext = - new FragmentInstanceContext(id, stateMachine, sessionInfo); + new FragmentInstanceContext(id, stateMachine, fragmentInstance.getSessionInfo()); + instanceContext.setDegreeOfParallelism(calculateDegreeOfParallelism(fragmentInstance)); instanceContext.initialize(); instanceContext.start(); return instanceContext; @@ -95,11 +102,16 @@ public static FragmentInstanceContext createFragmentInstanceContext( FragmentInstanceId id, FragmentInstanceStateMachine stateMachine, - SessionInfo sessionInfo, IDataRegionForQuery dataRegion, - Filter timeFilter) { + FragmentInstance fragmentInstance) { FragmentInstanceContext instanceContext = - new FragmentInstanceContext(id, stateMachine, sessionInfo, dataRegion, timeFilter); + new FragmentInstanceContext( + id, + stateMachine, + fragmentInstance.getSessionInfo(), + dataRegion, + fragmentInstance.getTimeFilter()); + instanceContext.setDegreeOfParallelism(calculateDegreeOfParallelism(fragmentInstance)); instanceContext.initialize(); instanceContext.start(); return instanceContext; @@ -368,4 +380,24 @@ sourcePaths = null; sharedQueryDataSource = null; } + + private static int calculateDegreeOfParallelism(FragmentInstance fragmentInstance) { + int systemDop = IoTDBDescriptor.getInstance().getConfig().getDegreeOfParallelism(); + int instanceNumInDataNode = fragmentInstance.getInstanceNumInDataNode(); + if (instanceNumInDataNode >= systemDop) { + return 1; + } else if (fragmentInstance.isRoot()) { + return systemDop / instanceNumInDataNode + systemDop % instanceNumInDataNode; + } else { + return systemDop / instanceNumInDataNode; + } + } + + public void setDegreeOfParallelism(int degreeOfParallelism) { + this.degreeOfParallelism = degreeOfParallelism; + } + + public int getDegreeOfParallelism() { + return degreeOfParallelism; + } }
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceManager.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceManager.java index 6e5390f..47fbd43 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceManager.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/fragment/FragmentInstanceManager.java
@@ -123,11 +123,7 @@ instanceId, fragmentInstanceId -> createFragmentInstanceContext( - fragmentInstanceId, - stateMachine, - instance.getSessionInfo(), - dataRegion, - instance.getTimeFilter())); + fragmentInstanceId, stateMachine, dataRegion, instance)); try { List<PipelineDriverFactory> driverFactories = @@ -190,7 +186,7 @@ instanceId, fragmentInstanceId -> createFragmentInstanceContext( - fragmentInstanceId, stateMachine, instance.getSessionInfo())); + fragmentInstanceId, stateMachine, instance)); try { List<PipelineDriverFactory> driverFactories =
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanContext.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanContext.java index 3dfde55..f6e4eb4 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanContext.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanContext.java
@@ -89,6 +89,7 @@ this.nextPipelineId = new AtomicInteger(0); this.driverContext = new DataDriverContext(instanceContext, getNextPipelineId()); this.pipelineDriverFactories = new ArrayList<>(); + this.degreeOfParallelism = instanceContext.getDegreeOfParallelism(); } // For creating subContext, differ from parent context mainly in driver context @@ -119,6 +120,7 @@ this.dataRegionTTL = Long.MAX_VALUE; this.driverContext = new SchemaDriverContext(instanceContext, schemaRegion, getNextPipelineId()); + this.degreeOfParallelism = instanceContext.getDegreeOfParallelism(); this.pipelineDriverFactories = new ArrayList<>(); }
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java index 6361a91..9ccf34b 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java
@@ -57,15 +57,17 @@ public class SimpleFragmentParallelPlanner implements IFragmentParallelPlaner { private static final Logger logger = LoggerFactory.getLogger(SimpleFragmentParallelPlanner.class); - private SubPlan subPlan; - private Analysis analysis; - private MPPQueryContext queryContext; + private final SubPlan subPlan; + private final Analysis analysis; + private final MPPQueryContext queryContext; // Record all the FragmentInstances belonged to same PlanFragment - Map<PlanFragmentId, FragmentInstance> instanceMap; + private final Map<PlanFragmentId, FragmentInstance> instanceMap; // Record which PlanFragment the PlanNode belongs - Map<PlanNodeId, Pair<PlanFragmentId, PlanNode>> planNodeMap; - List<FragmentInstance> fragmentInstanceList; + private final Map<PlanNodeId, Pair<PlanFragmentId, PlanNode>> planNodeMap; + private final List<FragmentInstance> fragmentInstanceList; + // Record which dataNode each instance allocated to + private final Map<TDataNodeLocation, List<FragmentInstance>> locationToInstanceMap; public SimpleFragmentParallelPlanner( SubPlan subPlan, Analysis analysis, MPPQueryContext context) { @@ -75,6 +77,7 @@ this.instanceMap = new HashMap<>(); this.planNodeMap = new HashMap<>(); this.fragmentInstanceList = new ArrayList<>(); + this.locationToInstanceMap = new HashMap<>(); } @Override @@ -90,6 +93,13 @@ recordPlanNodeRelation(fragment.getPlanNodeTree(), fragment.getId()); produceFragmentInstance(fragment); } + for (Map.Entry<TDataNodeLocation, List<FragmentInstance>> entry : + locationToInstanceMap.entrySet()) { + List<FragmentInstance> fragmentInstances = entry.getValue(); + int instanceNumInDataNode = fragmentInstances.size(); + fragmentInstances.forEach( + instance -> instance.setInstanceNumInDataNode(instanceNumInDataNode)); + } } private void produceFragmentInstance(PlanFragment fragment) { @@ -137,6 +147,9 @@ } instanceMap.putIfAbsent(fragment.getId(), fragmentInstance); fragmentInstanceList.add(fragmentInstance); + locationToInstanceMap + .computeIfAbsent(fragmentInstance.getHostDataNode(), x -> new ArrayList<>()) + .add(fragmentInstance); } private TDataNodeLocation selectTargetDataNode(TRegionReplicaSet regionReplicaSet) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/FragmentInstance.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/FragmentInstance.java index 97bd7e1..310b3f7 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/FragmentInstance.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/FragmentInstance.java
@@ -68,6 +68,8 @@ private final SessionInfo sessionInfo; + private int instanceNumInDataNode; + // We can add some more params for a specific FragmentInstance // So that we can make different FragmentInstance owns different data range. @@ -126,6 +128,14 @@ return executorType.getRegionReplicaSet(); } + public void setInstanceNumInDataNode(int instanceNumInDataNode) { + this.instanceNumInDataNode = instanceNumInDataNode; + } + + public int getInstanceNumInDataNode() { + return instanceNumInDataNode; + } + public PlanFragment getFragment() { return fragment; } @@ -183,6 +193,7 @@ boolean hasHostDataNode = ReadWriteIOUtils.readBool(buffer); fragmentInstance.hostDataNode = hasHostDataNode ? ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(buffer) : null; + fragmentInstance.setInstanceNumInDataNode(ReadWriteIOUtils.readInt(buffer)); return fragmentInstance; } @@ -205,6 +216,7 @@ if (hostDataNode != null) { ThriftCommonsSerDeUtils.serializeTDataNodeLocation(hostDataNode, outputStream); } + ReadWriteIOUtils.write(instanceNumInDataNode, outputStream); return ByteBuffer.wrap(publicBAOS.getBuf(), 0, publicBAOS.size()); } catch (IOException e) { logger.error("Unexpected error occurs when serializing this FragmentInstance.", e);