Weak Read Consistency Level Refactor & Print InternalAddress and port in explain analyze

Author: JackieTien97

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.commons.client.property.ClientPoolProperty.DefaultProperty;
 import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.commons.enums.ReadConsistencyLevel;
 import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.db.audit.AuditLogOperation;
@@ -975,7 +976,7 @@ public class IoTDBConfig {
   private long detailContainerMinDegradeMemoryInBytes = 1024 * 1024L;
   private int schemaThreadCount = 5;
 
-  private String readConsistencyLevel = "strong";
+  private ReadConsistencyLevel readConsistencyLevel = ReadConsistencyLevel.STRONG;
 
   /** Maximum execution time of a DriverTask */
   private int driverTaskExecutionTimeSliceInMs = 200;
@@ -3304,12 +3305,16 @@ public void setSchemaThreadCount(int schemaThreadCount) {
     this.schemaThreadCount = schemaThreadCount;
   }
 
-  public String getReadConsistencyLevel() {
+  public ReadConsistencyLevel getReadConsistencyLevel() {
     return readConsistencyLevel;
   }
 
   public void setReadConsistencyLevel(String readConsistencyLevel) {
-    this.readConsistencyLevel = readConsistencyLevel;
+    if ("weak".equalsIgnoreCase(readConsistencyLevel)) {
+      this.readConsistencyLevel = ReadConsistencyLevel.WEAK;
+    } else {
+      this.readConsistencyLevel = ReadConsistencyLevel.STRONG;
+    }
   }
 
   public int getDriverTaskExecutionTimeSliceInMs() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.queryengine.execution.fragment;
 
 import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
 import org.apache.iotdb.db.queryengine.exception.CpuNotEnoughException;
@@ -53,6 +54,7 @@
 public class FragmentInstanceExecution {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(FragmentInstanceExecution.class);
+  private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
   private final FragmentInstanceId instanceId;
   private final FragmentInstanceContext context;
 
@@ -152,7 +154,7 @@ private boolean fillFragmentInstanceStatistics(
       return false;
     }
     statistics.setDataRegion(((DataRegion) context.getDataRegion()).getDataRegionId());
-    statistics.setIp(IoTDBDescriptor.getInstance().getConfig().getAddressAndPort().ip);
+    statistics.setIp(CONFIG.getInternalAddress() + ":" + CONFIG.getInternalPort());
     statistics.setStartTimeInMS(context.getStartTime());
     statistics.setEndTimeInMS(
         context.isEndTimeUpdate() ? context.getEndTime() : System.currentTimeMillis());

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/AbstractFragmentParallelPlanner.java

@@ -22,9 +22,11 @@
 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.commons.enums.ReadConsistencyLevel;
 import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
 import org.apache.iotdb.commons.partition.QueryExecutor;
 import org.apache.iotdb.commons.partition.StorageExecutor;
+import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.queryengine.common.DataNodeEndPoints;
 import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
@@ -33,7 +35,6 @@
 import org.apache.iotdb.db.queryengine.plan.planner.exceptions.ReplicaSetUnreachableException;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.distribute.TableModelQueryFragmentPlanner;
 import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.commons.collections4.CollectionUtils;
@@ -45,24 +46,27 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
 
 public abstract class AbstractFragmentParallelPlanner implements IFragmentParallelPlaner {
   private static final Logger LOGGER =
-      LoggerFactory.getLogger(TableModelQueryFragmentPlanner.class);
+      LoggerFactory.getLogger(AbstractFragmentParallelPlanner.class);
+  private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
+  private final ReadConsistencyLevel readConsistencyLevel;
 
-  protected MPPQueryContext queryContext;
+  protected final MPPQueryContext queryContext;
 
   protected AbstractFragmentParallelPlanner(MPPQueryContext queryContext) {
     this.queryContext = queryContext;
+    this.readConsistencyLevel = CONFIG.getReadConsistencyLevel();
   }
 
   protected void selectExecutorAndHost(
       PlanFragment fragment,
       FragmentInstance fragmentInstance,
       Supplier<TRegionReplicaSet> replicaSetProvider,
-      Function<TRegionReplicaSet, TRegionReplicaSet> validator,
+      UnaryOperator<TRegionReplicaSet> validator,
       Map<TDataNodeLocation, List<FragmentInstance>> dataNodeFIMap) {
     // Get the target region for origin PlanFragment, then its instance will be distributed one
     // of them.
@@ -112,11 +116,7 @@ protected TDataNodeLocation selectTargetDataNode(TRegionReplicaSet regionReplica
       throw new IllegalArgumentException(
           String.format("regionReplicaSet is invalid: %s", regionReplicaSet));
     }
-    String readConsistencyLevel =
-        IoTDBDescriptor.getInstance().getConfig().getReadConsistencyLevel();
-    // TODO: (Chen Rongzhao) need to make the values of ReadConsistencyLevel as static variable or
-    // enums
-    boolean selectRandomDataNode = "weak".equals(readConsistencyLevel);
+    boolean selectRandomDataNode = ReadConsistencyLevel.WEAK == this.readConsistencyLevel;
 
     // When planning fragment onto specific DataNode, the DataNode whose endPoint is in
     // black list won't be considered because it may have connection issue now.

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/enums/ReadConsistencyLevel.java

@@ -0,0 +1,25 @@
+/*
+ * 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.commons.enums;
+
+public enum ReadConsistencyLevel {
+  STRONG,
+  WEAK
+}