You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2025/03/06 11:43:31 UTC
(iotdb) branch master updated: [Fix](query)(table model) GroupedAccumulator are not discovered before because of stream optimize
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 1669f503907 [Fix](query)(table model) GroupedAccumulator are not discovered before because of stream optimize
1669f503907 is described below
commit 1669f50390767358b42f3ea306f5d0d0ad38e40c
Author: Weihao Li <60...@users.noreply.github.com>
AuthorDate: Thu Mar 6 19:43:26 2025 +0800
[Fix](query)(table model) GroupedAccumulator are not discovered before because of stream optimize
---
.../it/query/recent/IoTDBTableAggregationIT.java | 4 +-
.../recent/IoTDBTableAggregationNonStreamIT.java | 56 ++++++++++++++++++++++
.../aggregation/grouped/GroupedAggregator.java | 4 +-
.../grouped/GroupedFirstAccumulator.java | 11 +++--
.../grouped/GroupedFirstByAccumulator.java | 2 +-
.../grouped/GroupedLastAccumulator.java | 1 +
.../grouped/GroupedLastByAccumulator.java | 2 +-
.../grouped/GroupedVarianceAccumulator.java | 2 +-
8 files changed, 71 insertions(+), 11 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java
index 0e1157b9238..dd47faebe97 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java
@@ -39,8 +39,8 @@ import static org.apache.iotdb.relational.it.db.it.IoTDBMultiTAGsWithAttributesT
@RunWith(IoTDBTestRunner.class)
@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
public class IoTDBTableAggregationIT {
- private static final String DATABASE_NAME = "test";
- private static final String[] createSqls =
+ protected static final String DATABASE_NAME = "test";
+ protected static final String[] createSqls =
new String[] {
"CREATE DATABASE " + DATABASE_NAME,
"USE " + DATABASE_NAME,
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationNonStreamIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationNonStreamIT.java
new file mode 100644
index 00000000000..5a402b81ff7
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationNonStreamIT.java
@@ -0,0 +1,56 @@
+/*
+ * 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.relational.it.query.recent;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+public class IoTDBTableAggregationNonStreamIT extends IoTDBTableAggregationIT {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().getConfig().getCommonConfig().setSortBufferSize(128 * 1024);
+ EnvFactory.getEnv().getConfig().getCommonConfig().setMaxTsBlockSizeInByte(4 * 1024);
+ EnvFactory.getEnv().initClusterEnvironment();
+ String original = createSqls[2];
+ // make 'province', 'city', 'region' be FIELD to cover cases using GroupedAccumulator
+ createSqls[2] =
+ "CREATE TABLE table1(province STRING FIELD, city STRING FIELD, region STRING FIELD, device_id STRING TAG, color STRING ATTRIBUTE, type STRING ATTRIBUTE, s1 INT32 FIELD, s2 INT64 FIELD, s3 FLOAT FIELD, s4 DOUBLE FIELD, s5 BOOLEAN FIELD, s6 TEXT FIELD, s7 STRING FIELD, s8 BLOB FIELD, s9 TIMESTAMP FIELD, s10 DATE FIELD)";
+ prepareTableData(createSqls);
+ // rollback original content
+ createSqls[2] = original;
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+}
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 86e92fba670..e416d4f43f0 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
@@ -76,7 +76,9 @@ public class GroupedAggregator {
if (step.isInputRaw()) {
// Use select-all AggregationMask here because filter of Agg-Function is not supported now
AggregationMask mask = AggregationMask.createSelectAll(block.getPositionCount());
-
+ if (maskChannel.isPresent()) {
+ mask.applyMaskBlock(block.getColumn(maskChannel.getAsInt()));
+ }
accumulator.addInput(groupIds, arguments, mask);
} else {
accumulator.addIntermediate(groupIds, arguments[0]);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstAccumulator.java
index bfbefd54abc..d72cf25ae69 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstAccumulator.java
@@ -246,7 +246,7 @@ public class GroupedFirstAccumulator implements GroupedAccumulator {
checkArgument(
columnBuilder instanceof BinaryColumnBuilder,
"intermediate input and output of First should be BinaryColumn");
- if (minTimes.get(groupId) == Long.MIN_VALUE) {
+ if (minTimes.get(groupId) == Long.MAX_VALUE) {
columnBuilder.appendNull();
} else {
columnBuilder.writeBinary(new Binary(serializeTimeWithValue(groupId)));
@@ -255,7 +255,7 @@ public class GroupedFirstAccumulator implements GroupedAccumulator {
@Override
public void evaluateFinal(int groupId, ColumnBuilder columnBuilder) {
- if (minTimes.get(groupId) == Long.MIN_VALUE) {
+ if (minTimes.get(groupId) == Long.MAX_VALUE) {
columnBuilder.appendNull();
} else {
switch (seriesDataType) {
@@ -360,6 +360,7 @@ public class GroupedFirstAccumulator implements GroupedAccumulator {
length += Integer.BYTES + values.length;
bytes = new byte[length];
longToBytes(minTimes.get(groupId), bytes, 0);
+ BytesUtils.intToBytes(values.length, bytes, Long.BYTES);
System.arraycopy(values, 0, bytes, length - values.length, values.length);
return bytes;
case BOOLEAN:
@@ -474,7 +475,7 @@ public class GroupedFirstAccumulator implements GroupedAccumulator {
if (mask.isSelectAll()) {
for (int i = 0; i < positionCount; i++) {
if (!valueColumn.isNull(i)) {
- updateFloatValue(groupIds[i], valueColumn.getFloat(i), timeColumn.getLong(i));
+ updateDoubleValue(groupIds[i], valueColumn.getDouble(i), timeColumn.getLong(i));
}
}
} else {
@@ -483,8 +484,8 @@ public class GroupedFirstAccumulator implements GroupedAccumulator {
for (int i = 0; i < positionCount; i++) {
position = selectedPositions[i];
if (!valueColumn.isNull(position)) {
- updateFloatValue(
- groupIds[position], valueColumn.getFloat(position), timeColumn.getLong(position));
+ updateDoubleValue(
+ groupIds[position], valueColumn.getDouble(position), timeColumn.getLong(position));
}
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstByAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstByAccumulator.java
index 79e333015ee..9a1bf72f680 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstByAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstByAccumulator.java
@@ -628,7 +628,7 @@ public class GroupedFirstByAccumulator implements GroupedAccumulator {
xNulls.set(groupId, true);
} else {
xNulls.set(groupId, false);
- xIntValues.set(groupId, xColumn.getInt(xIdx));
+ xBinaryValues.set(groupId, xColumn.getBinary(xIdx));
}
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastAccumulator.java
index 029b321365b..71e69f7a457 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastAccumulator.java
@@ -360,6 +360,7 @@ public class GroupedLastAccumulator implements GroupedAccumulator {
length += Integer.BYTES + values.length;
bytes = new byte[length];
longToBytes(maxTimes.get(groupId), bytes, 0);
+ BytesUtils.intToBytes(values.length, bytes, Long.BYTES);
System.arraycopy(values, 0, bytes, length - values.length, values.length);
return bytes;
case BOOLEAN:
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastByAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastByAccumulator.java
index a4c10874f8b..cfeca852bd2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastByAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastByAccumulator.java
@@ -627,7 +627,7 @@ public class GroupedLastByAccumulator implements GroupedAccumulator {
xNulls.set(groupId, true);
} else {
xNulls.set(groupId, false);
- xIntValues.set(groupId, xColumn.getInt(xIdx));
+ xBinaryValues.set(groupId, xColumn.getBinary(xIdx));
}
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedVarianceAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedVarianceAccumulator.java
index 117343d9494..9c244be3da2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedVarianceAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedVarianceAccumulator.java
@@ -313,7 +313,7 @@ public class GroupedVarianceAccumulator implements GroupedAccumulator {
continue;
}
- double value = column.getDouble(i);
+ double value = column.getDouble(position);
counts.increment(groupIds[position]);
double delta = value - means.get(groupIds[position]);
means.add(groupIds[position], delta / counts.get(groupIds[position]));