Optimize memtable scan

Author: shuwenwei

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

@@ -57,6 +57,7 @@
 import org.apache.iotdb.mpp.rpc.thrift.TFetchFragmentInstanceStatisticsResp;
 
 import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.TimeRange;
 import org.apache.tsfile.read.filter.basic.Filter;
 import org.apache.tsfile.read.filter.factory.FilterFactory;
 import org.apache.tsfile.utils.RamUsageEstimator;
@@ -66,6 +67,7 @@
 import java.time.ZoneId;
 import java.time.format.DateTimeParseException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -99,6 +101,7 @@ public class FragmentInstanceContext extends QueryContext {
 
   protected IDataRegionForQuery dataRegion;
   private Filter globalTimeFilter;
+  private List<TimeRange> globalTimeFilterTimeRanges;
 
   // it will only be used once, after sharedQueryDataSource being inited, it will be set to null
   private List<IFullPath> sourcePaths;
@@ -517,6 +520,23 @@ public Filter getGlobalTimeFilter() {
     return globalTimeFilter;
   }
 
+  public List<TimeRange> getGlobalTimeFilterTimeRanges() {
+    if (globalTimeFilter == null) {
+      return Collections.singletonList(new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE));
+    }
+    List<TimeRange> local = globalTimeFilterTimeRanges;
+    if (local == null) {
+      synchronized (this) {
+        local = globalTimeFilterTimeRanges;
+        if (local == null) {
+          local = globalTimeFilter.getTimeRanges();
+          globalTimeFilterTimeRanges = local;
+        }
+      }
+    }
+    return local;
+  }
+
   public void setTimeFilterForTableModel(Filter timeFilter) {
     if (globalTimeFilter == null) {
       globalTimeFilter = timeFilter;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java

@@ -35,6 +35,7 @@
 import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemPageReader;
 import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.DescPriorityMergeReader;
 import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.MergeReaderPriority;
+import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.NoDataPointReader;
 import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.PriorityMergeReader;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.utils.datastructure.MemPointIterator;
@@ -134,7 +135,10 @@ public class SeriesScanUtil implements Accountable {
           + RamUsageEstimator.shallowSizeOfInstance(IDeviceID.class)
           + RamUsageEstimator.shallowSizeOfInstance(TimeOrderUtils.class)
           + RamUsageEstimator.shallowSizeOfInstance(PaginationController.class)
-          + RamUsageEstimator.shallowSizeOfInstance(SeriesScanOptions.class);
+          + RamUsageEstimator.shallowSizeOfInstance(SeriesScanOptions.class)
+          + RamUsageEstimator.shallowSizeOfInstance(TimeRange.class);
+
+  protected TimeRange satisfiedTimeRange;
 
   public SeriesScanUtil(
       IFullPath seriesPath,
@@ -214,6 +218,20 @@ public void initQueryDataSource(QueryDataSource dataSource) {
     // init file index
     orderUtils.setCurSeqFileIndex(dataSource);
     curUnseqFileIndex = 0;
+
+    if (satisfiedTimeRange == null) {
+      long startTime = Long.MAX_VALUE;
+      long endTime = Long.MIN_VALUE;
+      if (scanOptions.getGlobalTimeFilter() == null) {
+        satisfiedTimeRange = new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE);
+        return;
+      }
+      for (TimeRange timeRange : context.getGlobalTimeFilterTimeRanges()) {
+        startTime = Math.min(startTime, timeRange.getMin());
+        endTime = Math.max(endTime, timeRange.getMax());
+      }
+      satisfiedTimeRange = new TimeRange(startTime, endTime);
+    }
   }
 
   protected PriorityMergeReader getPriorityMergeReader() {
@@ -681,6 +699,13 @@ private void unpackOneFakeMemChunkMetaData(
         readOnlyMemChunk.createMemPointIterator(
             orderUtils.getScanOrder(), scanOptions.getGlobalTimeFilter());
     for (Statistics<? extends Serializable> statistics : statisticsList) {
+      long orderTime = orderUtils.getOrderTime(statistics);
+      boolean canSkip =
+          (orderUtils.getAscending() && orderTime > satisfiedTimeRange.getMax())
+              || (!orderUtils.getAscending() && orderTime < satisfiedTimeRange.getMin());
+      if (canSkip) {
+        break;
+      }
       IVersionPageReader versionPageReader =
           new LazyMemVersionPageReader(
               context,
@@ -1458,6 +1483,7 @@ protected static class LazyMemVersionPageReader implements IVersionPageReader {
     protected final boolean isSeq;
     protected final boolean isAligned;
     private boolean inited = false;
+    private boolean hasData = true;
 
     LazyMemVersionPageReader(
         QueryContext context,
@@ -1477,11 +1503,14 @@ protected static class LazyMemVersionPageReader implements IVersionPageReader {
     }
 
     public IPointReader getPointReader() {
+      if (!hasData) {
+        return NoDataPointReader.getInstance();
+      }
       return memPointIterator;
     }
 
     public boolean hasNextBatch() {
-      return memPointIterator.hasNextBatch();
+      return hasData && memPointIterator.hasNextBatch();
     }
 
     public void setCurrentPageTimeRangeToMemPointIterator() {
@@ -1490,10 +1519,34 @@ public void setCurrentPageTimeRangeToMemPointIterator() {
       }
       if (statistics.getStartTime() > statistics.getEndTime()) {
         // empty
+        hasData = false;
         return;
       }
-      this.memPointIterator.setCurrentPageTimeRange(
-          new TimeRange(statistics.getStartTime(), statistics.getEndTime()));
+      Filter globalTimeFilter = ((FragmentInstanceContext) context).getGlobalTimeFilter();
+      if (globalTimeFilter == null) {
+        this.memPointIterator.setCurrentPageTimeRange(
+            new TimeRange(statistics.getStartTime(), statistics.getEndTime()));
+        return;
+      }
+
+      long startTime = statistics.getStartTime();
+      long endTime = statistics.getEndTime();
+      long minStart = Long.MAX_VALUE;
+      long maxEnd = Long.MIN_VALUE;
+      for (TimeRange timeRange :
+          ((FragmentInstanceContext) context).getGlobalTimeFilterTimeRanges()) {
+        if (timeRange.overlaps(new TimeRange(startTime, endTime))) {
+          minStart = Math.min(minStart, Math.max(timeRange.getMin(), startTime));
+          maxEnd = Math.max(maxEnd, Math.min(timeRange.getMax(), endTime));
+        }
+      }
+
+      if (minStart > maxEnd) {
+        hasData = false;
+        return;
+      }
+
+      this.memPointIterator.setCurrentPageTimeRange(new TimeRange(minStart, maxEnd));
     }
 
     public TsBlock nextBatch() {
@@ -1629,7 +1682,7 @@ public long getOrderTime(Statistics statistics) {
     @SuppressWarnings("squid:S3740")
     @Override
     public long getOverlapCheckTime(Statistics range) {
-      return range.getStartTime();
+      return Math.max(satisfiedTimeRange.getMin(), range.getStartTime());
     }
 
     @SuppressWarnings("squid:S3740")
@@ -1758,7 +1811,7 @@ public long getOrderTime(Statistics statistics) {
     @SuppressWarnings("squid:S3740")
     @Override
     public long getOverlapCheckTime(Statistics range) {
-      return range.getEndTime();
+      return Math.min(satisfiedTimeRange.getMax(), range.getEndTime());
     }
 
     @SuppressWarnings("squid:S3740")

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/NoDataPointReader.java

@@ -0,0 +1,59 @@
+/*
+ * 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.storageengine.dataregion.read.reader.common;
+
+import org.apache.tsfile.read.TimeValuePair;
+import org.apache.tsfile.read.reader.IPointReader;
+
+import java.io.IOException;
+
+public class NoDataPointReader implements IPointReader {
+
+  private NoDataPointReader() {}
+
+  private static final IPointReader instance = new NoDataPointReader();
+
+  public static IPointReader getInstance() {
+    return instance;
+  }
+
+  @Override
+  public boolean hasNextTimeValuePair() throws IOException {
+    return false;
+  }
+
+  @Override
+  public TimeValuePair nextTimeValuePair() throws IOException {
+    return null;
+  }
+
+  @Override
+  public TimeValuePair currentTimeValuePair() throws IOException {
+    return null;
+  }
+
+  @Override
+  public long getUsedMemorySize() {
+    return 0;
+  }
+
+  @Override
+  public void close() throws IOException {}
+}