[IOTDB-418] add some impl of NewSeriesReaderWithoutValueFilter by liutaohua · Pull Request #692 · apache/iotdb

liutaohua · 2019-12-30T06:37:44Z

The process of processing read data is a very complicated one, and previous read interfaces were obscure and confusing.
I reimplemented a reader based on which all the data in the files could be queried in order

I have mainly done the following work:

Renamed iAggregateReader in TsFile project to IPageSkipRader
Added IChunkReader. Based on this interface, there are two implementations, ChunkReader and MemChunkReader
Added ChunkMetadata for ReadonlyMemChunk and PageHeader for MemChunkReader
Added IAggregateReader and IRawReader interface and AbstractDataReader abstract class to complete all functions of the new reader, respectively realizing RawDataReaderWithoutValueFilter, SeriesDataReaderWithoutValueFilter and SeriesDataReaderWithValueFilter
Refactoring group by, raw query and aggregate query ,and so on
In the previous group by implementation, the open and closed interval definition was wrong, [0-100] I changed it to [0-100)

qiaojialin · 2019-12-30T11:09:34Z

...n/java/org/apache/iotdb/db/query/reader/seriesRelated/NewSeriesReaderWithoutValueFilter.java

+    Path seriesPath = queryDataSource.getSeriesPath();
+    TreeSet<TsFileResource> unseqTsFilesSet = new TreeSet<>((o1, o2) -> {
+      String queryMeasurement = seriesPath.getMeasurement();
+      List<Long> o1StartTimeList = o1.getChunkMetaDataList().stream()


This ChunkMetadataList should all belongs to this measurement. No need to filter again. Besides, only un closed TsFile has the chunkMetaDataList.

The sort could compare by the startTimeMap.get(seriesPath.getDeviceId())

…or-iotdb into new_series_reader � Conflicts: � server/src/main/java/org/apache/iotdb/db/query/executor/EngineQueryRouter.java � server/src/main/java/org/apache/iotdb/db/query/reader/chunkRelated/DiskChunkReader.java � tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/chunk/AbstractChunkReader.java

…or-iotdb into new_series_reader � Conflicts: � server/src/main/java/org/apache/iotdb/db/query/dataset/NewEngineDataSetWithoutValueFilter.java

jixuan1989 · 2020-01-11T02:20:35Z

please try to merge master to check whether the failed test can pass.

liutaohua · 2020-01-11T02:24:49Z

please try to merge master to check whether the failed test can pass.

please help me merge master to IoTDB new_series_reader

…or-iotdb into new_series_reader

samperson1997

Hi, thanks a lot for your effort and codes. You'd really done an excellent and fantastic job in this PR! I believe it will be quite great contribution after being merged.
After an initial quick view of changed files, I have found some tiny little problems. Most of them are just suggestions and you can decide whether to fix them or not. Afterwards, I will try to deeply review the logistic of all the codes.
Look forward to your further updating and contribution!

.../java/org/apache/iotdb/db/query/reader/seriesRelated/SeriesDataReaderWithoutValueFilter.java

...ain/java/org/apache/iotdb/db/query/reader/seriesRelated/RawDataReaderWithoutValueFilter.java

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/IAggregateReader.java

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/IRawReader.java

.../java/org/apache/iotdb/db/query/reader/seriesRelated/SeriesDataReaderWithoutValueFilter.java

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java

server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateFunction.java

server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/FirstValueAggrFunc.java

server/src/main/java/org/apache/iotdb/db/query/dataset/DeviceIterateDataSet.java

qiaojialin · 2020-01-12T03:03:45Z

server/src/main/java/org/apache/iotdb/db/engine/querycontext/ReadOnlyMemChunk.java

+    Statistics statsByType = Statistics.getStatsByType(dataType);
+    ChunkMetaData metaData = new ChunkMetaData(measurementUid, dataType, 0, statsByType);
+    if (!isEmpty()) {
+      List<TimeValuePair> sortedTimeValuePairList = getSortedTimeValuePairList();


could TimeValuePairSorter be replaced by TVList?

qiaojialin · 2020-01-12T03:12:04Z

server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateFunction.java

  public abstract AggreResultData getResult();

+
+  public abstract void calculateValueFromStatistics(Statistics chunkStatistics)


I suggest extracting AggreResultData out from AggregateFunction, each time calculateValueFromStatistics returns a AggreResultData, and merge outside.

qiaojialin · 2020-01-12T03:13:33Z

server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateFunction.java

+
+  public abstract void calculateValueFromPageData(BatchData dataInThisPage) throws IOException;
+
+  public abstract void calculateValueFromPageData(BatchData dataInThisPage,long bound) throws IOException;


add javadoc for bound

remove unused methods in this class

qiaojialin · 2020-01-13T00:56:31Z

server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/FirstValueAggrFunc.java

  }

+  @Override
+  public void calculateValueFromStatistics(Statistics chunkStatistics)


Suggested change

public void calculateValueFromStatistics(Statistics chunkStatistics)

public void calculateValueFromStatistics(Statistics statistics)

qiaojialin · 2020-01-13T02:57:49Z

server/src/main/java/org/apache/iotdb/db/query/dataset/DeviceIterateDataSet.java

            groupByPlan.setUnit(unit);
            groupByPlan.setDeduplicatedPaths(executePaths);
-            groupByPlan.setDeduplicatedDataTypes(dataTypes);
+            groupByPlan.setDeduplicatedDataTypes(tsDataTypes);


is this a bug before?

Yeah, looks like a bug from before

qiaojialin · 2020-01-13T03:00:14Z

server/src/main/java/org/apache/iotdb/db/query/dataset/NewEngineDataSetWithoutValueFilter.java

    timeHeap = new TreeSet<>();
    for (int i = 0; i < seriesReaderWithoutValueFilterList.size(); i++) {
-      ManagedSeriesReader reader = seriesReaderWithoutValueFilterList.get(i);
+      RawDataReaderWithoutValueFilter reader = seriesReaderWithoutValueFilterList.get(i);


The ManagedSeriesReader is for distribution, we'd better retain it.

done, but only batch was kept

qiaojialin · 2020-01-13T03:02:24Z

server/src/main/java/org/apache/iotdb/db/query/executor/AggregateEngineExecutor.java

+//  /**
+//   * constructor.
+//   */
+//  public AggregateEngineExecutor(AggregationPlan aggregationPlan) {


could this class be removed?

done, and I've removed all the unused classes and tests

qiaojialin · 2020-01-13T04:07:30Z

server/src/main/java/org/apache/iotdb/db/query/fill/LinearFill.java

+      if (!batchData.hasCurrent()) {
+        batchData = allDataReader.nextBatch();
+      }
+      afterPair = new TimeValuePair(batchData.currentTime(), batchData.currentTsPrimitiveType());


could we assure each batch from RawDataReader is not empty?

qiaojialin · 2020-01-13T04:14:21Z

server/src/main/java/org/apache/iotdb/db/query/fill/PreviousFill.java

+      if (!batchData.hasCurrent()) {
+        batchData = allDataReader.nextBatch();
+      }
+      cachedPair = new TimeValuePair(batchData.currentTime(), batchData.currentTsPrimitiveType());


The same question with LinearFill

qiaojialin · 2020-01-13T04:27:40Z

server/src/main/java/org/apache/iotdb/db/query/reader/chunkRelated/DiskChunkReader.java

@@ -36,10 +37,10 @@
 */
 public class DiskChunkReader implements IPointReader, IBatchReader {


DiskChunkReader may could be replaced by ChunkReader implement IPointReader.
Maybe ChunkReaderWrap could be replaced by IChunkReader

qiaojialin · 2020-01-13T05:35:02Z

server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateFunction.java

+   * Aggregate results cannot be calculated using Statistics directly, using the data in each page
+   *
+   * @param dataInThisPage the data in Page
+   * @param bound          the time upper bounder of data in unsequence data reader


calculate points whose time < bound

qiaojialin · 2020-01-13T05:49:12Z

server/src/main/java/org/apache/iotdb/db/query/reader/chunkRelated/ChunkReaderIterator.java

 * <p>
 */
-public class DiskChunkReader implements IPointReader, IBatchReader {
+public class ChunkReaderIterator implements IPointReader, IBatchReader {


This name is confusing... it seems could iterate chunk reader

rename to ChunkDataIterator

...rc/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java

samperson1997 · 2020-01-13T03:19:54Z

server/src/main/java/org/apache/iotdb/db/query/executor/NewAggregateEngineExecutor.java

+    while (newSeriesReader.hasNextChunk()) {
+      if (newSeriesReader.canUseChunkStatistics()) {
+        Statistics chunkStatistics = newSeriesReader.currentChunkStatistics();
+        function.calculateValueFromStatistics(chunkStatistics);
+        if (function.isCalculatedAggregationResult()) {
+          return function.getResult();
+        }
+        newSeriesReader.skipChunkData();
+        continue;
+      }
+      while (newSeriesReader.hasNextPage()) {
+        //cal by pageheader
+        if (newSeriesReader.canUsePageStatistics()) {
+          Statistics pageStatistic = newSeriesReader.currentChunkStatistics();
+          function.calculateValueFromStatistics(pageStatistic);
+          if (function.isCalculatedAggregationResult()) {
+            return function.getResult();
+          }
+          newSeriesReader.skipPageData();
+          continue;
+        }
+        //cal by pagedata
+        while (newSeriesReader.hasNextBatch()) {
+          function.calculateValueFromPageData(newSeriesReader.nextBatch());
+          if (function.isCalculatedAggregationResult()) {
+            return function.getResult();
+          }
+        }
+      }
+    }
+    return function.getResult();


extract these codes to avoid repeated codes in this file

qiaojialin · 2020-01-13T07:02:39Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+    this.context = context;
+    this.dataType = dataType;
+
+    if (filter != null) {


what if filter == null and has a TTL?

…or-iotdb into new_series_reader

qiaojialin · 2020-01-13T12:08:56Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+    } else if (seqChunkMetadatas.isEmpty() && !unseqChunkMetadatas.isEmpty()) {
+      chunkMetaData = unseqChunkMetadatas.pollFirst();
+    } else if (!seqChunkMetadatas.isEmpty()) {
+      // seq 和 unseq 的 chunk metadata 都不为空


Suggested change

// seq 和 unseq 的 chunk metadata 都不为空

// neither seqChunkMetadatas nor unseqChunkMetadatas is null

qiaojialin · 2020-01-13T12:13:35Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+   * unseq file is a very special file that intersects not only with an ordered file, but also with
+   * another unseq file. So we need a way to find all the files that might be used to intersect the
+   * current measurement point.


unseq files are very special files that intersect not only with sequence files, but also with
other unseq files. So we need to find all tsfiles that overlapped with current chunk and
extract chunks from the resource

qiaojialin · 2020-01-13T12:14:14Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+  private void fillOverlappedFiles() throws IOException {
+    while (!unseqFileResource.isEmpty()) {
+      Map<String, Long> startTimeMap = unseqFileResource.first().getStartTimeMap();
+      Long unSeqStartTime = startTimeMap.getOrDefault(seriesPath.getDevice(), Long.MIN_VALUE);


Suggested change

Long unSeqStartTime = startTimeMap.getOrDefault(seriesPath.getDevice(), Long.MIN_VALUE);

Long unSeqStartTime = startTimeMap.getOrDefault(seriesPath.getDevice(), Long.MAX_VALUE);

qiaojialin · 2020-01-13T12:14:34Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+    while (!unseqFileResource.isEmpty()) {
+      Map<String, Long> startTimeMap = unseqFileResource.first().getStartTimeMap();
+      Long unSeqStartTime = startTimeMap.getOrDefault(seriesPath.getDevice(), Long.MIN_VALUE);
+      if (chunkMetaData.getEndTime() > unSeqStartTime) {


Suggested change

if (chunkMetaData.getEndTime() > unSeqStartTime) {

if (chunkMetaData.getEndTime() >= unSeqStartTime) {

qiaojialin · 2020-01-13T12:29:35Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+
+  //for test
+  public AbstractDataReader(Path seriesPath, TSDataType dataType,
+      Filter filter, QueryContext context, List<TsFileResource> resources) throws IOException {


Suggested change

Filter filter, QueryContext context, List<TsFileResource> resources) throws IOException {

Filter filter, QueryContext context, List<TsFileResource> seqResources) throws IOException {

qiaojialin · 2020-01-13T12:30:49Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+   * another unseq file. So we need a way to find all the files that might be used to intersect the
+   * current measurement point.
+   */
+  private void fillOverlappedFiles() throws IOException {


Suggested change

private void fillOverlappedFiles() throws IOException {

private void unpackOverlappedFiles() throws IOException {

qiaojialin · 2020-01-13T12:49:57Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+    while (!unseqChunkMetadatas.isEmpty()) {
+      long startTime = unseqChunkMetadatas.first().getStartTime();
+
+      if (chunkMetaData.getEndTime() > startTime) {


Suggested change

if (chunkMetaData.getEndTime() > startTime) {

if (chunkMetaData.getEndTime() >= startTime) {

qiaojialin · 2020-01-13T12:50:09Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+    while (!seqChunkMetadatas.isEmpty()) {
+      long startTime = seqChunkMetadatas.get(0).getStartTime();
+
+      if (chunkMetaData.getEndTime() > startTime) {


Suggested change

if (chunkMetaData.getEndTime() > startTime) {

if (chunkMetaData.getEndTime() >= startTime) {

qiaojialin · 2020-01-14T00:55:13Z

server/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/AbstractDataReader.java

+    } else {
+      Chunk chunk = chunkLoader.getChunk(metaData);
+      chunkReader = new ChunkReader(chunk, filter);
+      chunkReader.hasNextSatisfiedPage();


This line is strange, it's better to remove it.

add some impl

8a6afad

qiaojialin reviewed Dec 30, 2019

View reviewed changes

liudw added 4 commits December 30, 2019 23:14

fix bugs

f4fd809

modify overlapped to Statistics

254c976

delete useless

e8bf53c

add asf header

a9a7c0d

qiaojialin changed the title ~~add some impl~~ add some impl of NewSeriesReaderWithoutValueFilter Dec 31, 2019

liudw added 13 commits January 2, 2020 15:55

fix GroupByWithoutValueFilterDataSet

3a2c30d

fix all GroupBy data

b05f90f

fix ifill bug

fdfa758

rewrite reader for merge muti unseq

268fad7

fix many bug

2a29285

revert tsfile changes

d9e66ff

add comments

56692e4

fix groupby [0-100)

71244c2

all done

33bb57c

fix merge test

5dee0bf

Merge branch 'new_series_reader' of https://github.com/apache/incubat…

80b44ae

…or-iotdb into new_series_reader � Conflicts: � server/src/main/java/org/apache/iotdb/db/query/dataset/NewEngineDataSetWithoutValueFilter.java

fix MergeOverLapTest

7a8c4af

liudw added 5 commits January 11, 2020 11:06

Merge branch 'new_series_reader' of https://github.com/apache/incubat…

d78f923

…or-iotdb into new_series_reader

fix MergeTaskTest

dcd47ee

fix MergeTaskTest

9deee19

close chunkloader

99f9dbe

close chunkloader

8dfdb9d

samperson1997 requested changes Jan 12, 2020

View reviewed changes

change with review

590ece1

qiaojialin requested changes Jan 13, 2020

View reviewed changes

qiaojialin reviewed Jan 13, 2020

View reviewed changes

samperson1997 requested changes Jan 13, 2020

View reviewed changes

qiaojialin reviewed Jan 13, 2020

View reviewed changes

liudw added 2 commits January 13, 2020 16:05

change with review

ec8e944

Merge branch 'new_series_reader' of https://github.com/apache/incubat…

a8d7ae1

…or-iotdb into new_series_reader

samperson1997 self-requested a review January 13, 2020 08:50

liudw added 4 commits January 13, 2020 17:10

fix group by in last max

e7b2dad

add javadoc for reader

a86aef7

rename AggregateFunction to AggregateResult

2fe8dcc

Merge branch 'new_series_reader' of https://github.com/apache/incubat…

6f6a1b8

…or-iotdb into new_series_reader

qiaojialin reviewed Jan 13, 2020

View reviewed changes

delete IPageSkipReader and modify AbstractDataReader

3a34bcc

qiaojialin reviewed Jan 14, 2020

View reviewed changes

qiaojialin approved these changes Jan 14, 2020

View reviewed changes

qiaojialin changed the title ~~add some impl of NewSeriesReaderWithoutValueFilter~~ [IOTDB-418] add some impl of NewSeriesReaderWithoutValueFilter Jan 14, 2020

qiaojialin merged commit 3092715 into apache:new_series_reader Jan 14, 2020

liutaohua deleted the new_series_reader branch January 14, 2020 05:21

		public abstract AggreResultData getResult();


		public abstract void calculateValueFromStatistics(Statistics chunkStatistics)


		public abstract void calculateValueFromPageData(BatchData dataInThisPage) throws IOException;

		public abstract void calculateValueFromPageData(BatchData dataInThisPage,long bound) throws IOException;

	public void calculateValueFromStatistics(Statistics chunkStatistics)
	public void calculateValueFromStatistics(Statistics statistics)

		@@ -36,10 +37,10 @@
		*/
		public class DiskChunkReader implements IPointReader, IBatchReader {

	// seq 和 unseq 的 chunk metadata 都不为空
	// neither seqChunkMetadatas nor unseqChunkMetadatas is null

	Long unSeqStartTime = startTimeMap.getOrDefault(seriesPath.getDevice(), Long.MIN_VALUE);
	Long unSeqStartTime = startTimeMap.getOrDefault(seriesPath.getDevice(), Long.MAX_VALUE);

	if (chunkMetaData.getEndTime() > unSeqStartTime) {
	if (chunkMetaData.getEndTime() >= unSeqStartTime) {

	Filter filter, QueryContext context, List<TsFileResource> resources) throws IOException {
	Filter filter, QueryContext context, List<TsFileResource> seqResources) throws IOException {

	private void fillOverlappedFiles() throws IOException {
	private void unpackOverlappedFiles() throws IOException {

	if (chunkMetaData.getEndTime() > startTime) {
	if (chunkMetaData.getEndTime() >= startTime) {

Conversation

liutaohua commented Dec 30, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

qiaojialin Dec 30, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jixuan1989 commented Jan 11, 2020

Uh oh!

liutaohua commented Jan 11, 2020

Uh oh!

samperson1997 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

liutaohua commented Dec 30, 2019 •

edited

Loading

qiaojialin Dec 30, 2019 •

edited

Loading

qiaojialin Jan 13, 2020 •

edited

Loading