Merge LastQueryScanNode of same device

Author: shuwenwei

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java

@@ -82,6 +82,13 @@ public class MPPQueryContext {
   // constructing some Expression and PlanNode.
   private final MemoryReservationManager memoryReservationManager;
 
+  private static final int minSizeToUseSampledTimeseriesOperandMemCost = 100;
+  private double avgTimeseriesOperandMemCost = 0;
+  private int numsOfSampledTimeseriesOperand = 0;
+  // When there is no view in a last query and no device exists in multiple regions,
+  // the updateScanNum process in distributed planning can be skipped.
+  private boolean needUpdateScanNumForLastQuery = false;
+
   private boolean userQuery = false;
 
   public MPPQueryContext(QueryId queryId) {
@@ -344,8 +351,30 @@ public void releaseMemoryReservedForFrontEnd(final long bytes) {
     this.memoryReservationManager.releaseMemoryCumulatively(bytes);
   }
 
+  public boolean useSampledAvgTimeseriesOperandMemCost() {
+    return numsOfSampledTimeseriesOperand >= minSizeToUseSampledTimeseriesOperandMemCost;
+  }
+
+  public long getAvgTimeseriesOperandMemCost() {
+    return (long) avgTimeseriesOperandMemCost;
+  }
+
+  public void calculateAvgTimeseriesOperandMemCost(long current) {
+    numsOfSampledTimeseriesOperand++;
+    avgTimeseriesOperandMemCost +=
+        (current - avgTimeseriesOperandMemCost) / numsOfSampledTimeseriesOperand;
+  }
+
   // endregion
 
+  public boolean needUpdateScanNumForLastQuery() {
+    return needUpdateScanNumForLastQuery;
+  }
+
+  public void setNeedUpdateScanNumForLastQuery(boolean needUpdateScanNumForLastQuery) {
+    this.needUpdateScanNumForLastQuery = needUpdateScanNumForLastQuery;
+  }
+
   public Optional<String> getDatabaseName() {
     return session.getDatabaseName();
   }

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

@@ -28,7 +28,9 @@
 
 import javax.annotation.Nullable;
 
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
 
 public class MemoryEstimationHelper {
 
@@ -41,6 +43,11 @@ public class MemoryEstimationHelper {
   private static final long MEASUREMENT_PATH_INSTANCE_SIZE =
       RamUsageEstimator.shallowSizeOfInstance(AlignedPath.class);
 
+  private static final long ARRAY_LIST_INSTANCE_SIZE =
+      RamUsageEstimator.shallowSizeOfInstance(ArrayList.class);
+  private static final long INTEGER_INSTANCE_SIZE =
+      RamUsageEstimator.shallowSizeOfInstance(Integer.class);
+
   private MemoryEstimationHelper() {
     // hide the constructor
   }
@@ -91,4 +98,25 @@ public static long getEstimatedSizeOfPartialPath(@Nullable final PartialPath par
     }
     return totalSize;
   }
+
+  // This method should only be called if the content in the current PartialPath comes from other
+  // structures whose memory cost have already been calculated.
+  public static long getEstimatedSizeOfCopiedPartialPath(@Nullable final PartialPath partialPath) {
+    if (partialPath == null) {
+      return 0;
+    }
+    return PARTIAL_PATH_INSTANCE_SIZE + RamUsageEstimator.shallowSizeOf(partialPath.getNodes());
+  }
+
+  public static long getEstimatedSizeOfIntegerArrayList(List<Integer> integerArrayList) {
+    if (integerArrayList == null) {
+      return 0L;
+    }
+    long size = ARRAY_LIST_INSTANCE_SIZE;
+    size +=
+        (long) RamUsageEstimator.NUM_BYTES_ARRAY_HEADER
+            + (long) integerArrayList.size() * (long) RamUsageEstimator.NUM_BYTES_OBJECT_REF;
+    size += INTEGER_INSTANCE_SIZE * integerArrayList.size();
+    return RamUsageEstimator.alignObjectSize(size);
+  }
 }

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

@@ -26,8 +26,8 @@
 
 import javax.annotation.concurrent.GuardedBy;
 
-import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -43,7 +43,7 @@ public class DataNodeQueryContext {
   private final ReentrantLock lock = new ReentrantLock();
 
   public DataNodeQueryContext(int dataNodeFINum) {
-    this.uncachedPathToSeriesScanInfo = new HashMap<>();
+    this.uncachedPathToSeriesScanInfo = new ConcurrentHashMap<>();
     this.dataNodeFINum = new AtomicInteger(dataNodeFINum);
   }
 
@@ -59,15 +59,24 @@ public Pair<AtomicInteger, TimeValuePair> getSeriesScanInfo(PartialPath path) {
     return uncachedPathToSeriesScanInfo.get(path);
   }
 
+  public Map<PartialPath, Pair<AtomicInteger, TimeValuePair>> getUncachedPathToSeriesScanInfo() {
+    return uncachedPathToSeriesScanInfo;
+  }
+
   public int decreaseDataNodeFINum() {
     return dataNodeFINum.decrementAndGet();
   }
 
-  public void lock() {
-    lock.lock();
+  public void lock(boolean isDeviceInMultiRegion) {
+    // When a device exists in only one region, there will be no intermediate state.
+    if (isDeviceInMultiRegion) {
+      lock.lock();
+    }
   }
 
-  public void unLock() {
-    lock.unlock();
+  public void unLock(boolean isDeviceInMultiRegion) {
+    if (isDeviceInMultiRegion) {
+      lock.unlock();
+    }
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/AbstractUpdateLastCacheOperator.java

@@ -64,12 +64,15 @@ public abstract class AbstractUpdateLastCacheOperator implements ProcessOperator
 
   protected String databaseName;
 
+  protected boolean deviceInMultiRegion;
+
   protected AbstractUpdateLastCacheOperator(
       final OperatorContext operatorContext,
       final Operator child,
       final TreeDeviceSchemaCacheManager treeDeviceSchemaCacheManager,
       final boolean needUpdateCache,
-      final boolean needUpdateNullEntry) {
+      final boolean needUpdateNullEntry,
+      final boolean deviceInMultiRegion) {
     this.operatorContext = operatorContext;
     this.child = child;
     this.lastCache = treeDeviceSchemaCacheManager;
@@ -78,6 +81,7 @@ protected AbstractUpdateLastCacheOperator(
     this.tsBlockBuilder = LastQueryUtil.createTsBlockBuilder(1);
     this.dataNodeQueryContext =
         operatorContext.getDriverContext().getFragmentInstanceContext().getDataNodeQueryContext();
+    this.deviceInMultiRegion = deviceInMultiRegion;
   }
 
   @Override
@@ -106,7 +110,7 @@ protected void mayUpdateLastCache(
       return;
     }
     try {
-      dataNodeQueryContext.lock();
+      dataNodeQueryContext.lock(deviceInMultiRegion);
       final Pair<AtomicInteger, TimeValuePair> seriesScanInfo =
           dataNodeQueryContext.getSeriesScanInfo(fullPath);
 
@@ -115,6 +119,20 @@ protected void mayUpdateLastCache(
         return;
       }
 
+      if (!deviceInMultiRegion) {
+        lastCache.updateLastCacheIfExists(
+            getDatabaseName(),
+            fullPath.getIDeviceID(),
+            new String[] {fullPath.getMeasurement()},
+            new TimeValuePair[] {
+              Objects.nonNull(value)
+                  ? new TimeValuePair(time, value)
+                  : needUpdateNullEntry ? TableDeviceLastCache.EMPTY_TIME_VALUE_PAIR : null
+            },
+            fullPath.isUnderAlignedEntity(),
+            new IMeasurementSchema[] {fullPath.getMeasurementSchema()});
+        return;
+      }
       // update cache in DataNodeQueryContext
       if (seriesScanInfo.right == null || time > seriesScanInfo.right.getTimestamp()) {
         if (Objects.nonNull(value)) {
@@ -135,7 +153,7 @@ protected void mayUpdateLastCache(
             new IMeasurementSchema[] {fullPath.getMeasurementSchema()});
       }
     } finally {
-      dataNodeQueryContext.unLock();
+      dataNodeQueryContext.unLock(deviceInMultiRegion);
     }
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/AlignedUpdateLastCacheOperator.java

@@ -47,9 +47,15 @@ public AlignedUpdateLastCacheOperator(
       AlignedPath seriesPath,
       TreeDeviceSchemaCacheManager treeDeviceSchemaCacheManager,
       boolean needUpdateCache,
-      boolean needUpdateNullEntry) {
+      boolean needUpdateNullEntry,
+      boolean deviceInMultiRegion) {
     super(
-        operatorContext, child, treeDeviceSchemaCacheManager, needUpdateCache, needUpdateNullEntry);
+        operatorContext,
+        child,
+        treeDeviceSchemaCacheManager,
+        needUpdateCache,
+        needUpdateNullEntry,
+        deviceInMultiRegion);
     this.seriesPath = seriesPath;
     this.devicePath = seriesPath.getDevicePath();
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/AlignedUpdateViewPathLastCacheOperator.java

@@ -41,14 +41,16 @@ public AlignedUpdateViewPathLastCacheOperator(
       TreeDeviceSchemaCacheManager treeDeviceSchemaCacheManager,
       boolean needUpdateCache,
       boolean needUpdateNullEntry,
-      String outputViewPath) {
+      String outputViewPath,
+      boolean deviceInMultiRegion) {
     super(
         operatorContext,
         child,
         seriesPath,
         treeDeviceSchemaCacheManager,
         needUpdateCache,
-        needUpdateNullEntry);
+        needUpdateNullEntry,
+        deviceInMultiRegion);
     checkArgument(seriesPath.getMeasurementList().size() == 1);
     this.outputViewPath = outputViewPath;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/LastQueryOperator.java

@@ -116,6 +116,7 @@ public TsBlock next() throws Exception {
           return null;
         } else if (!tsBlock.isEmpty()) {
           LastQueryUtil.appendLastValue(tsBlockBuilder, tsBlock);
+          return null;
         }
       } else {
         children.get(currentIndex).close();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/LastQuerySortOperator.java

@@ -153,7 +153,8 @@ private TsBlock buildResult() throws Exception {
 
     while (keepGoing(start, maxRuntime, endIndex)) {
 
-      if (prepareData()) {
+      prepareData();
+      if (previousTsBlock == null) {
         return null;
       }
 
@@ -179,21 +180,18 @@ private boolean keepGoing(long start, long maxRuntime, int endIndex) {
         && !tsBlockBuilder.isFull();
   }
 
-  private boolean prepareData() throws Exception {
+  private void prepareData() throws Exception {
     if (previousTsBlock == null || previousTsBlock.getPositionCount() <= previousTsBlockIndex) {
       if (children.get(currentIndex).hasNextWithTimer()) {
         previousTsBlock = children.get(currentIndex).nextWithTimer();
         previousTsBlockIndex = 0;
-        if (previousTsBlock == null) {
-          return true;
-        }
+        return;
       } else {
         children.get(currentIndex).close();
         children.set(currentIndex, null);
       }
       currentIndex++;
     }
-    return false;
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/last/UpdateLastCacheOperator.java

@@ -53,12 +53,33 @@ public UpdateLastCacheOperator(
       TreeDeviceSchemaCacheManager treeDeviceSchemaCacheManager,
       boolean needUpdateCache,
       boolean isNeedUpdateNullEntry) {
+    this(
+        operatorContext,
+        child,
+        fullPath,
+        dataType,
+        treeDeviceSchemaCacheManager,
+        needUpdateCache,
+        isNeedUpdateNullEntry,
+        true);
+  }
+
+  public UpdateLastCacheOperator(
+      OperatorContext operatorContext,
+      Operator child,
+      MeasurementPath fullPath,
+      TSDataType dataType,
+      TreeDeviceSchemaCacheManager treeDeviceSchemaCacheManager,
+      boolean needUpdateCache,
+      boolean isNeedUpdateNullEntry,
+      boolean deviceInMultiRegion) {
     super(
         operatorContext,
         child,
         treeDeviceSchemaCacheManager,
         needUpdateCache,
-        isNeedUpdateNullEntry);
+        isNeedUpdateNullEntry,
+        deviceInMultiRegion);
     this.fullPath = fullPath;
     this.dataType = dataType.name();
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java

@@ -114,6 +114,10 @@ public class Analysis implements IAnalysis {
   // map from device name to series/aggregation under this device
   private Set<Expression> sourceExpressions;
 
+  // In order to perform some optimization, when the source expression is
+  // not used later, nothing will be placed in this structure.
+  private boolean shouldHaveSourceExpression;
+
   // input expressions of aggregations to be calculated
   private Set<Expression> sourceTransformExpressions = new HashSet<>();
 
@@ -234,7 +238,9 @@ aggregation results last_value(temperature) and last_value(status), whereas buck
   // Key: non-writable view expression, Value: corresponding source expressions
   private Map<Expression, List<Expression>> lastQueryNonWritableViewSourceExpressionMap;
 
-  private Set<Expression> lastQueryBaseExpressions;
+  private Map<IDeviceID, Map<String, Expression>> lastQueryOutputPathToSourceExpressionMap;
+
+  private Set<IDeviceID> deviceExistViewSet;
 
   // header of result dataset
   private DatasetHeader respDatasetHeader;
@@ -619,6 +625,14 @@ public void setSourceExpressions(Set<Expression> sourceExpressions) {
     this.sourceExpressions = sourceExpressions;
   }
 
+  public void setShouldHaveSourceExpression(boolean shouldHaveSourceExpression) {
+    this.shouldHaveSourceExpression = shouldHaveSourceExpression;
+  }
+
+  public boolean shouldHaveSourceExpression() {
+    return shouldHaveSourceExpression;
+  }
+
   public Set<Expression> getSourceTransformExpressions() {
     return sourceTransformExpressions;
   }
@@ -886,12 +900,21 @@ public void setTimeseriesOrderingForLastQuery(Ordering timeseriesOrderingForLast
     this.timeseriesOrderingForLastQuery = timeseriesOrderingForLastQuery;
   }
 
-  public Set<Expression> getLastQueryBaseExpressions() {
-    return this.lastQueryBaseExpressions;
+  public Map<IDeviceID, Map<String, Expression>> getLastQueryOutputPathToSourceExpressionMap() {
+    return lastQueryOutputPathToSourceExpressionMap;
+  }
+
+  public void setLastQueryOutputPathToSourceExpressionMap(
+      Map<IDeviceID, Map<String, Expression>> lastQueryOutputPathToSourceExpressionMap) {
+    this.lastQueryOutputPathToSourceExpressionMap = lastQueryOutputPathToSourceExpressionMap;
+  }
+
+  public Set<IDeviceID> getDeviceExistViewSet() {
+    return deviceExistViewSet;
   }
 
-  public void setLastQueryBaseExpressions(Set<Expression> lastQueryBaseExpressions) {
-    this.lastQueryBaseExpressions = lastQueryBaseExpressions;
+  public void setDeviceExistViewSet(Set<IDeviceID> deviceExistViewSet) {
+    this.deviceExistViewSet = deviceExistViewSet;
   }
 
   public Map<Expression, List<Expression>> getLastQueryNonWritableViewSourceExpressionMap() {