You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2021/12/17 03:25:18 UTC
[iotdb] branch to0.12pushdownlimit created (now 729b616)
This is an automated email from the ASF dual-hosted git repository.
xiangweiwei pushed a change to branch to0.12pushdownlimit
in repository https://gitbox.apache.org/repos/asf/iotdb.git.
at 729b616 push down limit to read task in rawdatasetwithoutValueFilter
This branch includes the following new commits:
new 729b616 push down limit to read task in rawdatasetwithoutValueFilter
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: push down limit to read task in rawdatasetwithoutValueFilter
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiangweiwei pushed a commit to branch to0.12pushdownlimit
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 729b616ff993ef690ad8cb042c26e5ba3d6a4e87
Author: Alima777 <wx...@gmail.com>
AuthorDate: Fri Dec 17 11:24:11 2021 +0800
push down limit to read task in rawdatasetwithoutValueFilter
---
.../cluster/query/ClusterDataQueryExecutor.java | 6 +----
.../dataset/RawQueryDataSetWithoutValueFilter.java | 26 ++++++++++++----------
.../UDFRawQueryInputDataSetWithoutValueFilter.java | 11 +++------
.../apache/iotdb/db/query/dataset/UDTFDataSet.java | 7 +-----
.../db/query/executor/RawDataQueryExecutor.java | 6 +----
.../iotdb/db/query/udf/core/input/InputLayer.java | 9 +++-----
6 files changed, 23 insertions(+), 42 deletions(-)
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutor.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutor.java
index 92f100a..3550bbf 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutor.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterDataQueryExecutor.java
@@ -85,11 +85,7 @@ public class ClusterDataQueryExecutor extends RawDataQueryExecutor {
try {
List<ManagedSeriesReader> readersOfSelectedSeries = initMultSeriesReader(context);
return new RawQueryDataSetWithoutValueFilter(
- context.getQueryId(),
- queryPlan.getDeduplicatedPaths(),
- queryPlan.getDeduplicatedDataTypes(),
- readersOfSelectedSeries,
- queryPlan.isAscending());
+ context.getQueryId(), queryPlan, readersOfSelectedSeries);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new StorageEngineException(e.getMessage());
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithoutValueFilter.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithoutValueFilter.java
index a6c928e..a5f613d 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithoutValueFilter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithoutValueFilter.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.query.dataset;
import org.apache.iotdb.db.concurrent.WrappedRunnable;
-import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
import org.apache.iotdb.db.query.control.QueryTimeManager;
import org.apache.iotdb.db.query.pool.QueryTaskPoolManager;
import org.apache.iotdb.db.query.reader.series.ManagedSeriesReader;
@@ -177,18 +177,23 @@ public class RawQueryDataSetWithoutValueFilter extends QueryDataSet
/**
* constructor of EngineDataSetWithoutValueFilter.
*
- * @param paths paths in List structure
- * @param dataTypes time series data type
* @param readers readers in List(IPointReader) structure
*/
public RawQueryDataSetWithoutValueFilter(
- long queryId,
- List<PartialPath> paths,
- List<TSDataType> dataTypes,
- List<ManagedSeriesReader> readers,
- boolean ascending)
+ long queryId, RawDataQueryPlan queryPlan, List<ManagedSeriesReader> readers)
throws IOException, InterruptedException {
- super(new ArrayList<>(paths), dataTypes, ascending);
+ super(
+ new ArrayList<>(queryPlan.getDeduplicatedPaths()),
+ queryPlan.getDeduplicatedDataTypes(),
+ queryPlan.isAscending());
+ this.rowLimit = queryPlan.getRowLimit();
+ this.rowOffset = queryPlan.getRowOffset();
+ this.withoutAnyNull = queryPlan.isWithoutAnyNull();
+ this.withoutAllNull = queryPlan.isWithoutAllNull();
+ if (rowLimit != 0 && !withoutAllNull && !withoutAnyNull) {
+ batchDataLengthList = new int[readers.size()];
+ }
+
this.queryId = queryId;
this.seriesReaderList = readers;
blockingQueueArray = new BlockingQueue[readers.size()];
@@ -197,9 +202,6 @@ public class RawQueryDataSetWithoutValueFilter extends QueryDataSet
}
cachedBatchDataArray = new BatchData[readers.size()];
noMoreDataInQueueArray = new boolean[readers.size()];
- if (rowLimit != 0) {
- batchDataLengthList = new int[readers.size()];
- }
init();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFRawQueryInputDataSetWithoutValueFilter.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFRawQueryInputDataSetWithoutValueFilter.java
index e7c4917..7c43df4 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFRawQueryInputDataSetWithoutValueFilter.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDFRawQueryInputDataSetWithoutValueFilter.java
@@ -19,9 +19,8 @@
package org.apache.iotdb.db.query.dataset;
-import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.query.reader.series.ManagedSeriesReader;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import java.io.IOException;
import java.util.List;
@@ -30,13 +29,9 @@ public class UDFRawQueryInputDataSetWithoutValueFilter extends RawQueryDataSetWi
implements UDFInputDataSet {
public UDFRawQueryInputDataSetWithoutValueFilter(
- long queryId,
- List<PartialPath> paths,
- List<TSDataType> dataTypes,
- List<ManagedSeriesReader> readers,
- boolean ascending)
+ long queryId, UDTFPlan udtfPlan, List<ManagedSeriesReader> readers)
throws IOException, InterruptedException {
- super(queryId, paths, dataTypes, readers, ascending);
+ super(queryId, udtfPlan, readers);
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFDataSet.java
index 5761ef2..6dfbe7b 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/UDTFDataSet.java
@@ -95,12 +95,7 @@ public abstract class UDTFDataSet extends QueryDataSet {
queryId = queryContext.getQueryId();
this.udtfPlan = udtfPlan;
inputLayer =
- new InputLayer(
- queryId,
- UDF_READER_MEMORY_BUDGET_IN_MB,
- deduplicatedPaths,
- deduplicatedDataTypes,
- readersOfSelectedSeries);
+ new InputLayer(queryId, UDF_READER_MEMORY_BUDGET_IN_MB, udtfPlan, readersOfSelectedSeries);
udtfPlan.initializeUdfExecutors(queryId, UDF_COLLECTOR_MEMORY_BUDGET_IN_MB);
initTransformers(UDF_TRANSFORMER_MEMORY_BUDGET_IN_MB);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
index ac13c1e..d480d97 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
@@ -67,11 +67,7 @@ public class RawDataQueryExecutor {
List<ManagedSeriesReader> readersOfSelectedSeries = initManagedSeriesReader(context);
try {
return new RawQueryDataSetWithoutValueFilter(
- context.getQueryId(),
- queryPlan.getDeduplicatedPaths(),
- queryPlan.getDeduplicatedDataTypes(),
- readersOfSelectedSeries,
- queryPlan.isAscending());
+ context.getQueryId(), queryPlan, readersOfSelectedSeries);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new StorageEngineException(e.getMessage());
diff --git a/server/src/main/java/org/apache/iotdb/db/query/udf/core/input/InputLayer.java b/server/src/main/java/org/apache/iotdb/db/query/udf/core/input/InputLayer.java
index f8dc99a..46eae97 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/udf/core/input/InputLayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/udf/core/input/InputLayer.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.query.udf.core.input;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
import org.apache.iotdb.db.query.dataset.RawQueryDataSetWithValueFilter;
import org.apache.iotdb.db.query.dataset.UDFInputDataSet;
import org.apache.iotdb.db.query.dataset.UDFRawQueryInputDataSetWithoutValueFilter;
@@ -62,16 +63,12 @@ public class InputLayer {
/** InputLayerWithoutValueFilter */
public InputLayer(
- long queryId,
- float memoryBudgetInMB,
- List<PartialPath> paths,
- List<TSDataType> dataTypes,
- List<ManagedSeriesReader> readers)
+ long queryId, float memoryBudgetInMB, UDTFPlan udtfPlan, List<ManagedSeriesReader> readers)
throws QueryProcessException, IOException, InterruptedException {
constructInputLayer(
queryId,
memoryBudgetInMB,
- new UDFRawQueryInputDataSetWithoutValueFilter(queryId, paths, dataTypes, readers, true));
+ new UDFRawQueryInputDataSetWithoutValueFilter(queryId, udtfPlan, readers));
}
/** InputLayerWithValueFilter */