This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/fix_query_cycle
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/beyyes/fix_query_cycle by this
push:
new 4332a1cfd0c fix
4332a1cfd0c is described below
commit 4332a1cfd0c946dc938027a2901d9ba0f9da3b7b
Author: Beyyes <[email protected]>
AuthorDate: Wed Oct 18 21:06:23 2023 +0800
fix
---
.../distribution/DistributionPlanContext.java | 10 +-
.../plan/planner/distribution/SourceRewriter.java | 11 +-
.../distribution/DistributionPlannerCycleTest.java | 70 +++++
.../queryengine/plan/plan/distribution/Util2.java | 321 +++++++++++++++++++++
4 files changed, 406 insertions(+), 6 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanContext.java
index ce6b25c26dc..bc88fef1f69 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanContext.java
@@ -52,7 +52,15 @@ public class DistributionPlanContext {
return this;
}
- protected void setForceAddParent() {
+ public boolean isOneSeriesInMultiRegion() {
+ return oneSeriesInMultiRegion;
+ }
+
+ public boolean isForceAddParent() {
+ return this.forceAddParent;
+ }
+
+ public void setForceAddParent() {
this.forceAddParent = true;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
index f19d036822f..88b4526d1f7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
@@ -622,7 +622,7 @@ public class SourceRewriter extends
SimplePlanNodeRewriter<DistributionPlanConte
// For last query, we need to keep every FI's root node is
LastQueryMergeNode. So we
// force every region group have a parent node even if there is only 1
child for it.
context.setForceAddParent();
- PlanNode root = processRawMultiChildNode(node, context, true);
+ PlanNode root = processRawMultiChildNode(node, context, false);
if (context.queryMultiRegion) {
PlanNode newRoot = genLastQueryRootNode(node, context);
// add sort op for each if we add LastQueryMergeNode as root
@@ -694,11 +694,12 @@ public class SourceRewriter extends
SimplePlanNodeRewriter<DistributionPlanConte
if (containsAggregationSource(node)) {
return planAggregationWithTimeJoin(node, context);
}
- return Collections.singletonList(processRawMultiChildNode(node, context,
false));
+ return Collections.singletonList(processRawMultiChildNode(node, context,
true));
}
+ // Only `visitTimeJoin` and `visitLastQuery` invoke this method
private PlanNode processRawMultiChildNode(
- MultiChildProcessNode node, DistributionPlanContext context, boolean
isLast) {
+ MultiChildProcessNode node, DistributionPlanContext context, boolean
isTimeJoin) {
MultiChildProcessNode root = (MultiChildProcessNode) node.clone();
// Step 1: Get all source nodes. For the node which is not source, add it
as the child of
// current TimeJoinNode
@@ -740,7 +741,7 @@ public class SourceRewriter extends
SimplePlanNodeRewriter<DistributionPlanConte
// TODO: (xingtanzjr) optimize the procedure here to remove duplicated
TimeJoinNode
boolean addParent = false;
for (List<SourceNode> seriesScanNodes : sourceGroup.values()) {
- if (seriesScanNodes.size() == 1 && (!context.forceAddParent || !isLast))
{
+ if (seriesScanNodes.size() == 1 && (!context.isForceAddParent() ||
isTimeJoin)) {
root.addChild(seriesScanNodes.get(0));
continue;
}
@@ -750,7 +751,7 @@ public class SourceRewriter extends
SimplePlanNodeRewriter<DistributionPlanConte
// At last, we can use the parameter `addParent` to judge whether to
create new
// MultiChildNode.
boolean appendToRootDirectly =
- sourceGroup.size() == 1 || (!addParent && !context.forceAddParent);
+ sourceGroup.size() == 1 || (!addParent &&
!context.isForceAddParent());
if (appendToRootDirectly) {
seriesScanNodes.forEach(root::addChild);
addParent = true;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/plan/distribution/DistributionPlannerCycleTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/plan/distribution/DistributionPlannerCycleTest.java
new file mode 100644
index 00000000000..b764a032fd0
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/plan/distribution/DistributionPlannerCycleTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.plan.distribution;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.MeasurementPath;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
+import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
+import
org.apache.iotdb.db.queryengine.plan.planner.distribution.DistributionPlanner;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.SubPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.TimeSeriesSchemaScanNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.LimitNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesScanNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.junit.Assert.assertEquals;
+
+public class DistributionPlannerCycleTest {
+ @Test
+ public void aaTest() {
+ QueryId queryId = new QueryId("test");
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new
TEndPoint());
+
+ // 6. order by time, offset + limit
+ // on top of TOP-K NODE, LIMIT-NODE is necessary
+ String sql = "select * from root.sg.**";
+ Analysis analysis = Util2.analyze(sql, context);
+ PlanNode logicalPlanNode = Util2.genLogicalPlan(analysis, context);
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new
LogicalQueryPlan(context, logicalPlanNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ System.out.println("a");
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/plan/distribution/Util2.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/plan/distribution/Util2.java
new file mode 100644
index 00000000000..c78d0bdf855
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/plan/distribution/Util2.java
@@ -0,0 +1,321 @@
+/*
+ * 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.plan.distribution;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.commons.partition.DataPartition;
+import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
+import org.apache.iotdb.commons.partition.SchemaNodeManagementPartition;
+import org.apache.iotdb.commons.partition.SchemaPartition;
+import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.path.PathPatternTree;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
+import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree;
+import org.apache.iotdb.db.queryengine.common.schematree.node.SchemaEntityNode;
+import
org.apache.iotdb.db.queryengine.common.schematree.node.SchemaInternalNode;
+import
org.apache.iotdb.db.queryengine.common.schematree.node.SchemaMeasurementNode;
+import org.apache.iotdb.db.queryengine.common.schematree.node.SchemaNode;
+import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
+import org.apache.iotdb.db.queryengine.plan.analyze.Analyzer;
+import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaComputationWithAutoCreation;
+import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
+import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator;
+import org.apache.iotdb.db.queryengine.plan.planner.LogicalPlanner;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.statement.Statement;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement;
+import org.apache.iotdb.db.schemaengine.template.Template;
+import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+import org.mockito.Mockito;
+
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class Util2 {
+ public static final Analysis ANALYSIS = constructAnalysis();
+
+ public static Analysis constructAnalysis() {
+ SeriesPartitionExecutor executor =
+ SeriesPartitionExecutor.getSeriesPartitionExecutor(
+
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
+
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
+ Analysis analysis = new Analysis();
+
+ String device1 = "root.sg.d1";
+ String device2 = "root.sg.d22";
+ String device3 = "root.sg.d333";
+
+ TRegionReplicaSet dataRegion1 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 1),
+ Arrays.asList(genDataNodeLocation(11, "192.0.1.1")));
+
+ TRegionReplicaSet dataRegion2 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 2),
+ Arrays.asList(genDataNodeLocation(21, "192.0.1.1")));
+
+ DataPartition dataPartition =
+ new DataPartition(
+
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
+
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
+
+ Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot,
List<TRegionReplicaSet>>>>
+ dataPartitionMap = new HashMap<>();
+ Map<TSeriesPartitionSlot, Map<TTimePartitionSlot,
List<TRegionReplicaSet>>> sgPartitionMap =
+ new HashMap<>();
+
+ List<TRegionReplicaSet> d1DataRegions = new ArrayList<>();
+ d1DataRegions.add(dataRegion1);
+ Map<TTimePartitionSlot, List<TRegionReplicaSet>> d1DataRegionMap = new
HashMap<>();
+ d1DataRegionMap.put(new TTimePartitionSlot(), d1DataRegions);
+
+ List<TRegionReplicaSet> d2DataRegions = new ArrayList<>();
+ d2DataRegions.add(dataRegion2);
+ Map<TTimePartitionSlot, List<TRegionReplicaSet>> d2DataRegionMap = new
HashMap<>();
+ d2DataRegionMap.put(new TTimePartitionSlot(), d2DataRegions);
+
+ sgPartitionMap.put(executor.getSeriesPartitionSlot(device1),
d1DataRegionMap);
+ sgPartitionMap.put(executor.getSeriesPartitionSlot(device2),
d2DataRegionMap);
+
+ dataPartitionMap.put("root.sg", sgPartitionMap);
+
+ dataPartition.setDataPartitionMap(dataPartitionMap);
+
+ analysis.setDataPartitionInfo(dataPartition);
+
+ // construct schema partition
+ SchemaPartition schemaPartition =
+ new SchemaPartition(
+
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
+
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
+ Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>>
schemaPartitionMap = new HashMap<>();
+
+ TRegionReplicaSet schemaRegion1 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 11),
+ Arrays.asList(genDataNodeLocation(11, "192.0.1.1")));
+
+ TRegionReplicaSet schemaRegion2 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 21),
+ Arrays.asList(genDataNodeLocation(21, "192.0.1.1")));
+
+ Map<TSeriesPartitionSlot, TRegionReplicaSet> schemaRegionMap = new
HashMap<>();
+ schemaRegionMap.put(executor.getSeriesPartitionSlot(device1),
schemaRegion1);
+ schemaRegionMap.put(executor.getSeriesPartitionSlot(device2),
schemaRegion2);
+ schemaRegionMap.put(executor.getSeriesPartitionSlot(device3),
schemaRegion2);
+ schemaPartitionMap.put("root.sg", schemaRegionMap);
+ schemaPartition.setSchemaPartitionMap(schemaPartitionMap);
+
+ analysis.setDataPartitionInfo(dataPartition);
+ analysis.setSchemaPartitionInfo(schemaPartition);
+ analysis.setSchemaTree(genSchemaTree());
+ // to avoid some special case which is not the point of test
+ analysis.setStatement(Mockito.mock(QueryStatement.class));
+ Mockito.when(analysis.getStatement().isQuery()).thenReturn(false);
+ return analysis;
+ }
+
+ private static ISchemaTree genSchemaTree() {
+ SchemaNode root = new SchemaInternalNode("root");
+
+ SchemaNode sg = new SchemaInternalNode("sg");
+ root.addChild("sg", sg);
+
+ SchemaMeasurementNode s1 =
+ new SchemaMeasurementNode("s1", new MeasurementSchema("s1",
TSDataType.INT32));
+ SchemaMeasurementNode s2 =
+ new SchemaMeasurementNode("s2", new MeasurementSchema("s2",
TSDataType.DOUBLE));
+
+ SchemaMeasurementNode t1 =
+ new SchemaMeasurementNode("t1", new MeasurementSchema("t1",
TSDataType.INT32));
+ SchemaMeasurementNode t2 =
+ new SchemaMeasurementNode("t2", new MeasurementSchema("t2",
TSDataType.DOUBLE));
+ SchemaMeasurementNode t3 =
+ new SchemaMeasurementNode("t3", new MeasurementSchema("t3",
TSDataType.DOUBLE));
+
+ SchemaEntityNode d1 = new SchemaEntityNode("d1");
+ sg.addChild("d1", d1);
+ d1.addChild("s1", s1);
+ d1.addChild("s2", s2);
+
+ SchemaEntityNode d2 = new SchemaEntityNode("d22");
+ sg.addChild("d22", d2);
+ d2.addChild("t1", t1);
+ d2.addChild("t2", t2);
+ d2.addChild("t3", t3);
+
+ ClusterSchemaTree tree = new ClusterSchemaTree(root);
+ tree.setDatabases(Collections.singleton("root.sg"));
+
+ return tree;
+ }
+
+ public static Analysis analyze(String sql, MPPQueryContext context) {
+ Statement statement = StatementGenerator.createStatement(sql,
ZonedDateTime.now().getOffset());
+ Analyzer analyzer = new Analyzer(context, getFakePartitionFetcher(),
getFakeSchemaFetcher());
+ return analyzer.analyze(statement);
+ }
+
+ public static PlanNode genLogicalPlan(Analysis analysis, MPPQueryContext
context) {
+ LogicalPlanner planner = new LogicalPlanner(context, new ArrayList<>());
+ return planner.plan(analysis).getRootNode();
+ }
+
+ private static ISchemaFetcher getFakeSchemaFetcher() {
+ return new ISchemaFetcher() {
+ @Override
+ public ISchemaTree fetchSchema(PathPatternTree patternTree,
MPPQueryContext context) {
+ return ANALYSIS.getSchemaTree();
+ }
+
+ @Override
+ public ISchemaTree fetchSchemaWithTags(PathPatternTree patternTree,
MPPQueryContext context) {
+ return ANALYSIS.getSchemaTree();
+ }
+
+ @Override
+ public void fetchAndComputeSchemaWithAutoCreate(
+ ISchemaComputationWithAutoCreation schemaComputationWithAutoCreation,
+ MPPQueryContext context) {}
+
+ @Override
+ public void fetchAndComputeSchemaWithAutoCreate(
+ List<? extends ISchemaComputationWithAutoCreation>
schemaComputationWithAutoCreationList,
+ MPPQueryContext context) {}
+
+ @Override
+ public ISchemaTree fetchSchemaListWithAutoCreate(
+ List<PartialPath> devicePath,
+ List<String[]> measurements,
+ List<TSDataType[]> tsDataTypes,
+ List<TSEncoding[]> encodings,
+ List<CompressionType[]> compressionTypes,
+ List<Boolean> aligned,
+ MPPQueryContext context) {
+ return ANALYSIS.getSchemaTree();
+ }
+
+ @Override
+ public Pair<Template, PartialPath> checkTemplateSetInfo(PartialPath
devicePath) {
+ return null;
+ }
+
+ @Override
+ public Pair<Template, PartialPath> checkTemplateSetAndPreSetInfo(
+ PartialPath timeSeriesPath, String alias) {
+ return null;
+ }
+
+ @Override
+ public Map<Integer, Template> checkAllRelatedTemplate(PartialPath
pathPattern) {
+ return null;
+ }
+
+ @Override
+ public Pair<Template, List<PartialPath>> getAllPathsSetTemplate(String
templateName) {
+ return null;
+ }
+ };
+ }
+
+ private static IPartitionFetcher getFakePartitionFetcher() {
+ return new IPartitionFetcher() {
+ @Override
+ public SchemaPartition getSchemaPartition(PathPatternTree patternTree) {
+ return ANALYSIS.getSchemaPartitionInfo();
+ }
+
+ @Override
+ public SchemaPartition getOrCreateSchemaPartition(
+ PathPatternTree patternTree, String userName) {
+ return ANALYSIS.getSchemaPartitionInfo();
+ }
+
+ @Override
+ public DataPartition getDataPartition(
+ Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
+ return ANALYSIS.getDataPartitionInfo();
+ }
+
+ @Override
+ public DataPartition getDataPartitionWithUnclosedTimeRange(
+ Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
+ return ANALYSIS.getDataPartitionInfo();
+ }
+
+ @Override
+ public DataPartition getOrCreateDataPartition(
+ Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
+ return ANALYSIS.getDataPartitionInfo();
+ }
+
+ @Override
+ public DataPartition getOrCreateDataPartition(
+ List<DataPartitionQueryParam> dataPartitionQueryParams, String
userName) {
+ return ANALYSIS.getDataPartitionInfo();
+ }
+
+ @Override
+ public SchemaNodeManagementPartition
getSchemaNodeManagementPartitionWithLevel(
+ PathPatternTree patternTree, PathPatternTree scope, Integer level) {
+ return null;
+ }
+
+ @Override
+ public boolean updateRegionCache(TRegionRouteReq req) {
+ return false;
+ }
+
+ @Override
+ public void invalidAllCache() {}
+ };
+ }
+
+ private static TDataNodeLocation genDataNodeLocation(int dataNodeId, String
ip) {
+ return new TDataNodeLocation()
+ .setDataNodeId(dataNodeId)
+ .setClientRpcEndPoint(new TEndPoint(ip, 9000))
+ .setMPPDataExchangeEndPoint(new TEndPoint(ip, 9001))
+ .setInternalEndPoint(new TEndPoint(ip, 9002));
+ }
+}
