You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/03/25 10:17:57 UTC
[iotdb] 01/02: add SeriesScanOperater implementation
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch ty-mpp
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 116354a2e522c2f7ccf05bf00ebf6e6b1a578e2b
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Fri Mar 25 18:12:37 2022 +0800
add SeriesScanOperater implementation
---
.../apache/iotdb/db/mpp/buffer/ISinkHandle.java | 2 +-
.../apache/iotdb/db/mpp/buffer/ISourceHandle.java | 2 +-
.../apache/iotdb/db/mpp/buffer/SourceHandle.java | 2 +-
.../db/mpp/execution/FragmentInstanceContext.java | 54 +-
.../org/apache/iotdb/db/mpp/operator/Operator.java | 8 +-
.../iotdb/db/mpp/operator/OperatorContext.java | 13 +-
.../iotdb/db/mpp/operator/SeriesScanOperator.java | 39 -
.../db/mpp/operator/process/AggregateOperator.java | 2 +-
.../mpp/operator/process/DeviceMergeOperator.java | 2 +-
.../db/mpp/operator/process/FillOperator.java | 2 +-
.../mpp/operator/process/FilterNullOperator.java | 2 +-
.../mpp/operator/process/GroupByLevelOperator.java | 2 +-
.../db/mpp/operator/process/LimitOperator.java | 8 +-
.../db/mpp/operator/process/OffsetOperator.java | 2 +-
.../db/mpp/operator/process/SortOperator.java | 2 +-
.../db/mpp/operator/process/TimeJoinOperator.java | 9 +-
.../db/mpp/operator/sink/FragmentSinkOperator.java | 2 +-
.../iotdb/db/mpp/operator/sink/SinkOperator.java | 2 +-
.../source/SeriesAggregateScanOperator.java | 2 +-
.../db/mpp/operator/source/SeriesScanOperator.java | 109 +-
.../db/mpp/operator/source/SeriesScanUtil.java | 1372 ++++++++++++++++++++
.../iotdb/db/query/context/QueryContext.java | 4 +-
.../query/reader/chunk/MemAlignedPageReader.java | 7 +
.../iotdb/db/query/reader/chunk/MemPageReader.java | 46 +
.../apache/iotdb/tsfile/read}/common/Column.java | 2 +-
.../iotdb/tsfile/read}/common/TimeColumn.java | 2 +-
.../apache/iotdb/tsfile/read}/common/TsBlock.java | 69 +-
.../iotdb/tsfile/read}/common/TsBlockMetadata.java | 2 +-
.../iotdb/tsfile/read/reader/IPageReader.java | 3 +
.../tsfile/read/reader/page/AlignedPageReader.java | 7 +
.../iotdb/tsfile/read/reader/page/PageReader.java | 53 +
31 files changed, 1738 insertions(+), 95 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISinkHandle.java b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISinkHandle.java
index 533cb40..0d64cbc 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISinkHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISinkHandle.java
@@ -18,7 +18,7 @@
*/
package org.apache.iotdb.db.mpp.buffer;
-import org.apache.iotdb.db.mpp.common.TsBlock;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISourceHandle.java b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISourceHandle.java
index b33d3ef..b493759 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISourceHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/ISourceHandle.java
@@ -18,7 +18,7 @@
*/
package org.apache.iotdb.db.mpp.buffer;
-import org.apache.iotdb.db.mpp.common.TsBlock;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
index b1373d7..0216556 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/buffer/SourceHandle.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.mpp.buffer;
-import org.apache.iotdb.db.mpp.common.TsBlock;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceContext.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceContext.java
index eb00953..6a93b9b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceContext.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceContext.java
@@ -18,23 +18,29 @@
*/
package org.apache.iotdb.db.mpp.execution;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
+import org.apache.iotdb.db.metadata.path.AlignedPath;
+import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.query.context.QueryContext;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import static com.google.common.base.Preconditions.checkArgument;
-public class FragmentInstanceContext {
+public class FragmentInstanceContext extends QueryContext {
private FragmentInstanceId id;
// TODO if we split one fragment instance into multiple pipelines to run, we need to replace it
// with CopyOnWriteArrayList or some other thread safe data structure
private final List<OperatorContext> operatorContexts = new ArrayList<>();
-
private final long createNanos = System.nanoTime();
// private final GcMonitor gcMonitor;
@@ -60,8 +66,50 @@ public class FragmentInstanceContext {
operatorId);
}
- OperatorContext operatorContext = new OperatorContext(operatorId, planNodeId, operatorType);
+ OperatorContext operatorContext =
+ new OperatorContext(operatorId, planNodeId, operatorType, this);
operatorContexts.add(operatorContext);
return operatorContext;
}
+
+ /**
+ * Find the modifications of timeseries 'path' in 'modFile'. If they are not in the cache, read
+ * them from 'modFile' and put then into the cache.
+ */
+ public List<Modification> getPathModifications(ModificationFile modFile, PartialPath path) {
+ Map<String, List<Modification>> fileModifications =
+ filePathModCache.computeIfAbsent(modFile.getFilePath(), k -> new ConcurrentHashMap<>());
+ return fileModifications.computeIfAbsent(
+ path.getFullPath(),
+ k -> {
+ List<Modification> allModifications = fileModCache.get(modFile.getFilePath());
+ if (allModifications == null) {
+ allModifications = (List<Modification>) modFile.getModifications();
+ fileModCache.put(modFile.getFilePath(), allModifications);
+ }
+ List<Modification> finalPathModifications = new ArrayList<>();
+ if (!allModifications.isEmpty()) {
+ allModifications.forEach(
+ modification -> {
+ if (modification.getPath().matchFullPath(path)) {
+ finalPathModifications.add(modification);
+ }
+ });
+ }
+ return finalPathModifications;
+ });
+ }
+
+ /**
+ * Find the modifications of all aligned 'paths' in 'modFile'. If they are not in the cache, read
+ * them from 'modFile' and put then into the cache.
+ */
+ public List<List<Modification>> getPathModifications(ModificationFile modFile, AlignedPath path) {
+ int n = path.getMeasurementList().size();
+ List<List<Modification>> ans = new ArrayList<>(n);
+ for (int i = 0; i < n; i++) {
+ ans.add(getPathModifications(modFile, path.getPathWithMeasurement(i)));
+ }
+ return ans;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/Operator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/Operator.java
index aa7ae8a..d545c4f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/Operator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/Operator.java
@@ -18,10 +18,12 @@
*/
package org.apache.iotdb.db.mpp.operator;
-import org.apache.iotdb.db.mpp.common.TsBlock;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
+import java.io.IOException;
+
import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
public interface Operator extends AutoCloseable {
@@ -38,10 +40,10 @@ public interface Operator extends AutoCloseable {
}
/** Gets next tsBlock from this operator. If no data is currently available, return null. */
- TsBlock next();
+ TsBlock next() throws IOException;
/** @return true if the operator has more data, otherwise false */
- boolean hasNext();
+ boolean hasNext() throws IOException;
/** This method will always be called before releasing the Operator reference. */
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/OperatorContext.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/OperatorContext.java
index feedc13..a2e31d5 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/OperatorContext.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/OperatorContext.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.mpp.operator;
+import org.apache.iotdb.db.mpp.execution.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
/**
@@ -30,14 +31,24 @@ public class OperatorContext {
private final int operatorId;
private final PlanNodeId planNodeId;
private final String operatorType;
+ private final FragmentInstanceContext instanceContext;
- public OperatorContext(int operatorId, PlanNodeId planNodeId, String operatorType) {
+ public OperatorContext(
+ int operatorId,
+ PlanNodeId planNodeId,
+ String operatorType,
+ FragmentInstanceContext instanceContext) {
this.operatorId = operatorId;
this.planNodeId = planNodeId;
this.operatorType = operatorType;
+ this.instanceContext = instanceContext;
}
public int getOperatorId() {
return operatorId;
}
+
+ public FragmentInstanceContext getInstanceContext() {
+ return instanceContext;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/SeriesScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/SeriesScanOperator.java
deleted file mode 100644
index 43969af..0000000
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/SeriesScanOperator.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.mpp.operator;
-
-import org.apache.iotdb.db.mpp.common.TsBlock;
-
-public class SeriesScanOperator implements Operator {
-
- @Override
- public OperatorContext getOperatorContext() {
- return null;
- }
-
- @Override
- public TsBlock next() {
- return null;
- }
-
- @Override
- public boolean hasNext() {
- return false;
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/AggregateOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/AggregateOperator.java
index 6366b29..c439ad4 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/AggregateOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/AggregateOperator.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.operator.process;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/DeviceMergeOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/DeviceMergeOperator.java
index 408296a..b9fce3a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/DeviceMergeOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/DeviceMergeOperator.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.operator.process;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/FillOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/FillOperator.java
index 66dc3ea..e2732cf 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/FillOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/FillOperator.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.operator.process;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/FilterNullOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/FilterNullOperator.java
index 8d15250..d8a33ba 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/FilterNullOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/FilterNullOperator.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.operator.process;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/GroupByLevelOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/GroupByLevelOperator.java
index 10e9daa..6e19233 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/GroupByLevelOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/GroupByLevelOperator.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.operator.process;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/LimitOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/LimitOperator.java
index e6e9e61..47c81bc 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/LimitOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/LimitOperator.java
@@ -18,12 +18,14 @@
*/
package org.apache.iotdb.db.mpp.operator.process;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.Operator;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
+import java.io.IOException;
+
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
@@ -51,7 +53,7 @@ public class LimitOperator implements ProcessOperator {
}
@Override
- public TsBlock next() {
+ public TsBlock next() throws IOException {
TsBlock block = child.next();
TsBlock res = block;
if (block.getCount() <= remainingLimit) {
@@ -64,7 +66,7 @@ public class LimitOperator implements ProcessOperator {
}
@Override
- public boolean hasNext() {
+ public boolean hasNext() throws IOException {
return child.hasNext();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/OffsetOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/OffsetOperator.java
index 25b4bc9..e9eef20 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/OffsetOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/OffsetOperator.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.operator.process;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/SortOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/SortOperator.java
index 0199d52..f0ac49b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/SortOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/SortOperator.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.operator.process;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/TimeJoinOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/TimeJoinOperator.java
index 6d511ff..c5665ee 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/TimeJoinOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/process/TimeJoinOperator.java
@@ -18,15 +18,16 @@
*/
package org.apache.iotdb.db.mpp.operator.process;
-import org.apache.iotdb.db.mpp.common.TimeColumn;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.Operator;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
import org.apache.iotdb.db.mpp.sql.statement.component.OrderBy;
import org.apache.iotdb.db.utils.datastructure.TimeSelector;
+import org.apache.iotdb.tsfile.read.common.TimeColumn;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
+import java.io.IOException;
import java.util.List;
public class TimeJoinOperator implements ProcessOperator {
@@ -81,7 +82,7 @@ public class TimeJoinOperator implements ProcessOperator {
}
@Override
- public TsBlock next() {
+ public TsBlock next() throws IOException {
// end time for returned TsBlock this time, it's the min end time among all the children
// TsBlocks
long currentEndTime = 0;
@@ -131,7 +132,7 @@ public class TimeJoinOperator implements ProcessOperator {
}
@Override
- public boolean hasNext() {
+ public boolean hasNext() throws IOException {
for (int i = 0; i < inputCount; i++) {
if (!empty(i)) {
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/sink/FragmentSinkOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/sink/FragmentSinkOperator.java
index aca892f..7be5cfa 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/sink/FragmentSinkOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/sink/FragmentSinkOperator.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.operator.sink;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/sink/SinkOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/sink/SinkOperator.java
index d362c1d..e5a3e15 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/sink/SinkOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/sink/SinkOperator.java
@@ -18,8 +18,8 @@
*/
package org.apache.iotdb.db.mpp.operator.sink;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.Operator;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
public interface SinkOperator extends Operator {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesAggregateScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesAggregateScanOperator.java
index fcd4605..0ef6b09 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesAggregateScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesAggregateScanOperator.java
@@ -18,9 +18,9 @@
*/
package org.apache.iotdb.db.mpp.operator.source;
-import org.apache.iotdb.db.mpp.common.TsBlock;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import com.google.common.util.concurrent.ListenableFuture;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanOperator.java
index c0bc9ca..a056c13 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanOperator.java
@@ -7,7 +7,7 @@
* "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
+ * 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
@@ -18,41 +18,114 @@
*/
package org.apache.iotdb.db.mpp.operator.source;
-import org.apache.iotdb.db.mpp.common.TsBlock;
+import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.mpp.operator.Operator;
import org.apache.iotdb.db.mpp.operator.OperatorContext;
-import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
-import com.google.common.util.concurrent.ListenableFuture;
+import java.io.IOException;
+import java.util.Set;
-public class SeriesScanOperator implements SourceOperator {
+public class SeriesScanOperator implements Operator {
+
+ private final OperatorContext operatorContext;
+ private final SeriesScanUtil seriesScanUtil;
+ private TsBlock tsBlock;
+ private boolean hasCachedTsBlock = false;
+
+ public SeriesScanOperator(
+ PartialPath seriesPath,
+ Set<String> allSensors,
+ TSDataType dataType,
+ OperatorContext context,
+ QueryDataSource dataSource,
+ Filter timeFilter,
+ Filter valueFilter,
+ boolean ascending) {
+ this.operatorContext = context;
+ this.seriesScanUtil =
+ new SeriesScanUtil(
+ seriesPath,
+ allSensors,
+ dataType,
+ context.getInstanceContext(),
+ dataSource,
+ timeFilter,
+ valueFilter,
+ ascending);
+ }
@Override
public OperatorContext getOperatorContext() {
- return null;
+ return operatorContext;
}
@Override
- public ListenableFuture<Void> isBlocked() {
- return SourceOperator.super.isBlocked();
+ public TsBlock next() throws IOException {
+ if (hasCachedTsBlock || hasNext()) {
+ hasCachedTsBlock = false;
+ return tsBlock;
+ }
+ throw new IOException("no next batch");
}
@Override
- public TsBlock next() {
- return null;
+ public boolean hasNext() throws IOException {
+
+ if (hasCachedTsBlock) {
+ return true;
+ }
+
+ /*
+ * consume page data firstly
+ */
+ if (readPageData()) {
+ hasCachedTsBlock = true;
+ return true;
+ }
+
+ /*
+ * consume chunk data secondly
+ */
+ if (readChunkData()) {
+ hasCachedTsBlock = true;
+ return true;
+ }
+
+ /*
+ * consume next file finally
+ */
+ while (seriesScanUtil.hasNextFile()) {
+ if (readChunkData()) {
+ hasCachedTsBlock = true;
+ return true;
+ }
+ }
+ return hasCachedTsBlock;
}
- @Override
- public boolean hasNext() {
+ private boolean readChunkData() throws IOException {
+ while (seriesScanUtil.hasNextChunk()) {
+ if (readPageData()) {
+ return true;
+ }
+ }
return false;
}
- @Override
- public void close() throws Exception {
- SourceOperator.super.close();
+ private boolean readPageData() throws IOException {
+ while (seriesScanUtil.hasNextPage()) {
+ tsBlock = seriesScanUtil.nextPage();
+ if (!isEmpty(tsBlock)) {
+ return true;
+ }
+ }
+ return false;
}
- @Override
- public PlanNodeId getSourceId() {
- return null;
+ private boolean isEmpty(TsBlock tsBlock) {
+ return tsBlock == null || !tsBlock.hasNext();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanUtil.java b/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanUtil.java
new file mode 100644
index 0000000..9e774dc
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/operator/source/SeriesScanUtil.java
@@ -0,0 +1,1372 @@
+/*
+ * 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.mpp.operator.source;
+
+import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.metadata.idtable.IDTable;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.mpp.execution.FragmentInstanceContext;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.reader.universal.DescPriorityMergeReader;
+import org.apache.iotdb.db.query.reader.universal.PriorityMergeReader;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.AlignedTimeSeriesMetadata;
+import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata;
+import org.apache.iotdb.tsfile.file.metadata.ITimeSeriesMetadata;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.filter.basic.UnaryFilter;
+import org.apache.iotdb.tsfile.read.reader.IAlignedPageReader;
+import org.apache.iotdb.tsfile.read.reader.IPageReader;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.*;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
+
+public class SeriesScanUtil {
+ private final FragmentInstanceContext context;
+
+ // The path of the target series which will be scanned.
+ private final PartialPath seriesPath;
+
+ // all the sensors in this device;
+ protected final Set<String> allSensors;
+ protected final TSDataType dataType;
+
+ // inner class of SeriesReader for order purpose
+ private TimeOrderUtils orderUtils;
+
+ /*
+ * There is at most one is not null between timeFilter and valueFilter
+ *
+ * timeFilter is pushed down to all pages (seq, unseq) without correctness problem
+ *
+ * valueFilter is pushed down to non-overlapped page only
+ */
+ private final Filter timeFilter;
+ private final Filter valueFilter;
+
+ private final QueryDataSource dataSource;
+
+ /*
+ * file index
+ */
+ protected int curSeqFileIndex;
+ protected int curUnseqFileIndex;
+
+ /*
+ * TimeSeriesMetadata cache
+ */
+ protected ITimeSeriesMetadata firstTimeSeriesMetadata;
+ protected final List<ITimeSeriesMetadata> seqTimeSeriesMetadata = new LinkedList<>();
+ protected final PriorityQueue<ITimeSeriesMetadata> unSeqTimeSeriesMetadata;
+
+ /*
+ * chunk cache
+ */
+ protected IChunkMetadata firstChunkMetadata;
+ protected final PriorityQueue<IChunkMetadata> cachedChunkMetadata;
+
+ /*
+ * page cache
+ */
+ protected VersionPageReader firstPageReader;
+ protected final List<VersionPageReader> seqPageReaders = new LinkedList<>();
+ protected final PriorityQueue<VersionPageReader> unSeqPageReaders;
+
+ /*
+ * point cache
+ */
+ protected final PriorityMergeReader mergeReader;
+
+ /*
+ * result cache
+ */
+ protected boolean hasCachedNextOverlappedPage;
+ protected TsBlock cachedTsBlock;
+
+ public SeriesScanUtil(
+ PartialPath seriesPath,
+ Set<String> allSensors,
+ TSDataType dataType,
+ FragmentInstanceContext context,
+ QueryDataSource dataSource,
+ Filter timeFilter,
+ Filter valueFilter,
+ boolean ascending) {
+ this.seriesPath = IDTable.translateQueryPath(seriesPath);
+ this.allSensors = allSensors;
+ this.dataType = dataType;
+ this.context = context;
+ this.dataSource = dataSource;
+ this.timeFilter = timeFilter;
+ this.valueFilter = valueFilter;
+ if (ascending) {
+ this.orderUtils = new AscTimeOrderUtils();
+ mergeReader = getPriorityMergeReader();
+ this.curSeqFileIndex = 0;
+ this.curUnseqFileIndex = 0;
+ } else {
+ this.orderUtils = new DescTimeOrderUtils();
+ mergeReader = getDescPriorityMergeReader();
+ this.curSeqFileIndex = dataSource.getSeqResourcesSize() - 1;
+ this.curUnseqFileIndex = 0;
+ }
+
+ unSeqTimeSeriesMetadata =
+ new PriorityQueue<>(
+ orderUtils.comparingLong(
+ timeSeriesMetadata -> orderUtils.getOrderTime(timeSeriesMetadata.getStatistics())));
+ cachedChunkMetadata =
+ new PriorityQueue<>(
+ orderUtils.comparingLong(
+ chunkMetadata -> orderUtils.getOrderTime(chunkMetadata.getStatistics())));
+ unSeqPageReaders =
+ new PriorityQueue<>(
+ orderUtils.comparingLong(
+ versionPageReader -> orderUtils.getOrderTime(versionPageReader.getStatistics())));
+ }
+
+ protected PriorityMergeReader getPriorityMergeReader() {
+ return new PriorityMergeReader();
+ }
+
+ protected DescPriorityMergeReader getDescPriorityMergeReader() {
+ return new DescPriorityMergeReader();
+ }
+
+ public boolean isEmpty() throws IOException {
+ return !(hasNextPage() || hasNextChunk() || hasNextFile());
+ }
+
+ boolean hasNextFile() throws IOException {
+
+ if (!unSeqPageReaders.isEmpty()
+ || firstPageReader != null
+ || mergeReader.hasNextTimeValuePair()) {
+ throw new IOException(
+ "all cached pages should be consumed first unSeqPageReaders.isEmpty() is "
+ + unSeqPageReaders.isEmpty()
+ + " firstPageReader != null is "
+ + (firstPageReader != null)
+ + " mergeReader.hasNextTimeValuePair() = "
+ + mergeReader.hasNextTimeValuePair());
+ }
+
+ if (firstChunkMetadata != null || !cachedChunkMetadata.isEmpty()) {
+ throw new IOException("all cached chunks should be consumed first");
+ }
+
+ if (firstTimeSeriesMetadata != null) {
+ return true;
+ }
+
+ while (firstTimeSeriesMetadata == null
+ && (orderUtils.hasNextSeqResource()
+ || orderUtils.hasNextUnseqResource()
+ || !seqTimeSeriesMetadata.isEmpty()
+ || !unSeqTimeSeriesMetadata.isEmpty())) {
+ // init first time series metadata whose startTime is minimum
+ tryToUnpackAllOverlappedFilesToTimeSeriesMetadata();
+ }
+
+ return firstTimeSeriesMetadata != null;
+ }
+
+ boolean isFileOverlapped() throws IOException {
+ if (firstTimeSeriesMetadata == null) {
+ throw new IOException("no first file");
+ }
+
+ Statistics fileStatistics = firstTimeSeriesMetadata.getStatistics();
+ return !seqTimeSeriesMetadata.isEmpty()
+ && orderUtils.isOverlapped(fileStatistics, seqTimeSeriesMetadata.get(0).getStatistics())
+ || !unSeqTimeSeriesMetadata.isEmpty()
+ && orderUtils.isOverlapped(
+ fileStatistics, unSeqTimeSeriesMetadata.peek().getStatistics());
+ }
+
+ Statistics currentFileStatistics() {
+ return firstTimeSeriesMetadata.getStatistics();
+ }
+
+ Statistics currentFileStatistics(int index) throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics by index from alignedTimeSeriesMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getStatistics(index);
+ }
+
+ public Statistics currentFileTimeStatistics() throws IOException {
+ if (!(firstTimeSeriesMetadata instanceof AlignedTimeSeriesMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
+ return ((AlignedTimeSeriesMetadata) firstTimeSeriesMetadata).getTimeStatistics();
+ }
+
+ boolean currentFileModified() throws IOException {
+ if (firstTimeSeriesMetadata == null) {
+ throw new IOException("no first file");
+ }
+ return firstTimeSeriesMetadata.isModified();
+ }
+
+ void skipCurrentFile() {
+ firstTimeSeriesMetadata = null;
+ }
+
+ /**
+ * This method should be called after hasNextFile() until no next chunk, make sure that all
+ * overlapped chunks are consumed
+ */
+ boolean hasNextChunk() throws IOException {
+
+ if (!unSeqPageReaders.isEmpty()
+ || firstPageReader != null
+ || mergeReader.hasNextTimeValuePair()) {
+ throw new IOException(
+ "all cached pages should be consumed first unSeqPageReaders.isEmpty() is "
+ + unSeqPageReaders.isEmpty()
+ + " firstPageReader != null is "
+ + (firstPageReader != null)
+ + " mergeReader.hasNextTimeValuePair() = "
+ + mergeReader.hasNextTimeValuePair());
+ }
+
+ if (firstChunkMetadata != null) {
+ return true;
+ }
+
+ while (firstChunkMetadata == null && (!cachedChunkMetadata.isEmpty() || hasNextFile())) {
+ initFirstChunkMetadata();
+ }
+ return firstChunkMetadata != null;
+ }
+
+ /** construct first chunk metadata */
+ private void initFirstChunkMetadata() throws IOException {
+ if (firstTimeSeriesMetadata != null) {
+ /*
+ * try to unpack all overlapped TimeSeriesMetadata to cachedChunkMetadata
+ */
+ unpackAllOverlappedTsFilesToTimeSeriesMetadata(
+ orderUtils.getOverlapCheckTime(firstTimeSeriesMetadata.getStatistics()));
+ unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(
+ orderUtils.getOverlapCheckTime(firstTimeSeriesMetadata.getStatistics()), true);
+ } else {
+ /*
+ * first time series metadata is already unpacked, consume cached ChunkMetadata
+ */
+ while (!cachedChunkMetadata.isEmpty()) {
+ firstChunkMetadata = cachedChunkMetadata.peek();
+ unpackAllOverlappedTsFilesToTimeSeriesMetadata(
+ orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics()));
+ unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(
+ orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics()), false);
+ if (firstChunkMetadata.equals(cachedChunkMetadata.peek())) {
+ firstChunkMetadata = cachedChunkMetadata.poll();
+ break;
+ }
+ }
+ }
+ if (valueFilter != null
+ && firstChunkMetadata != null
+ && !isChunkOverlapped()
+ && !firstChunkMetadata.isModified()
+ && !valueFilter.satisfy(firstChunkMetadata.getStatistics())) {
+ skipCurrentChunk();
+ }
+ }
+
+ private void unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(
+ long endpointTime, boolean init) throws IOException {
+ while (!seqTimeSeriesMetadata.isEmpty()
+ && orderUtils.isOverlapped(endpointTime, seqTimeSeriesMetadata.get(0).getStatistics())) {
+ unpackOneTimeSeriesMetadata(seqTimeSeriesMetadata.remove(0));
+ }
+ while (!unSeqTimeSeriesMetadata.isEmpty()
+ && orderUtils.isOverlapped(endpointTime, unSeqTimeSeriesMetadata.peek().getStatistics())) {
+ unpackOneTimeSeriesMetadata(unSeqTimeSeriesMetadata.poll());
+ }
+
+ if (firstTimeSeriesMetadata != null
+ && orderUtils.isOverlapped(endpointTime, firstTimeSeriesMetadata.getStatistics())) {
+ unpackOneTimeSeriesMetadata(firstTimeSeriesMetadata);
+ firstTimeSeriesMetadata = null;
+ }
+
+ if (init && firstChunkMetadata == null && !cachedChunkMetadata.isEmpty()) {
+ firstChunkMetadata = cachedChunkMetadata.poll();
+ }
+ }
+
+ protected void unpackOneTimeSeriesMetadata(ITimeSeriesMetadata timeSeriesMetadata)
+ throws IOException {
+ List<IChunkMetadata> chunkMetadataList =
+ FileLoaderUtils.loadChunkMetadataList(timeSeriesMetadata);
+ chunkMetadataList.forEach(chunkMetadata -> chunkMetadata.setSeq(timeSeriesMetadata.isSeq()));
+
+ cachedChunkMetadata.addAll(chunkMetadataList);
+ }
+
+ boolean isChunkOverlapped() throws IOException {
+ if (firstChunkMetadata == null) {
+ throw new IOException("no first chunk");
+ }
+
+ Statistics chunkStatistics = firstChunkMetadata.getStatistics();
+ return !cachedChunkMetadata.isEmpty()
+ && orderUtils.isOverlapped(chunkStatistics, cachedChunkMetadata.peek().getStatistics());
+ }
+
+ Statistics currentChunkStatistics() {
+ return firstChunkMetadata.getStatistics();
+ }
+
+ Statistics currentChunkStatistics(int index) throws IOException {
+ if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
+ throw new IOException("Can only get statistics by index from vectorChunkMetaData");
+ }
+ return ((AlignedChunkMetadata) firstChunkMetadata).getStatistics(index);
+ }
+
+ Statistics currentChunkTimeStatistics() throws IOException {
+ if (!(firstChunkMetadata instanceof AlignedChunkMetadata)) {
+ throw new IOException("Can only get statistics of time column from alignedChunkMetaData");
+ }
+ return ((AlignedChunkMetadata) firstChunkMetadata).getTimeStatistics();
+ }
+
+ boolean currentChunkModified() throws IOException {
+ if (firstChunkMetadata == null) {
+ throw new IOException("no first chunk");
+ }
+ return firstChunkMetadata.isModified();
+ }
+
+ void skipCurrentChunk() {
+ firstChunkMetadata = null;
+ }
+
+ /**
+ * This method should be called after hasNextChunk() until no next page, make sure that all
+ * overlapped pages are consumed
+ */
+ @SuppressWarnings("squid:S3776")
+ // Suppress high Cognitive Complexity warning
+ boolean hasNextPage() throws IOException {
+
+ /*
+ * has overlapped data before
+ */
+ if (hasCachedNextOverlappedPage) {
+ return true;
+ } else if (mergeReader.hasNextTimeValuePair() || firstPageOverlapped()) {
+ if (hasNextOverlappedPage()) {
+ cachedTsBlock = nextOverlappedPage();
+ if (cachedTsBlock != null && cachedTsBlock.hasNext()) {
+ hasCachedNextOverlappedPage = true;
+ return true;
+ }
+ }
+ }
+
+ if (firstPageReader != null) {
+ return true;
+ }
+
+ /*
+ * construct first page reader
+ */
+ if (firstChunkMetadata != null) {
+ /*
+ * try to unpack all overlapped ChunkMetadata to cachedPageReaders
+ */
+ unpackAllOverlappedChunkMetadataToPageReaders(
+ orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics()), true);
+ } else {
+ /*
+ * first chunk metadata is already unpacked, consume cached pages
+ */
+ initFirstPageReader();
+ }
+
+ if (isExistOverlappedPage()) {
+ return true;
+ }
+
+ // make sure firstPageReader won't be null while the unSeqPageReaders has more cached page
+ // readers
+ while (firstPageReader == null && (!seqPageReaders.isEmpty() || !unSeqPageReaders.isEmpty())) {
+
+ initFirstPageReader();
+
+ if (isExistOverlappedPage()) {
+ return true;
+ }
+ }
+ return firstPageReader != null;
+ }
+
+ private boolean isExistOverlappedPage() throws IOException {
+ if (firstPageOverlapped()) {
+ /*
+ * next page is overlapped, read overlapped data and cache it
+ */
+ if (hasNextOverlappedPage()) {
+ cachedTsBlock = nextOverlappedPage();
+ if (cachedTsBlock != null && cachedTsBlock.hasNext()) {
+ hasCachedNextOverlappedPage = true;
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private boolean firstPageOverlapped() throws IOException {
+ if (firstPageReader == null) {
+ return false;
+ }
+
+ long endpointTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics());
+ unpackAllOverlappedTsFilesToTimeSeriesMetadata(endpointTime);
+ unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(endpointTime, false);
+ unpackAllOverlappedChunkMetadataToPageReaders(endpointTime, false);
+
+ return (!seqPageReaders.isEmpty()
+ && orderUtils.isOverlapped(
+ firstPageReader.getStatistics(), seqPageReaders.get(0).getStatistics()))
+ || (!unSeqPageReaders.isEmpty()
+ && orderUtils.isOverlapped(
+ firstPageReader.getStatistics(), unSeqPageReaders.peek().getStatistics())
+ || (mergeReader.hasNextTimeValuePair()
+ && orderUtils.isOverlapped(
+ mergeReader.currentTimeValuePair().getTimestamp(),
+ firstPageReader.getStatistics())));
+ }
+
+ private void unpackAllOverlappedChunkMetadataToPageReaders(long endpointTime, boolean init)
+ throws IOException {
+ if (firstChunkMetadata != null
+ && orderUtils.isOverlapped(endpointTime, firstChunkMetadata.getStatistics())) {
+ unpackOneChunkMetaData(firstChunkMetadata);
+ firstChunkMetadata = null;
+ }
+ // In case unpacking too many sequence chunks
+ boolean hasMeetSeq = false;
+ while (!cachedChunkMetadata.isEmpty()
+ && orderUtils.isOverlapped(endpointTime, cachedChunkMetadata.peek().getStatistics())) {
+ if (cachedChunkMetadata.peek().isSeq() && hasMeetSeq) {
+ break;
+ } else if (cachedChunkMetadata.peek().isSeq()) {
+ hasMeetSeq = true;
+ }
+ unpackOneChunkMetaData(cachedChunkMetadata.poll());
+ }
+ if (init
+ && firstPageReader == null
+ && (!seqPageReaders.isEmpty() || !unSeqPageReaders.isEmpty())) {
+ initFirstPageReader();
+ }
+ }
+
+ private void unpackOneChunkMetaData(IChunkMetadata chunkMetaData) throws IOException {
+ List<IPageReader> pageReaderList =
+ FileLoaderUtils.loadPageReaderList(chunkMetaData, timeFilter);
+
+ if (chunkMetaData.isSeq()) {
+ if (orderUtils.getAscending()) {
+ for (IPageReader iPageReader : pageReaderList) {
+ seqPageReaders.add(
+ new VersionPageReader(
+ chunkMetaData.getVersion(),
+ chunkMetaData.getOffsetOfChunkHeader(),
+ iPageReader,
+ true));
+ }
+ } else {
+ for (int i = pageReaderList.size() - 1; i >= 0; i--) {
+ seqPageReaders.add(
+ new VersionPageReader(
+ chunkMetaData.getVersion(),
+ chunkMetaData.getOffsetOfChunkHeader(),
+ pageReaderList.get(i),
+ true));
+ }
+ }
+ } else {
+ pageReaderList.forEach(
+ pageReader ->
+ unSeqPageReaders.add(
+ new VersionPageReader(
+ chunkMetaData.getVersion(),
+ chunkMetaData.getOffsetOfChunkHeader(),
+ pageReader,
+ false)));
+ }
+ }
+
+ /**
+ * This method should be called after calling hasNextPage.
+ *
+ * <p>hasNextPage may cache firstPageReader if it is not overlapped or cached a BatchData if the
+ * first page is overlapped
+ */
+ boolean isPageOverlapped() throws IOException {
+
+ /*
+ * has an overlapped page
+ */
+ if (hasCachedNextOverlappedPage) {
+ return true;
+ }
+
+ /*
+ * has a non-overlapped page in firstPageReader
+ */
+ if (mergeReader.hasNextTimeValuePair()
+ && ((orderUtils.getAscending()
+ && mergeReader.currentTimeValuePair().getTimestamp()
+ <= firstPageReader.getStatistics().getEndTime())
+ || (!orderUtils.getAscending()
+ && mergeReader.currentTimeValuePair().getTimestamp()
+ >= firstPageReader.getStatistics().getStartTime()))) {
+ throw new IOException("overlapped data should be consumed first");
+ }
+
+ Statistics firstPageStatistics = firstPageReader.getStatistics();
+
+ return !unSeqPageReaders.isEmpty()
+ && orderUtils.isOverlapped(firstPageStatistics, unSeqPageReaders.peek().getStatistics());
+ }
+
+ Statistics currentPageStatistics() {
+ if (firstPageReader == null) {
+ return null;
+ }
+ return firstPageReader.getStatistics();
+ }
+
+ Statistics currentPageStatistics(int index) throws IOException {
+ if (firstPageReader == null) {
+ return null;
+ }
+ if (!(firstPageReader.isAlignedPageReader())) {
+ throw new IOException("Can only get statistics by index from AlignedPageReader");
+ }
+ return firstPageReader.getStatistics(index);
+ }
+
+ Statistics currentPageTimeStatistics() throws IOException {
+ if (firstPageReader == null) {
+ return null;
+ }
+ if (!(firstPageReader.isAlignedPageReader())) {
+ throw new IOException("Can only get statistics of time column from AlignedPageReader");
+ }
+ return firstPageReader.getTimeStatistics();
+ }
+
+ boolean currentPageModified() throws IOException {
+ if (firstPageReader == null) {
+ throw new IOException("no first page");
+ }
+ return firstPageReader.isModified();
+ }
+
+ void skipCurrentPage() {
+ firstPageReader = null;
+ }
+
+ /** This method should only be used when the method isPageOverlapped() return true. */
+ TsBlock nextPage() throws IOException {
+
+ if (hasCachedNextOverlappedPage) {
+ hasCachedNextOverlappedPage = false;
+ return cachedTsBlock;
+ } else {
+
+ /*
+ * next page is not overlapped, push down value filter if it exists
+ */
+ if (valueFilter != null) {
+ firstPageReader.setFilter(valueFilter);
+ }
+ TsBlock tsBlock = firstPageReader.getAllSatisfiedPageData(orderUtils.getAscending());
+ firstPageReader = null;
+
+ return tsBlock;
+ }
+ }
+
+ /**
+ * read overlapped data till currentLargestEndTime in mergeReader, if current batch does not
+ * contain data, read till next currentLargestEndTime again
+ */
+ @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
+ private boolean hasNextOverlappedPage() throws IOException {
+
+ if (hasCachedNextOverlappedPage) {
+ return true;
+ }
+
+ tryToPutAllDirectlyOverlappedUnseqPageReadersIntoMergeReader();
+
+ while (true) {
+
+ // may has overlapped data
+ if (mergeReader.hasNextTimeValuePair()) {
+
+ // TODO we still need to consider data type, ascending and descending here
+ cachedTsBlock = new TsBlock();
+ long currentPageEndPointTime = mergeReader.getCurrentReadStopTime();
+ while (mergeReader.hasNextTimeValuePair()) {
+
+ /*
+ * get current first point in mergeReader, this maybe overlapped later
+ */
+ TimeValuePair timeValuePair = mergeReader.currentTimeValuePair();
+
+ if (orderUtils.isExcessEndpoint(timeValuePair.getTimestamp(), currentPageEndPointTime)) {
+ /*
+ * when the merged point excesses the currentPageEndPointTime, we have read all overlapped data before currentPageEndPointTime
+ * 1. has cached batch data, we don't need to read more data, just use the cached data later
+ * 2. has first page reader, which means first page reader last endTime < currentTimeValuePair.getTimestamp(),
+ * we could just use the first page reader later
+ * 3. sequence page reader is not empty, which means first page reader last endTime < currentTimeValuePair.getTimestamp(),
+ * we could use the first sequence page reader later
+ */
+ if (cachedTsBlock.hasNext() || firstPageReader != null || !seqPageReaders.isEmpty()) {
+ break;
+ }
+ // so, we don't have other data except mergeReader
+ currentPageEndPointTime = mergeReader.getCurrentReadStopTime();
+ }
+
+ // unpack all overlapped data for the first timeValuePair
+ unpackAllOverlappedTsFilesToTimeSeriesMetadata(timeValuePair.getTimestamp());
+ unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(
+ timeValuePair.getTimestamp(), false);
+ unpackAllOverlappedChunkMetadataToPageReaders(timeValuePair.getTimestamp(), false);
+ unpackAllOverlappedUnseqPageReadersToMergeReader(timeValuePair.getTimestamp());
+
+ // update if there are unpacked unSeqPageReaders
+ timeValuePair = mergeReader.currentTimeValuePair();
+
+ // from now, the unsequence reader is all unpacked, so we don't need to consider it
+ // we has first page reader now
+ if (firstPageReader != null) {
+ // if current timeValuePair excesses the first page reader's end time, we just use the
+ // cached data
+ if ((orderUtils.getAscending()
+ && timeValuePair.getTimestamp() > firstPageReader.getStatistics().getEndTime())
+ || (!orderUtils.getAscending()
+ && timeValuePair.getTimestamp()
+ < firstPageReader.getStatistics().getStartTime())) {
+ hasCachedNextOverlappedPage = cachedTsBlock.hasNext();
+ return hasCachedNextOverlappedPage;
+ } else if (orderUtils.isOverlapped(
+ timeValuePair.getTimestamp(), firstPageReader.getStatistics())) {
+ // current timeValuePair is overlapped with firstPageReader, add it to merged reader
+ // and update endTime to the max end time
+ mergeReader.addReader(
+ firstPageReader
+ .getAllSatisfiedPageData(orderUtils.getAscending())
+ .getTsBlockIterator(),
+ firstPageReader.version,
+ orderUtils.getOverlapCheckTime(firstPageReader.getStatistics()),
+ context);
+ currentPageEndPointTime =
+ updateEndPointTime(currentPageEndPointTime, firstPageReader);
+ firstPageReader = null;
+ }
+ }
+
+ // the seq page readers is not empty, just like first page reader
+ if (!seqPageReaders.isEmpty()) {
+ if ((orderUtils.getAscending()
+ && timeValuePair.getTimestamp()
+ > seqPageReaders.get(0).getStatistics().getEndTime())
+ || (!orderUtils.getAscending()
+ && timeValuePair.getTimestamp()
+ < seqPageReaders.get(0).getStatistics().getStartTime())) {
+ hasCachedNextOverlappedPage = cachedTsBlock.hasNext();
+ return hasCachedNextOverlappedPage;
+ } else if (orderUtils.isOverlapped(
+ timeValuePair.getTimestamp(), seqPageReaders.get(0).getStatistics())) {
+ VersionPageReader pageReader = seqPageReaders.remove(0);
+ mergeReader.addReader(
+ pageReader
+ .getAllSatisfiedPageData(orderUtils.getAscending())
+ .getTsBlockIterator(),
+ pageReader.version,
+ orderUtils.getOverlapCheckTime(pageReader.getStatistics()),
+ context);
+ currentPageEndPointTime = updateEndPointTime(currentPageEndPointTime, pageReader);
+ }
+ }
+
+ /*
+ * get the latest first point in mergeReader
+ */
+ timeValuePair = mergeReader.nextTimeValuePair();
+
+ Object valueForFilter = timeValuePair.getValue().getValue();
+
+ // TODO fix value filter firstNotNullObject, currently, if it's a value filter, it will
+ // only accept AlignedPath with only one sub sensor
+ if (timeValuePair.getValue().getDataType() == TSDataType.VECTOR) {
+ for (TsPrimitiveType tsPrimitiveType : timeValuePair.getValue().getVector()) {
+ if (tsPrimitiveType != null) {
+ valueForFilter = tsPrimitiveType.getValue();
+ break;
+ }
+ }
+ }
+
+ if (valueFilter == null
+ || valueFilter.satisfy(timeValuePair.getTimestamp(), valueForFilter)) {
+ switch (dataType) {
+ case BOOLEAN:
+ // tsBlock.putBoolean(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getBoolean());
+ break;
+ case INT32:
+ // tsBlock.putInt(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getInt());
+ break;
+ case INT64:
+ // tsBlock.putLong(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getLong());
+ break;
+ case FLOAT:
+ // tsBlock.putFloat(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getFloat());
+ break;
+ case DOUBLE:
+ // tsBlock.putDouble(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getDouble());
+ break;
+ case TEXT:
+ // tsBlock.putBinary(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getBinary());
+ break;
+ default:
+ throw new UnSupportedDataTypeException(String.valueOf(dataType));
+ }
+ }
+ }
+ hasCachedNextOverlappedPage = cachedTsBlock.hasNext();
+ /*
+ * if current overlapped page has valid data, return, otherwise read next overlapped page
+ */
+ if (hasCachedNextOverlappedPage) {
+ return true;
+ } else if (mergeReader.hasNextTimeValuePair()) {
+ // condition: seqPage.endTime < mergeReader.currentTime
+ return false;
+ }
+ } else {
+ return false;
+ }
+ }
+ }
+
+ private long updateEndPointTime(long currentPageEndPointTime, VersionPageReader pageReader) {
+ if (orderUtils.getAscending()) {
+ return Math.min(currentPageEndPointTime, pageReader.getStatistics().getEndTime());
+ } else {
+ return Math.max(currentPageEndPointTime, pageReader.getStatistics().getStartTime());
+ }
+ }
+
+ private void tryToPutAllDirectlyOverlappedUnseqPageReadersIntoMergeReader() throws IOException {
+
+ /*
+ * no cached page readers
+ */
+ if (firstPageReader == null && unSeqPageReaders.isEmpty() && seqPageReaders.isEmpty()) {
+ return;
+ }
+
+ /*
+ * init firstPageReader
+ */
+ if (firstPageReader == null) {
+ initFirstPageReader();
+ }
+
+ long currentPageEndpointTime;
+ if (mergeReader.hasNextTimeValuePair()) {
+ currentPageEndpointTime = mergeReader.getCurrentReadStopTime();
+ } else {
+ currentPageEndpointTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics());
+ }
+
+ /*
+ * put all currently directly overlapped unseq page reader to merge reader
+ */
+ unpackAllOverlappedUnseqPageReadersToMergeReader(currentPageEndpointTime);
+ }
+
+ private void initFirstPageReader() throws IOException {
+ while (this.firstPageReader == null) {
+ VersionPageReader firstPageReader = getFirstPageReaderFromCachedReaders();
+
+ // unpack overlapped page using current page reader
+ if (firstPageReader != null) {
+ long overlapCheckTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics());
+ unpackAllOverlappedTsFilesToTimeSeriesMetadata(overlapCheckTime);
+ unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(overlapCheckTime, false);
+ unpackAllOverlappedChunkMetadataToPageReaders(overlapCheckTime, false);
+
+ // this page after unpacking must be the first page
+ if (firstPageReader.equals(getFirstPageReaderFromCachedReaders())) {
+ this.firstPageReader = firstPageReader;
+ if (!seqPageReaders.isEmpty() && firstPageReader.equals(seqPageReaders.get(0))) {
+ seqPageReaders.remove(0);
+ break;
+ } else if (!unSeqPageReaders.isEmpty()
+ && firstPageReader.equals(unSeqPageReaders.peek())) {
+ unSeqPageReaders.poll();
+ break;
+ }
+ }
+ } else {
+ return;
+ }
+ }
+ }
+
+ // We use get() and peek() here in case it's not the first page reader before unpacking
+ private VersionPageReader getFirstPageReaderFromCachedReaders() {
+ VersionPageReader firstPageReader = null;
+ if (!seqPageReaders.isEmpty() && !unSeqPageReaders.isEmpty()) {
+ if (orderUtils.isTakeSeqAsFirst(
+ seqPageReaders.get(0).getStatistics(), unSeqPageReaders.peek().getStatistics())) {
+ firstPageReader = seqPageReaders.get(0);
+ } else {
+ firstPageReader = unSeqPageReaders.peek();
+ }
+ } else if (!seqPageReaders.isEmpty()) {
+ firstPageReader = seqPageReaders.get(0);
+ } else if (!unSeqPageReaders.isEmpty()) {
+ firstPageReader = unSeqPageReaders.peek();
+ }
+ return firstPageReader;
+ }
+
+ private void unpackAllOverlappedUnseqPageReadersToMergeReader(long endpointTime)
+ throws IOException {
+ while (!unSeqPageReaders.isEmpty()
+ && orderUtils.isOverlapped(endpointTime, unSeqPageReaders.peek().data.getStatistics())) {
+ putPageReaderToMergeReader(unSeqPageReaders.poll());
+ }
+ if (firstPageReader != null
+ && !firstPageReader.isSeq()
+ && orderUtils.isOverlapped(endpointTime, firstPageReader.getStatistics())) {
+ putPageReaderToMergeReader(firstPageReader);
+ firstPageReader = null;
+ }
+ }
+
+ private void putPageReaderToMergeReader(VersionPageReader pageReader) throws IOException {
+ mergeReader.addReader(
+ pageReader.getAllSatisfiedPageData(orderUtils.getAscending()).getTsBlockIterator(),
+ pageReader.version,
+ orderUtils.getOverlapCheckTime(pageReader.getStatistics()),
+ context);
+ }
+
+ private TsBlock nextOverlappedPage() throws IOException {
+ if (hasCachedNextOverlappedPage || hasNextOverlappedPage()) {
+ hasCachedNextOverlappedPage = false;
+ return cachedTsBlock;
+ }
+ throw new IOException("No more batch data");
+ }
+
+ private LinkedList<TsFileResource> sortUnSeqFileResources(List<TsFileResource> tsFileResources) {
+ return tsFileResources.stream()
+ .sorted(orderUtils.comparingLong(tsFileResource -> orderUtils.getOrderTime(tsFileResource)))
+ .collect(Collectors.toCollection(LinkedList::new));
+ }
+
+ /**
+ * unpack all overlapped seq/unseq files and find the first TimeSeriesMetadata
+ *
+ * <p>Because there may be too many files in the scenario used by the user, we cannot open all the
+ * chunks at once, which may cause OOM, so we can only unpack one file at a time when needed. This
+ * approach is likely to be ubiquitous, but it keeps the system running smoothly
+ */
+ @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
+ protected void tryToUnpackAllOverlappedFilesToTimeSeriesMetadata() throws IOException {
+ /*
+ * Fill sequence TimeSeriesMetadata List until it is not empty
+ */
+ while (seqTimeSeriesMetadata.isEmpty() && orderUtils.hasNextSeqResource()) {
+ unpackSeqTsFileResource();
+ }
+
+ /*
+ * Fill unSequence TimeSeriesMetadata Priority Queue until it is not empty
+ */
+ while (unSeqTimeSeriesMetadata.isEmpty() && orderUtils.hasNextUnseqResource()) {
+ unpackUnseqTsFileResource();
+ }
+
+ /*
+ * find end time of the first TimeSeriesMetadata
+ */
+ long endTime = -1L;
+ if (!seqTimeSeriesMetadata.isEmpty() && unSeqTimeSeriesMetadata.isEmpty()) {
+ // only has seq
+ endTime = orderUtils.getOverlapCheckTime(seqTimeSeriesMetadata.get(0).getStatistics());
+ } else if (seqTimeSeriesMetadata.isEmpty() && !unSeqTimeSeriesMetadata.isEmpty()) {
+ // only has unseq
+ endTime = orderUtils.getOverlapCheckTime(unSeqTimeSeriesMetadata.peek().getStatistics());
+ } else if (!seqTimeSeriesMetadata.isEmpty()) {
+ // has seq and unseq
+ endTime =
+ orderUtils.getCurrentEndPoint(
+ seqTimeSeriesMetadata.get(0).getStatistics(),
+ unSeqTimeSeriesMetadata.peek().getStatistics());
+ }
+
+ /*
+ * unpack all directly overlapped seq/unseq files with first TimeSeriesMetadata
+ */
+ if (endTime != -1) {
+ unpackAllOverlappedTsFilesToTimeSeriesMetadata(endTime);
+ }
+
+ /*
+ * update the first TimeSeriesMetadata
+ */
+ if (!seqTimeSeriesMetadata.isEmpty() && unSeqTimeSeriesMetadata.isEmpty()) {
+ // only has seq
+ firstTimeSeriesMetadata = seqTimeSeriesMetadata.remove(0);
+ } else if (seqTimeSeriesMetadata.isEmpty() && !unSeqTimeSeriesMetadata.isEmpty()) {
+ // only has unseq
+ firstTimeSeriesMetadata = unSeqTimeSeriesMetadata.poll();
+ } else if (!seqTimeSeriesMetadata.isEmpty()) {
+ // has seq and unseq
+ if (orderUtils.isTakeSeqAsFirst(
+ seqTimeSeriesMetadata.get(0).getStatistics(),
+ unSeqTimeSeriesMetadata.peek().getStatistics())) {
+ firstTimeSeriesMetadata = seqTimeSeriesMetadata.remove(0);
+ } else {
+ firstTimeSeriesMetadata = unSeqTimeSeriesMetadata.poll();
+ }
+ }
+ if (valueFilter != null
+ && firstTimeSeriesMetadata != null
+ && !isFileOverlapped()
+ && !firstTimeSeriesMetadata.isModified()
+ && !valueFilter.satisfy(firstTimeSeriesMetadata.getStatistics())) {
+ firstTimeSeriesMetadata = null;
+ }
+ }
+
+ protected void unpackAllOverlappedTsFilesToTimeSeriesMetadata(long endpointTime)
+ throws IOException {
+ while (orderUtils.hasNextUnseqResource()
+ && orderUtils.isOverlapped(endpointTime, orderUtils.getNextUnseqFileResource(false))) {
+ unpackUnseqTsFileResource();
+ }
+ while (orderUtils.hasNextSeqResource()
+ && orderUtils.isOverlapped(endpointTime, orderUtils.getNextSeqFileResource(false))) {
+ unpackSeqTsFileResource();
+ }
+ }
+
+ protected void unpackSeqTsFileResource() throws IOException {
+ ITimeSeriesMetadata timeseriesMetadata =
+ loadTimeSeriesMetadata(
+ orderUtils.getNextSeqFileResource(true),
+ seriesPath,
+ context,
+ getAnyFilter(),
+ allSensors);
+ if (timeseriesMetadata != null) {
+ timeseriesMetadata.setSeq(true);
+ seqTimeSeriesMetadata.add(timeseriesMetadata);
+ }
+ }
+
+ protected void unpackUnseqTsFileResource() throws IOException {
+ ITimeSeriesMetadata timeseriesMetadata =
+ loadTimeSeriesMetadata(
+ orderUtils.getNextUnseqFileResource(true),
+ seriesPath,
+ context,
+ getAnyFilter(),
+ allSensors);
+ if (timeseriesMetadata != null) {
+ timeseriesMetadata.setModified(true);
+ timeseriesMetadata.setSeq(false);
+ unSeqTimeSeriesMetadata.add(timeseriesMetadata);
+ }
+ }
+
+ protected ITimeSeriesMetadata loadTimeSeriesMetadata(
+ TsFileResource resource,
+ PartialPath seriesPath,
+ QueryContext context,
+ Filter filter,
+ Set<String> allSensors)
+ throws IOException {
+ return FileLoaderUtils.loadTimeSeriesMetadata(
+ resource, seriesPath, context, filter, allSensors);
+ }
+
+ protected Filter getAnyFilter() {
+ return timeFilter != null ? timeFilter : valueFilter;
+ }
+
+ void setTimeFilter(long timestamp) {
+ ((UnaryFilter) timeFilter).setValue(timestamp);
+ }
+
+ Filter getTimeFilter() {
+ return timeFilter;
+ }
+
+ private class VersionPageReader {
+
+ protected PriorityMergeReader.MergeReaderPriority version;
+ protected IPageReader data;
+
+ protected boolean isSeq;
+
+ VersionPageReader(long version, long offset, IPageReader data, boolean isSeq) {
+ this.version = new PriorityMergeReader.MergeReaderPriority(version, offset);
+ this.data = data;
+ this.isSeq = isSeq;
+ }
+
+ public boolean isAlignedPageReader() {
+ return data instanceof IAlignedPageReader;
+ }
+
+ Statistics getStatistics() {
+ return data.getStatistics();
+ }
+
+ Statistics getStatistics(int index) throws IOException {
+ if (!(data instanceof IAlignedPageReader)) {
+ throw new IOException("Can only get statistics by index from AlignedPageReader");
+ }
+ return ((IAlignedPageReader) data).getStatistics(index);
+ }
+
+ Statistics getTimeStatistics() throws IOException {
+ if (!(data instanceof IAlignedPageReader)) {
+ throw new IOException("Can only get statistics of time column from AlignedPageReader");
+ }
+ return ((IAlignedPageReader) data).getTimeStatistics();
+ }
+
+ TsBlock getAllSatisfiedPageData(boolean ascending) throws IOException {
+ return data.getAllSatisfiedData(ascending);
+ }
+
+ void setFilter(Filter filter) {
+ data.setFilter(filter);
+ }
+
+ boolean isModified() {
+ return data.isModified();
+ }
+
+ public boolean isSeq() {
+ return isSeq;
+ }
+ }
+
+ public interface TimeOrderUtils {
+
+ long getOrderTime(Statistics<? extends Object> statistics);
+
+ long getOrderTime(TsFileResource fileResource);
+
+ long getOverlapCheckTime(Statistics<? extends Object> range);
+
+ boolean isOverlapped(Statistics<? extends Object> left, Statistics<? extends Object> right);
+
+ boolean isOverlapped(long time, Statistics<? extends Object> right);
+
+ boolean isOverlapped(long time, TsFileResource right);
+
+ <T> Comparator<T> comparingLong(ToLongFunction<? super T> keyExtractor);
+
+ long getCurrentEndPoint(long time, Statistics<? extends Object> statistics);
+
+ long getCurrentEndPoint(
+ Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics);
+
+ boolean isExcessEndpoint(long time, long endpointTime);
+
+ /** Return true if taking first page reader from seq readers */
+ boolean isTakeSeqAsFirst(
+ Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics);
+
+ boolean getAscending();
+
+ boolean hasNextSeqResource();
+
+ boolean hasNextUnseqResource();
+
+ TsFileResource getNextSeqFileResource(boolean isDelete);
+
+ TsFileResource getNextUnseqFileResource(boolean isDelete);
+ }
+
+ class DescTimeOrderUtils implements TimeOrderUtils {
+
+ @Override
+ public long getOrderTime(Statistics statistics) {
+ return statistics.getEndTime();
+ }
+
+ @Override
+ public long getOrderTime(TsFileResource fileResource) {
+ return fileResource.getEndTime(seriesPath.getDevice());
+ }
+
+ @Override
+ public long getOverlapCheckTime(Statistics range) {
+ return range.getStartTime();
+ }
+
+ @Override
+ public boolean isOverlapped(Statistics left, Statistics right) {
+ return left.getStartTime() <= right.getEndTime();
+ }
+
+ @Override
+ public boolean isOverlapped(long time, Statistics right) {
+ return time <= right.getEndTime();
+ }
+
+ @Override
+ public boolean isOverlapped(long time, TsFileResource right) {
+ return time <= right.getEndTime(seriesPath.getDevice());
+ }
+
+ @Override
+ public <T> Comparator<T> comparingLong(ToLongFunction<? super T> keyExtractor) {
+ Objects.requireNonNull(keyExtractor);
+ return (Comparator<T> & Serializable)
+ (c1, c2) -> Long.compare(keyExtractor.applyAsLong(c2), keyExtractor.applyAsLong(c1));
+ }
+
+ @Override
+ public long getCurrentEndPoint(long time, Statistics<? extends Object> statistics) {
+ return Math.max(time, statistics.getStartTime());
+ }
+
+ @Override
+ public long getCurrentEndPoint(
+ Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) {
+ return Math.max(seqStatistics.getStartTime(), unseqStatistics.getStartTime());
+ }
+
+ @Override
+ public boolean isExcessEndpoint(long time, long endpointTime) {
+ return time < endpointTime;
+ }
+
+ @Override
+ public boolean isTakeSeqAsFirst(
+ Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) {
+ return seqStatistics.getEndTime() > unseqStatistics.getEndTime();
+ }
+
+ @Override
+ public boolean getAscending() {
+ return false;
+ }
+
+ @Override
+ public boolean hasNextSeqResource() {
+ while (dataSource.hasNextSeqResource(curSeqFileIndex, getAscending())) {
+ TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex);
+ if (tsFileResource != null
+ && tsFileResource.isSatisfied(seriesPath.getDevice(), timeFilter, null, true, false)) {
+ break;
+ }
+ curSeqFileIndex--;
+ }
+ return dataSource.hasNextSeqResource(curSeqFileIndex, getAscending());
+ }
+
+ @Override
+ public boolean hasNextUnseqResource() {
+ while (dataSource.hasNextUnseqResource(curUnseqFileIndex)) {
+ TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex);
+ if (tsFileResource != null
+ && tsFileResource.isSatisfied(seriesPath.getDevice(), timeFilter, null, false, false)) {
+ break;
+ }
+ curUnseqFileIndex++;
+ }
+ return dataSource.hasNextUnseqResource(curUnseqFileIndex);
+ }
+
+ @Override
+ public TsFileResource getNextSeqFileResource(boolean isDelete) {
+ TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex);
+ if (isDelete) {
+ curSeqFileIndex--;
+ }
+ return tsFileResource;
+ }
+
+ @Override
+ public TsFileResource getNextUnseqFileResource(boolean isDelete) {
+ TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex);
+ if (isDelete) {
+ curUnseqFileIndex++;
+ }
+ return tsFileResource;
+ }
+ }
+
+ class AscTimeOrderUtils implements TimeOrderUtils {
+
+ @Override
+ public long getOrderTime(Statistics statistics) {
+ return statistics.getStartTime();
+ }
+
+ @Override
+ public long getOrderTime(TsFileResource fileResource) {
+ return fileResource.getStartTime(seriesPath.getDevice());
+ }
+
+ @Override
+ public long getOverlapCheckTime(Statistics range) {
+ return range.getEndTime();
+ }
+
+ @Override
+ public boolean isOverlapped(Statistics left, Statistics right) {
+ return left.getEndTime() >= right.getStartTime();
+ }
+
+ @Override
+ public boolean isOverlapped(long time, Statistics right) {
+ return time >= right.getStartTime();
+ }
+
+ @Override
+ public boolean isOverlapped(long time, TsFileResource right) {
+ return time >= right.getStartTime(seriesPath.getDevice());
+ }
+
+ @Override
+ public <T> Comparator<T> comparingLong(ToLongFunction<? super T> keyExtractor) {
+ Objects.requireNonNull(keyExtractor);
+ return (Comparator<T> & Serializable)
+ (c1, c2) -> Long.compare(keyExtractor.applyAsLong(c1), keyExtractor.applyAsLong(c2));
+ }
+
+ @Override
+ public long getCurrentEndPoint(long time, Statistics<? extends Object> statistics) {
+ return Math.min(time, statistics.getEndTime());
+ }
+
+ @Override
+ public long getCurrentEndPoint(
+ Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) {
+ return Math.min(seqStatistics.getEndTime(), unseqStatistics.getEndTime());
+ }
+
+ @Override
+ public boolean isExcessEndpoint(long time, long endpointTime) {
+ return time > endpointTime;
+ }
+
+ @Override
+ public boolean isTakeSeqAsFirst(
+ Statistics<? extends Object> seqStatistics, Statistics<? extends Object> unseqStatistics) {
+ return seqStatistics.getStartTime() < unseqStatistics.getStartTime();
+ }
+
+ @Override
+ public boolean getAscending() {
+ return true;
+ }
+
+ @Override
+ public boolean hasNextSeqResource() {
+ while (dataSource.hasNextSeqResource(curSeqFileIndex, getAscending())) {
+ TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex);
+ if (tsFileResource != null
+ && tsFileResource.isSatisfied(seriesPath.getDevice(), timeFilter, null, true, false)) {
+ break;
+ }
+ curSeqFileIndex++;
+ }
+ return dataSource.hasNextSeqResource(curSeqFileIndex, getAscending());
+ }
+
+ @Override
+ public boolean hasNextUnseqResource() {
+ while (dataSource.hasNextUnseqResource(curUnseqFileIndex)) {
+ TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex);
+ if (tsFileResource != null
+ && tsFileResource.isSatisfied(seriesPath.getDevice(), timeFilter, null, false, false)) {
+ break;
+ }
+ curUnseqFileIndex++;
+ }
+ return dataSource.hasNextUnseqResource(curUnseqFileIndex);
+ }
+
+ @Override
+ public TsFileResource getNextSeqFileResource(boolean isDelete) {
+ TsFileResource tsFileResource = dataSource.getSeqResourceByIndex(curSeqFileIndex);
+ if (isDelete) {
+ curSeqFileIndex++;
+ }
+ return tsFileResource;
+ }
+
+ @Override
+ public TsFileResource getNextUnseqFileResource(boolean isDelete) {
+ TsFileResource tsFileResource = dataSource.getUnseqResourceByIndex(curUnseqFileIndex);
+ if (isDelete) {
+ curUnseqFileIndex++;
+ }
+ return tsFileResource;
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/context/QueryContext.java b/server/src/main/java/org/apache/iotdb/db/query/context/QueryContext.java
index df1b008..0c87e9e 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/context/QueryContext.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/context/QueryContext.java
@@ -39,14 +39,14 @@ public class QueryContext {
* The outer key is the path of a ModificationFile, the inner key in the name of a timeseries and
* the value is the Modifications of a timeseries in this file.
*/
- private final Map<String, Map<String, List<Modification>>> filePathModCache =
+ protected final Map<String, Map<String, List<Modification>>> filePathModCache =
new ConcurrentHashMap<>();
/**
* The key is the path of a ModificationFile and the value is all Modifications in this file. We
* use this field because each call of Modification.getModifications() return a copy of the
* Modifications, and we do not want it to create multiple copies within a query.
*/
- private final Map<String, List<Modification>> fileModCache = new HashMap<>();
+ protected final Map<String, List<Modification>> fileModCache = new HashMap<>();
private long queryId;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemAlignedPageReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemAlignedPageReader.java
index d863fb5..7216ea7 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemAlignedPageReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemAlignedPageReader.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.read.common.BatchDataFactory;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
import org.apache.iotdb.tsfile.read.reader.IAlignedPageReader;
@@ -79,6 +80,12 @@ public class MemAlignedPageReader implements IPageReader, IAlignedPageReader {
}
@Override
+ public TsBlock getAllSatisfiedData(boolean ascending) throws IOException {
+ // TODO need to implement for mpp
+ throw new IllegalStateException("We have not implemented this method.");
+ }
+
+ @Override
public Statistics getStatistics() {
return chunkMetadata.getStatistics();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemPageReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemPageReader.java
index ea64694..79997b5 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemPageReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemPageReader.java
@@ -18,12 +18,14 @@
*/
package org.apache.iotdb.db.query.reader.chunk;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.read.common.BatchDataFactory;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
import org.apache.iotdb.tsfile.read.reader.IPageReader;
@@ -60,6 +62,50 @@ public class MemPageReader implements IPageReader {
}
@Override
+ public TsBlock getAllSatisfiedData(boolean ascending) throws IOException {
+ TSDataType dataType = chunkMetadata.getDataType();
+ // TODO we still need to consider data type, ascending and descending here
+
+ TsBlock tsBlock = new TsBlock();
+ while (timeValuePairIterator.hasNextTimeValuePair()) {
+ TimeValuePair timeValuePair = timeValuePairIterator.nextTimeValuePair();
+ if (valueFilter == null
+ || valueFilter.satisfy(
+ timeValuePair.getTimestamp(), timeValuePair.getValue().getValue())) {
+ switch (dataType) {
+ case BOOLEAN:
+ // tsBlock.putBoolean(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getBoolean());
+ break;
+ case INT32:
+ // tsBlock.putInt(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getInt());
+ break;
+ case INT64:
+ // tsBlock.putLong(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getLong());
+ break;
+ case FLOAT:
+ // tsBlock.putFloat(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getFloat());
+ break;
+ case DOUBLE:
+ // tsBlock.putDouble(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getDouble());
+ break;
+ case TEXT:
+ // tsBlock.putBinary(timeValuePair.getTimestamp(),
+ // timeValuePair.getValue().getBinary());
+ break;
+ default:
+ throw new UnSupportedDataTypeException(String.valueOf(dataType));
+ }
+ }
+ }
+ return tsBlock;
+ }
+
+ @Override
public Statistics getStatistics() {
return chunkMetadata.getStatistics();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/Column.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/Column.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/mpp/common/Column.java
rename to tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/Column.java
index 79e1136..c4933e1 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/Column.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/Column.java
@@ -16,6 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.common;
+package org.apache.iotdb.tsfile.read.common;
public interface Column {}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/TimeColumn.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TimeColumn.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/mpp/common/TimeColumn.java
rename to tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TimeColumn.java
index eecc18f..19937f6 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/TimeColumn.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TimeColumn.java
@@ -16,6 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.common;
+package org.apache.iotdb.tsfile.read.common;
public class TimeColumn implements Column {}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/TsBlock.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TsBlock.java
similarity index 70%
rename from server/src/main/java/org/apache/iotdb/db/mpp/common/TsBlock.java
rename to tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TsBlock.java
index f34faa0..a6bffbf 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/TsBlock.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TsBlock.java
@@ -16,9 +16,10 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.common;
+package org.apache.iotdb.tsfile.read.common;
-import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.reader.IPointReader;
import static java.lang.String.format;
@@ -52,10 +53,7 @@ public class TsBlock {
return false;
}
- // Get next row in current tablet
- public RowRecord getNext() {
- return null;
- }
+ public void next() {}
public TsBlockMetadata getMetadata() {
return metadata;
@@ -115,4 +113,63 @@ public class TsBlock {
return rowIndex;
}
+
+ public TsBlockIterator getTsBlockIterator() {
+ return new TsBlockIterator();
+ }
+
+ // TODO need to be implemented when the data structure is defined
+ private class TsBlockIterator implements IPointReader, IBatchDataIterator {
+
+ @Override
+ public boolean hasNext() {
+ return TsBlock.this.hasNext();
+ }
+
+ @Override
+ public boolean hasNext(long minBound, long maxBound) {
+ return hasNext();
+ }
+
+ @Override
+ public void next() {
+ TsBlock.this.next();
+ }
+
+ @Override
+ public long currentTime() {
+ return -1;
+ }
+
+ @Override
+ public Object currentValue() {
+ return null;
+ }
+
+ @Override
+ public void reset() {}
+
+ @Override
+ public int totalLength() {
+ return TsBlock.this.getCount();
+ }
+
+ @Override
+ public boolean hasNextTimeValuePair() {
+ return hasNext();
+ }
+
+ @Override
+ public TimeValuePair nextTimeValuePair() {
+ return null;
+ }
+
+ @Override
+ public TimeValuePair currentTimeValuePair() {
+ return null;
+ }
+
+ @Override
+ public void close() {}
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/TsBlockMetadata.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TsBlockMetadata.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/mpp/common/TsBlockMetadata.java
rename to tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TsBlockMetadata.java
index 443c236..7d3a5dd 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/TsBlockMetadata.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/common/TsBlockMetadata.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.common;
+package org.apache.iotdb.tsfile.read.common;
import java.util.List;
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/IPageReader.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/IPageReader.java
index 93a39cf..db76550 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/IPageReader.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/IPageReader.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.tsfile.read.reader;
import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import java.io.IOException;
@@ -32,6 +33,8 @@ public interface IPageReader {
BatchData getAllSatisfiedPageData(boolean ascending) throws IOException;
+ TsBlock getAllSatisfiedData(boolean ascending) throws IOException;
+
Statistics getStatistics();
void setFilter(Filter filter);
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java
index 86b367c..2efe6802 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/AlignedPageReader.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.read.common.BatchDataFactory;
import org.apache.iotdb.tsfile.read.common.TimeRange;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
import org.apache.iotdb.tsfile.read.reader.IAlignedPageReader;
@@ -102,6 +103,12 @@ public class AlignedPageReader implements IPageReader, IAlignedPageReader {
return pageData.flip();
}
+ @Override
+ public TsBlock getAllSatisfiedData(boolean ascending) throws IOException {
+ // TODO need to implement for mpp
+ throw new IllegalStateException("We have not implemented this method.");
+ }
+
public void setDeleteIntervalList(List<List<TimeRange>> list) {
for (int i = 0; i < valueCount; i++) {
if (valuePageReaderList.get(i) != null) {
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/PageReader.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/PageReader.java
index cb600f3..c1a4c59 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/PageReader.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/page/PageReader.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.read.common.BatchDataFactory;
import org.apache.iotdb.tsfile.read.common.TimeRange;
+import org.apache.iotdb.tsfile.read.common.TsBlock;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
import org.apache.iotdb.tsfile.read.reader.IPageReader;
@@ -154,6 +155,58 @@ public class PageReader implements IPageReader {
}
@Override
+ public TsBlock getAllSatisfiedData(boolean ascending) throws IOException {
+ // TODO we still need to consider data type, ascending and descending here
+ TsBlock tsBlock = new TsBlock();
+ if (filter == null || filter.satisfy(getStatistics())) {
+ while (timeDecoder.hasNext(timeBuffer)) {
+ long timestamp = timeDecoder.readLong(timeBuffer);
+ switch (dataType) {
+ case BOOLEAN:
+ boolean aBoolean = valueDecoder.readBoolean(valueBuffer);
+ if (!isDeleted(timestamp) && (filter == null || filter.satisfy(timestamp, aBoolean))) {
+ // tsBlock.putBoolean(timestamp, aBoolean);
+ }
+ break;
+ case INT32:
+ int anInt = valueDecoder.readInt(valueBuffer);
+ if (!isDeleted(timestamp) && (filter == null || filter.satisfy(timestamp, anInt))) {
+ // tsBlock.putInt(timestamp, anInt);
+ }
+ break;
+ case INT64:
+ long aLong = valueDecoder.readLong(valueBuffer);
+ if (!isDeleted(timestamp) && (filter == null || filter.satisfy(timestamp, aLong))) {
+ // tsBlock.putLong(timestamp, aLong);
+ }
+ break;
+ case FLOAT:
+ float aFloat = valueDecoder.readFloat(valueBuffer);
+ if (!isDeleted(timestamp) && (filter == null || filter.satisfy(timestamp, aFloat))) {
+ // tsBlock.putFloat(timestamp, aFloat);
+ }
+ break;
+ case DOUBLE:
+ double aDouble = valueDecoder.readDouble(valueBuffer);
+ if (!isDeleted(timestamp) && (filter == null || filter.satisfy(timestamp, aDouble))) {
+ // tsBlock.putDouble(timestamp, aDouble);
+ }
+ break;
+ case TEXT:
+ Binary aBinary = valueDecoder.readBinary(valueBuffer);
+ if (!isDeleted(timestamp) && (filter == null || filter.satisfy(timestamp, aBinary))) {
+ // tsBlock.putBinary(timestamp, aBinary);
+ }
+ break;
+ default:
+ throw new UnSupportedDataTypeException(String.valueOf(dataType));
+ }
+ }
+ }
+ return tsBlock;
+ }
+
+ @Override
public Statistics getStatistics() {
return pageHeader.getStatistics();
}