You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ch...@apache.org on 2024/12/29 07:39:23 UTC
(iotdb) 01/01: done
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udaf_refine
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit afe9dce7c353c01033efdba2e092cda9f4a9c2dd
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Sun Dec 29 15:39:05 2024 +0800
done
---
.../source/relational/TableAggregationTableScanOperator.java | 6 ++++++
.../operator/source/relational/aggregation/AggregationOperator.java | 1 +
.../operator/source/relational/aggregation/TableAccumulator.java | 4 ++++
.../operator/source/relational/aggregation/TableAggregator.java | 4 ++++
.../aggregation/UserDefinedAggregateFunctionAccumulator.java | 6 ++++++
.../source/relational/aggregation/grouped/GroupedAccumulator.java | 4 ++++
.../source/relational/aggregation/grouped/GroupedAggregator.java | 4 ++++
.../aggregation/grouped/GroupedUserDefinedAggregateAccumulator.java | 6 ++++++
.../relational/aggregation/grouped/HashAggregationOperator.java | 1 +
.../aggregation/grouped/StreamingAggregationOperator.java | 1 +
10 files changed, 37 insertions(+)
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 8ef9ddbbf04..c00e7f2bc64 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
@@ -865,4 +865,10 @@ public class TableAggregationTableScanOperator extends AbstractDataSourceOperato
+ (resultTsBlockBuilder == null ? 0 : resultTsBlockBuilder.getRetainedSizeInBytes())
+ RamUsageEstimator.sizeOfCollection(deviceEntries);
}
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ tableAggregators.forEach(TableAggregator::close);
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AggregationOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AggregationOperator.java
index 81abe6a62f1..d2ec8af83ff 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AggregationOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AggregationOperator.java
@@ -123,6 +123,7 @@ public class AggregationOperator implements ProcessOperator {
@Override
public void close() throws Exception {
child.close();
+ aggregators.forEach(TableAggregator::close);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java
index 82137628f03..410414dfe00 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableAccumulator.java
@@ -57,4 +57,8 @@ public interface TableAccumulator {
default boolean removable() {
return false;
}
+
+ default void close() {
+ // do nothing
+ }
}
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 0ed023d5dbd..ad6dae73bc1 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
@@ -106,4 +106,8 @@ public class TableAggregator {
public int getChannelCount() {
return this.inputChannels.length;
}
+
+ public void close() {
+ accumulator.close();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/UserDefinedAggregateFunctionAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/UserDefinedAggregateFunctionAccumulator.java
index 58a5a011b77..1bd6d6430b4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/UserDefinedAggregateFunctionAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/UserDefinedAggregateFunctionAccumulator.java
@@ -120,4 +120,10 @@ public class UserDefinedAggregateFunctionAccumulator implements TableAccumulator
public void reset() {
state.reset();
}
+
+ @Override
+ public void close() {
+ aggregateFunction.beforeDestroy();
+ state.destroyState();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAccumulator.java
index 7fe6b41e838..8f628718f68 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAccumulator.java
@@ -39,4 +39,8 @@ public interface GroupedAccumulator {
void prepareFinal();
void reset();
+
+ default void close() {
+ // do nothing
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAggregator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAggregator.java
index af24f05842d..ae94595a4e5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAggregator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAggregator.java
@@ -98,4 +98,8 @@ public class GroupedAggregator {
public long getEstimatedSize() {
return accumulator.getEstimatedSize();
}
+
+ public void close() {
+ accumulator.close();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedUserDefinedAggregateAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedUserDefinedAggregateAccumulator.java
index 90a41d53866..8e5279197ca 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedUserDefinedAggregateAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedUserDefinedAggregateAccumulator.java
@@ -131,4 +131,10 @@ public class GroupedUserDefinedAggregateAccumulator implements GroupedAccumulato
public void reset() {
stateArray.reset();
}
+
+ @Override
+ public void close() {
+ aggregateFunction.beforeDestroy();
+ stateArray.forEach(State::destroyState);
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/HashAggregationOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/HashAggregationOperator.java
index 35fbd530404..31ddcbe293b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/HashAggregationOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/HashAggregationOperator.java
@@ -197,6 +197,7 @@ public class HashAggregationOperator extends AbstractOperator {
@Override
public void close() throws Exception {
child.close();
+ aggregators.forEach(GroupedAggregator::close);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java
index 4ab86c91069..7b0d37c365f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java
@@ -234,6 +234,7 @@ public class StreamingAggregationOperator extends AbstractOperator {
@Override
public void close() throws Exception {
child.close();
+ aggregators.forEach(TableAggregator::close);
}
@Override