This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch agg_table_scan
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/agg_table_scan by this push:
new 321a4f932ca perfect aggregator and accumulator
321a4f932ca is described below
commit 321a4f932cacc8a9df54413ddf7f1446a7ccf7c9
Author: Beyyes <[email protected]>
AuthorDate: Mon Sep 30 12:01:47 2024 +0800
perfect aggregator and accumulator
---
.../execution/aggregation/TreeAggregator.java | 2 +-
.../TableAggregationTableScanOperator.java | 13 +++++++-----
.../source/relational/aggregation/Accumulator.java | 13 ++++++++++++
.../relational/aggregation/AvgAccumulator.java | 24 ++++++++++++++++++++++
.../relational/aggregation/CountAccumulator.java | 14 +++++++++++++
.../relational/aggregation/TableAggregator.java | 22 ++++++++++++++++++++
.../plan/planner/TableOperatorGenerator.java | 7 ++++++-
7 files changed, 88 insertions(+), 7 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/TreeAggregator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/TreeAggregator.java
index 769a878b32e..8c06ca299c7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/TreeAggregator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/TreeAggregator.java
@@ -43,7 +43,7 @@ public class TreeAggregator implements IAggregator {
// In some intermediate result input, inputLocation[] should include two
columns
protected List<InputLocation[]> inputLocationList;
protected final AggregationStep step;
- protected static final QueryExecutionMetricSet QUERY_EXECUTION_METRICS =
+ public static final QueryExecutionMetricSet QUERY_EXECUTION_METRICS =
QueryExecutionMetricSet.getInstance();
// Used for SeriesAggregateScanOperator
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
index b3e073941f7..2793a81325c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableAggregationTableScanOperator.java
@@ -20,7 +20,6 @@
package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
import org.apache.iotdb.commons.path.AlignedFullPath;
-import org.apache.iotdb.db.queryengine.execution.aggregation.TreeAggregator;
import
org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITimeRangeIterator;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import
org.apache.iotdb.db.queryengine.execution.operator.source.AbstractSeriesAggregationScanOperator;
@@ -371,19 +370,21 @@ public class TableAggregationTableScanOperator extends
AbstractSeriesAggregation
}
protected void calcFromStatistics(Statistics timeStatistics, Statistics[]
valueStatistics) {
- for (TreeAggregator aggregator : aggregators) {
+ for (TableAggregator aggregator : aggregators) {
if (aggregator.hasFinalResult()) {
continue;
}
- aggregator.processStatistics(timeStatistics, valueStatistics);
+ aggregator.processStatistics(valueStatistics);
}
}
+ boolean canUseStatistics = true;
+
@SuppressWarnings({"squid:S3776", "squid:S135", "squid:S3740"})
protected boolean readAndCalcFromFile() throws IOException {
// start stopwatch
long start = System.nanoTime();
- while (System.nanoTime() - start < leftRuntimeOfOneNextCall &&
seriesScanUtil.hasNextFile()) {
+ while (seriesScanUtil.hasNextFile()) {
if (canUseStatistics && seriesScanUtil.canUseCurrentFileStatistics()) {
Statistics fileTimeStatistics =
seriesScanUtil.currentFileTimeStatistics();
if (fileTimeStatistics.getStartTime() > curTimeRange.getMax()) {
@@ -424,7 +425,7 @@ public class TableAggregationTableScanOperator extends
AbstractSeriesAggregation
protected boolean readAndCalcFromChunk() throws IOException {
// start stopwatch
long start = System.nanoTime();
- while (System.nanoTime() - start < leftRuntimeOfOneNextCall &&
seriesScanUtil.hasNextChunk()) {
+ while (seriesScanUtil.hasNextChunk()) {
if (canUseStatistics && seriesScanUtil.canUseCurrentChunkStatistics()) {
Statistics chunkTimeStatistics =
seriesScanUtil.currentChunkTimeStatistics();
if (chunkTimeStatistics.getStartTime() > curTimeRange.getMax()) {
@@ -461,6 +462,8 @@ public class TableAggregationTableScanOperator extends
AbstractSeriesAggregation
return false;
}
+ long leftRuntimeOfOneNextCall = Long.MAX_VALUE;
+
@SuppressWarnings({"squid:S3776", "squid:S135", "squid:S3740"})
protected boolean readAndCalcFromPage() throws IOException {
// start stopwatch
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/Accumulator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/Accumulator.java
index 734fa21c00c..c335a4bcd39 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/Accumulator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/Accumulator.java
@@ -15,6 +15,7 @@ package
org.apache.iotdb.db.queryengine.execution.operator.source.relational.agg
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.file.metadata.statistics.Statistics;
public interface Accumulator {
long getEstimatedSize();
@@ -29,5 +30,17 @@ public interface Accumulator {
void evaluateFinal(ColumnBuilder columnBuilder);
+ /**
+ * This method can only be used in AggTableScan. For first/first_by or
last/last_by in decreasing
+ * order, we can get final result by the first record.
+ */
+ boolean hasFinalResult();
+
+ /**
+ * This method can only be used in AggTableScan, it will use different
statistics based on the
+ * type of Accumulator.
+ */
+ void addStatistics(Statistics statistics);
+
void reset();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
index d5c9eff3630..15e9d020a1f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
@@ -21,6 +21,8 @@ package
org.apache.iotdb.db.queryengine.execution.operator.source.relational.agg
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.statistics.IntegerStatistics;
+import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.read.common.block.column.BinaryColumn;
import org.apache.tsfile.read.common.block.column.BinaryColumnBuilder;
import org.apache.tsfile.utils.Binary;
@@ -182,6 +184,28 @@ public class AvgAccumulator implements Accumulator {
}
}
+ @Override
+ public boolean hasFinalResult() {
+ return false;
+ }
+
+ @Override
+ public void addStatistics(Statistics statistics) {
+ if (statistics == null) {
+ return;
+ }
+ initResult = true;
+ countValue += statistics.getCount();
+ if (statistics instanceof IntegerStatistics) {
+ sumValue += statistics.getSumLongValue();
+ } else {
+ sumValue += statistics.getSumDoubleValue();
+ }
+ if (countValue == 0) {
+ initResult = false;
+ }
+ }
+
@Override
public void reset() {
initResult = false;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
index 3f876564376..7deb2d287bf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java
@@ -20,6 +20,7 @@ package
org.apache.iotdb.db.queryengine.execution.operator.source.relational.agg
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.utils.RamUsageEstimator;
import static com.google.common.base.Preconditions.checkArgument;
@@ -73,6 +74,19 @@ public class CountAccumulator implements Accumulator {
columnBuilder.writeLong(countState);
}
+ @Override
+ public boolean hasFinalResult() {
+ return false;
+ }
+
+ @Override
+ public void addStatistics(Statistics statistics) {
+ if (statistics == null) {
+ return;
+ }
+ countState += statistics.getCount();
+ }
+
@Override
public void reset() {
countState = 0;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
index 527dab06b77..8e9726d55ae 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAggregator.java
@@ -19,6 +19,7 @@ import com.google.common.primitives.Ints;
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.read.common.block.TsBlock;
import java.util.List;
@@ -26,6 +27,8 @@ import java.util.OptionalInt;
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
+import static
org.apache.iotdb.db.queryengine.execution.aggregation.TreeAggregator.QUERY_EXECUTION_METRICS;
+import static
org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.AGGREGATION_FROM_STATISTICS;
public class TableAggregator {
private final Accumulator accumulator;
@@ -71,6 +74,25 @@ public class TableAggregator {
}
}
+ /** Used for SeriesAggregateScanOperator. */
+ public void processStatistics(Statistics[] valueStatistics) {
+ long startTime = System.nanoTime();
+ try {
+ // TODO verify the rightness
+ for (int valueIndex : inputChannels) {
+ // int valueIndex = inputLocations[0].getValueColumnIndex();
+ accumulator.addStatistics(valueStatistics[valueIndex]);
+ }
+ } finally {
+ QUERY_EXECUTION_METRICS.recordExecutionCost(
+ AGGREGATION_FROM_STATISTICS, System.nanoTime() - startTime);
+ }
+ }
+
+ public boolean hasFinalResult() {
+ return accumulator.hasFinalResult();
+ }
+
public void reset() {
accumulator.reset();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
index 74c788b325b..c4e1bc9ee8a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
@@ -51,6 +51,7 @@ import
org.apache.iotdb.db.queryengine.execution.operator.schema.source.SchemaSo
import
org.apache.iotdb.db.queryengine.execution.operator.sink.IdentitySinkOperator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanOperator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.ExchangeOperator;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableAggregationTableScanOperator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableFullOuterJoinOperator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableInnerJoinOperator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
@@ -1030,7 +1031,11 @@ public class TableOperatorGenerator extends
PlanVisitor<Operator, LocalExecution
for (Map.Entry<Symbol, AggregationNode.Aggregation> entry :
node.getAggregations().entrySet()) {
TableAggregator aggregator =
buildAggregator(
- null, childLayout, entry.getValue(), node.getStep(),
context.getTypeProvider());
+ entry.getKey(),
+ childLayout,
+ entry.getValue(),
+ node.getStep(),
+ context.getTypeProvider());
aggregators.add(aggregator);
}
