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 <[email protected]>
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]));
