[IOTDB-6]Value filter query optimization

docs/Documentation/UserGuideV0.7.0/2-Concept.md

@@ -191,7 +191,6 @@ A column of data contains all values belonging to a time series and the timestam
 ## Data Type
 IoTDB supports six data types in total: BOOLEAN (Boolean), INT32 (Integer), INT64 (Long Integer), FLOAT (Single Precision Floating Point), DOUBLE (Double Precision Floating Point), TEXT (String).
 
-
 The time series of FLOAT and DOUBLE type can specify (MAX\_POINT\_NUMBER, see [this page](#iotdb-query-statement) for more information on how to specify), which is the number of digits after the decimal point of the floating point number, if the encoding method is [RLE](#encoding) or [TS\_2DIFF](#encoding) (Refer to [Create Timeseries Statement](#chapter-5-iotdb-sql-documentation) for more information on how to specify). If MAX\_POINT\_NUMBER is not specified, the system will use [float\_precision](#encoding) in the configuration file "tsfile-format.properties" for configuration for the configuration method.
 
 * For Float data value, The data range is (-Integer.MAX_VALUE, Integer.MAX_VALUE), rather than Float.MAX_VALUE, and the max_point_number is 19, it is because of the limition of function Math.round(float) in Java.
@@ -248,4 +247,4 @@ The four encodings described in the previous sections are applicable to differen
 
 When the time series is written and encoded as binary data according to the specified type, IoTDB compresses the data using compression technology to further improve space storage efficiency. Although both encoding and compression are designed to improve storage efficiency, encoding techniques are usually only available for specific data types (e.g., second-order differential encoding is only suitable for INT32 or INT64 data type, and storing floating-point numbers requires multiplying them by 10m to convert to integers), after which the data is converted to a binary stream. The compression method (SNAPPY) compresses the binary stream, so the use of the compression method is no longer limited by the data type.
 
-IoTDB allows you to specify the compression method of the column when creating a time series. IoTDB now supports two kinds of compression: UNCOMPRESSED (no compression) and SNAPPY compression. The specified syntax for compression is detailed in [Create Timeseries Statement](#chapter-5-iotdb-sql-documentation).
+IoTDB allows you to specify the compression method of the column when creating a time series. IoTDB now supports two kinds of compression: UNCOMPRESSED (no compression) and SNAPPY compression. The specified syntax for compression is detailed in [Create Timeseries Statement](#chapter-5-iotdb-sql-documentation).

iotdb/src/main/java/org/apache/iotdb/db/query/dataset/EngineDataSetWithTimeGenerator.java

@@ -35,6 +35,7 @@ public class EngineDataSetWithTimeGenerator extends QueryDataSet {
 
   private EngineTimeGenerator timeGenerator;
   private List<EngineReaderByTimeStamp> readers;
+  //TODO add null filed
 
   /**
    * constructor of EngineDataSetWithTimeGenerator.
@@ -62,11 +63,11 @@ public RowRecord next() throws IOException {
     RowRecord rowRecord = new RowRecord(timestamp);
     for (int i = 0; i < readers.size(); i++) {
       EngineReaderByTimeStamp reader = readers.get(i);
-      TsPrimitiveType tsPrimitiveType = reader.getValueInTimestamp(timestamp);
-      if (tsPrimitiveType == null) {
+      Object value = reader.getValueInTimestamp(timestamp);
+      if (value == null) {
         rowRecord.addField(new Field(null));
       } else {
-        rowRecord.addField(getField(tsPrimitiveType.getValue(), dataTypes.get(i)));
+        rowRecord.addField(getField(value, dataTypes.get(i)));
       }
     }
 

iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineExecutorWithTimeGenerator.java

@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.apache.iotdb.db.query.executor;
 
 import java.io.IOException;
@@ -31,9 +32,8 @@
 import org.apache.iotdb.db.query.dataset.EngineDataSetWithTimeGenerator;
 import org.apache.iotdb.db.query.factory.SeriesReaderFactory;
 import org.apache.iotdb.db.query.reader.merge.EngineReaderByTimeStamp;
-import org.apache.iotdb.db.query.reader.merge.PriorityMergeReader;
 import org.apache.iotdb.db.query.reader.merge.PriorityMergeReaderByTimestamp;
-import org.apache.iotdb.db.query.reader.sequence.SequenceDataReader;
+import org.apache.iotdb.db.query.reader.sequence.SequenceDataReaderByTimestamp;
 import org.apache.iotdb.db.query.timegenerator.EngineTimeGenerator;
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 import org.apache.iotdb.tsfile.read.common.Path;
@@ -106,13 +106,13 @@ private List<EngineReaderByTimeStamp> getReadersOfSelectedPaths(List<Path> paths
       PriorityMergeReaderByTimestamp mergeReaderByTimestamp = new PriorityMergeReaderByTimestamp();
 
       // reader for sequence data
-      SequenceDataReader tsFilesReader = new SequenceDataReader(queryDataSource.getSeqDataSource(),
-          null, context);
+      SequenceDataReaderByTimestamp tsFilesReader = new SequenceDataReaderByTimestamp(
+          queryDataSource.getSeqDataSource(), context);
       mergeReaderByTimestamp.addReaderWithPriority(tsFilesReader, 1);
 
       // reader for unSequence data
-      PriorityMergeReader unSeqMergeReader = SeriesReaderFactory.getInstance()
-          .createUnSeqMergeReader(queryDataSource.getOverflowSeriesDataSource(), null);
+      PriorityMergeReaderByTimestamp unSeqMergeReader = SeriesReaderFactory.getInstance()
+          .createUnSeqMergeReaderByTimestamp(queryDataSource.getOverflowSeriesDataSource());
       mergeReaderByTimestamp.addReaderWithPriority(unSeqMergeReader, 2);
 
       readersOfSelectedSeries.add(mergeReaderByTimestamp);

iotdb/src/main/java/org/apache/iotdb/db/query/factory/SeriesReaderFactory.java

@@ -28,11 +28,14 @@
 import org.apache.iotdb.db.query.context.QueryContext;
 import org.apache.iotdb.db.query.control.FileReaderManager;
 import org.apache.iotdb.db.query.reader.IReader;
+import org.apache.iotdb.db.query.reader.mem.MemChunkReaderByTimestamp;
 import org.apache.iotdb.db.query.reader.mem.MemChunkReaderWithFilter;
 import org.apache.iotdb.db.query.reader.mem.MemChunkReaderWithoutFilter;
 import org.apache.iotdb.db.query.reader.merge.PriorityMergeReader;
+import org.apache.iotdb.db.query.reader.merge.PriorityMergeReaderByTimestamp;
 import org.apache.iotdb.db.query.reader.sequence.SealedTsFilesReader;
 import org.apache.iotdb.db.query.reader.unsequence.EngineChunkReader;
+import org.apache.iotdb.db.query.reader.unsequence.EngineChunkReaderByTimestamp;
 import org.apache.iotdb.db.utils.QueryUtils;
 import org.apache.iotdb.tsfile.common.constant.StatisticConstant;
 import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
@@ -45,6 +48,7 @@
 import org.apache.iotdb.tsfile.read.filter.DigestForFilter;
 import org.apache.iotdb.tsfile.read.filter.basic.Filter;
 import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderByTimestamp;
 import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithFilter;
 import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithoutFilter;
 import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReader;
@@ -125,7 +129,50 @@ public PriorityMergeReader createUnSeqMergeReader(
     return unSeqMergeReader;
   }
 
-  // TODO createUnSeqMergeReaderByTime a method with filter
+  /**
+   * This method is used to create unsequence insert reader by timestamp for IoTDB request, such as
+   * query, aggregation and groupby request.
+   */
+  public PriorityMergeReaderByTimestamp createUnSeqMergeReaderByTimestamp(
+      OverflowSeriesDataSource overflowSeriesDataSource)
+      throws IOException {
+
+    PriorityMergeReaderByTimestamp unSeqMergeReader = new PriorityMergeReaderByTimestamp();
+
+    int priorityValue = 1;
+
+    for (OverflowInsertFile overflowInsertFile : overflowSeriesDataSource
+        .getOverflowInsertFileList()) {
+
+      // store only one opened file stream into manager, to avoid too many opened files
+      TsFileSequenceReader unClosedTsFileReader = FileReaderManager.getInstance()
+          .get(overflowInsertFile.getFilePath(), false);
+
+      ChunkLoaderImpl chunkLoader = new ChunkLoaderImpl(unClosedTsFileReader);
+
+      for (ChunkMetaData chunkMetaData : overflowInsertFile.getChunkMetaDataList()) {
+
+        Chunk chunk = chunkLoader.getChunk(chunkMetaData);
+        ChunkReaderByTimestamp chunkReader = new ChunkReaderByTimestamp(chunk);
+
+        unSeqMergeReader
+            .addReaderWithPriority(
+                new EngineChunkReaderByTimestamp(chunkReader, unClosedTsFileReader),
+                priorityValue);
+        priorityValue++;
+      }
+    }
+
+    // add reader for MemTable
+    if (overflowSeriesDataSource.hasRawChunk()) {
+      unSeqMergeReader.addReaderWithPriority(
+          new MemChunkReaderByTimestamp(overflowSeriesDataSource.getReadableMemChunk()),
+          priorityValue);
+    }
+
+    // TODO add external sort when needed
+    return unSeqMergeReader;
+  }
 
   /**
    * This method is used to construct reader for merge process in IoTDB. To merge only one TsFile

iotdb/src/main/java/org/apache/iotdb/db/query/reader/mem/MemChunkReaderByTimestamp.java

@@ -16,17 +16,18 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.apache.iotdb.db.query.reader.mem;
 
 import java.io.IOException;
 import java.util.Iterator;
 import org.apache.iotdb.db.engine.memtable.TimeValuePairSorter;
+import org.apache.iotdb.db.query.reader.IReader;
 import org.apache.iotdb.db.query.reader.merge.EngineReaderByTimeStamp;
 import org.apache.iotdb.db.utils.TimeValuePair;
-import org.apache.iotdb.db.utils.TsPrimitiveType;
 import org.apache.iotdb.tsfile.read.common.BatchData;
 
-public class MemChunkReaderByTimestamp implements EngineReaderByTimeStamp {
+public class MemChunkReaderByTimestamp implements EngineReaderByTimeStamp, IReader {
 
   private Iterator<TimeValuePair> timeValuePairIterator;
   private boolean hasCachedTimeValuePair;
@@ -65,15 +66,16 @@ public void close() {
   }
 
   // TODO consider change timeValuePairIterator to List structure, and use binary search instead of
-  // sequential search
-  @Override
-  public TsPrimitiveType getValueInTimestamp(long timestamp) throws IOException {
+  /**
+   * sequential search
+   */
+  @Override public Object getValueInTimestamp(long timestamp) throws IOException {
     while (hasNext()) {
       TimeValuePair timeValuePair = next();
-      long time = timeValuePair.getTimestamp();
-      if (time == timestamp) {
-        return timeValuePair.getValue();
-      } else if (time > timestamp) {
+      long currentMemTime = timeValuePair.getTimestamp();
+      if (currentMemTime == timestamp) {
+        return timeValuePair.getValue().getValue();
+      } else if (currentMemTime > timestamp) {
         hasCachedTimeValuePair = true;
         cachedTimeValuePair = timeValuePair;
         break;

iotdb/src/main/java/org/apache/iotdb/db/query/reader/mem/MemChunkReaderWithoutFilter.java

@@ -24,7 +24,9 @@
 import org.apache.iotdb.db.utils.TimeValuePair;
 import org.apache.iotdb.tsfile.read.common.BatchData;
 
-// TODO merge MemChunkReaderWithoutFilter and MemChunkReaderWithFilter to one class
+/**
+ * TODO merge MemChunkReaderWithoutFilter and MemChunkReaderWithFilter to one class
+ */
 public class MemChunkReaderWithoutFilter implements IReader {
 
   private Iterator<TimeValuePair> timeValuePairIterator;

iotdb/src/main/java/org/apache/iotdb/db/query/reader/merge/EngineReaderByTimeStamp.java

@@ -16,17 +16,20 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.apache.iotdb.db.query.reader.merge;
 
 import java.io.IOException;
-import org.apache.iotdb.db.query.reader.IReader;
-import org.apache.iotdb.db.utils.TsPrimitiveType;
 
-public interface EngineReaderByTimeStamp extends IReader {
+public interface EngineReaderByTimeStamp {
 
   /**
    * Given a timestamp, the reader is supposed to return the corresponding value in the timestamp.
    * If no value in this timestamp, null will be returned.
    */
-  TsPrimitiveType getValueInTimestamp(long timestamp) throws IOException;
+  Object getValueInTimestamp(long timestamp) throws IOException;
+
+  boolean hasNext() throws IOException;
+
+  void close() throws IOException;
 }

iotdb/src/main/java/org/apache/iotdb/db/query/reader/merge/PriorityMergeReaderByTimestamp.java

@@ -16,44 +16,61 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.apache.iotdb.db.query.reader.merge;
 
 import java.io.IOException;
-import org.apache.iotdb.db.utils.TimeValuePair;
-import org.apache.iotdb.db.utils.TsPrimitiveType;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
- * TODO the process of PriorityMergeReaderByTimestamp can be optimized.
+ * <p>
+ * Usage: Get value in timestamp by sorting time-value pair in multiple readers with time and
+ * priority. (1) merge multiple chunk group readers in the unsequence file (2）merge sequence reader,
+ * unsequence reader and mem reader
+ * </p>
  */
-public class PriorityMergeReaderByTimestamp extends PriorityMergeReader implements
-    EngineReaderByTimeStamp {
+public class PriorityMergeReaderByTimestamp implements EngineReaderByTimeStamp {
+
+  private List<EngineReaderByTimeStamp> readerList = new ArrayList<>();
+  private List<Integer> priorityList = new ArrayList<>();
 
-  private boolean hasCachedTimeValuePair;
-  private TimeValuePair cachedTimeValuePair;
+  /**
+   * This function doesn't sort reader by priority. So you have to call this function in order of
+   * reader priority from small to large.
+   */
+  public void addReaderWithPriority(EngineReaderByTimeStamp reader, int priority) {
+    readerList.add(reader);
+    priorityList.add(priority);
+  }
 
   @Override
-  public TsPrimitiveType getValueInTimestamp(long timestamp) throws IOException {
-
-    if (hasCachedTimeValuePair) {
-      if (cachedTimeValuePair.getTimestamp() == timestamp) {
-        hasCachedTimeValuePair = false;
-        return cachedTimeValuePair.getValue();
-      } else if (cachedTimeValuePair.getTimestamp() > timestamp) {
-        return null;
+  public Object getValueInTimestamp(long timestamp) throws IOException {
+    Object value = null;
+    for (int i = readerList.size() - 1; i >= 0; i--) {
+      value = readerList.get(i).getValueInTimestamp(timestamp);
+      if (value != null) {
+        return value;
       }
     }
+    return value;
+  }
 
-    while (hasNext()) {
-      cachedTimeValuePair = next();
-      if (cachedTimeValuePair.getTimestamp() == timestamp) {
-        hasCachedTimeValuePair = false;
-        return cachedTimeValuePair.getValue();
-      } else if (cachedTimeValuePair.getTimestamp() > timestamp) {
-        hasCachedTimeValuePair = true;
-        return null;
-      }
+  @Override
+  public void close() throws IOException {
+    for (EngineReaderByTimeStamp reader : readerList) {
+      reader.close();
     }
+  }
 
-    return null;
+  @Override
+  public boolean hasNext() throws IOException {
+    for (int i = readerList.size() - 1; i >= 0; i--) {
+      if (readerList.get(i).hasNext()) {
+        return true;
+      }
+    }
+    return false;
   }
+
 }