This is an automated email from the ASF dual-hosted git repository.
jackietien 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 bd0716e8dd3 Support correlated quantified comparison
bd0716e8dd3 is described below
commit bd0716e8dd3e8224e2c571570fb9396cde3ddd19
Author: Liao Lanyu <[email protected]>
AuthorDate: Thu May 15 09:26:47 2025 +0800
Support correlated quantified comparison
---
.../IoTDBCorrelatedQuantifiedComparisonIT.java | 153 +++++++++++++++++++++
.../relational/aggregation/AccumulatorFactory.java | 3 +
.../grouped/GroupedCountAllAccumulator.java | 87 ++++++++++++
3 files changed, 243 insertions(+)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/subquery/correlated/IoTDBCorrelatedQuantifiedComparisonIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/subquery/correlated/IoTDBCorrelatedQuantifiedComparisonIT.java
new file mode 100644
index 00000000000..c4c41f2e196
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/subquery/correlated/IoTDBCorrelatedQuantifiedComparisonIT.java
@@ -0,0 +1,153 @@
+/*
+ * 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.subquery.correlated;
+
+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.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData;
+import static org.apache.iotdb.db.it.utils.TestUtils.tableAssertTestFail;
+import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest;
+import static
org.apache.iotdb.relational.it.query.recent.subquery.SubqueryDataSetUtils.CREATE_SQLS;
+import static
org.apache.iotdb.relational.it.query.recent.subquery.SubqueryDataSetUtils.DATABASE_NAME;
+import static
org.apache.iotdb.relational.it.query.recent.subquery.SubqueryDataSetUtils.NUMERIC_MEASUREMENTS;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+public class IoTDBCorrelatedQuantifiedComparisonIT {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().getConfig().getCommonConfig().setSortBufferSize(128 *
1024);
+
EnvFactory.getEnv().getConfig().getCommonConfig().setMaxTsBlockSizeInByte(4 *
1024);
+ EnvFactory.getEnv().initClusterEnvironment();
+ prepareTableData(CREATE_SQLS);
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void testCorrelatedQuantifiedComparisonInWhereClause() {
+ String sql;
+ String[] expectedHeader;
+ String[] retArray;
+
+ sql =
+ "SELECT cast(%s AS INT32) as %s FROM table1 t1 WHERE device_id = 'd01'
and %s >= any(SELECT %s from table3 t3 where t1.%s = t3.%s)";
+ retArray = new String[] {"30,", "40,"};
+ for (String measurement : NUMERIC_MEASUREMENTS) {
+ expectedHeader = new String[] {measurement};
+ tableResultSetEqualTest(
+ String.format(
+ sql, measurement, measurement, measurement, measurement,
measurement, measurement),
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ sql =
+ "SELECT cast(%s AS INT32) as %s FROM table1 t1 WHERE device_id = 'd01'
and %s >= some(SELECT %s from table3 t3 where t1.%s = t3.%s)";
+ retArray = new String[] {"30,", "40,"};
+ for (String measurement : NUMERIC_MEASUREMENTS) {
+ expectedHeader = new String[] {measurement};
+ tableResultSetEqualTest(
+ String.format(
+ sql, measurement, measurement, measurement, measurement,
measurement, measurement),
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ sql =
+ "SELECT cast(%s AS INT32) as %s FROM table1 t1 WHERE device_id = 'd01'
and %s >= all(SELECT %s from table3 t3 where t1.%s = t3.%s)";
+ retArray = new String[] {"30,", "40,", "50,", "60,", "70,"};
+ for (String measurement : NUMERIC_MEASUREMENTS) {
+ expectedHeader = new String[] {measurement};
+ tableResultSetEqualTest(
+ String.format(
+ sql, measurement, measurement, measurement, measurement,
measurement, measurement),
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+ }
+
+ @Test
+ public void testQuantifiedComparisonInSelectClause() {
+ String sql;
+ String[] expectedHeader;
+ String[] retArray;
+
+ sql =
+ "SELECT %s >= any(SELECT (%s) from table3 t3 WHERE device_id = 'd01'
and t1.s1 = t3.s1) from table1 t1 where device_id = 'd01'";
+ expectedHeader = new String[] {"_col0"};
+ retArray = new String[] {"true,", "true,", "false,", "false,", "false,"};
+ for (String measurement : NUMERIC_MEASUREMENTS) {
+ tableResultSetEqualTest(
+ String.format(sql, measurement, measurement), expectedHeader,
retArray, DATABASE_NAME);
+ }
+
+ sql =
+ "SELECT %s >= some(SELECT (%s) from table3 t3 WHERE device_id = 'd01'
and t1.s1 = t3.s1) from table1 t1 where device_id = 'd01'";
+ expectedHeader = new String[] {"_col0"};
+ retArray = new String[] {"true,", "true,", "false,", "false,", "false,"};
+ for (String measurement : NUMERIC_MEASUREMENTS) {
+ tableResultSetEqualTest(
+ String.format(sql, measurement, measurement), expectedHeader,
retArray, DATABASE_NAME);
+ }
+
+ sql =
+ "SELECT %s >= all(SELECT (%s) from table3 t3 WHERE device_id = 'd01'
and t1.s1 = t3.s1) from table1 t1 where device_id = 'd01'";
+ expectedHeader = new String[] {"_col0"};
+ retArray = new String[] {"true,", "true,", "true,", "true,", "true,"};
+ for (String measurement : NUMERIC_MEASUREMENTS) {
+ tableResultSetEqualTest(
+ String.format(sql, measurement, measurement), expectedHeader,
retArray, DATABASE_NAME);
+ }
+ }
+
+ @Test
+ public void testCorrelatedQuantifiedComparisonLegalityCheck() {
+ // Legality check: Correlated subqueries can only access columns from the
immediately outer
+ // scope and cannot access columns from the further outer queries.
+ tableAssertTestFail(
+ "select s1 from table1 t1 where s1 >= any(select s1 from table3 t3
where t1.s1 = t3.s1 and s1 >= any(select s1 from table2 t2 where t2.s1 =
t1.s1))",
+ "701: Given correlated subquery is not supported",
+ DATABASE_NAME);
+
+ // Legality check: Correlated subqueries with limit clause and limit count
greater than 1 is not
+ // supported for now
+ tableAssertTestFail(
+ "select s1 from table3 t3 where 30 = t3.s1 and s1 >= any(select s1
from table2 t2 where t2.s1 = t3.s1 limit 2)",
+ "Decorrelation for LIMIT with row count greater than 1 is not
supported yet",
+ DATABASE_NAME);
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java
index d61f2993118..bd972417a40 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java
@@ -26,6 +26,7 @@ import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggr
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedApproxCountDistinctAccumulator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAvgAccumulator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedCountAccumulator;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedCountAllAccumulator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedCountIfAccumulator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedExtremeAccumulator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedFirstAccumulator;
@@ -201,6 +202,8 @@ public class AccumulatorFactory {
switch (aggregationType) {
case COUNT:
return new GroupedCountAccumulator();
+ case COUNT_ALL:
+ return new GroupedCountAllAccumulator();
case COUNT_IF:
return new GroupedCountIfAccumulator();
case AVG:
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedCountAllAccumulator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedCountAllAccumulator.java
new file mode 100644
index 00000000000..afa5c1e34e0
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedCountAllAccumulator.java
@@ -0,0 +1,87 @@
+/*
+ * 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.db.queryengine.execution.operator.source.relational.aggregation.grouped;
+
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.AggregationMask;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.LongBigArray;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+public class GroupedCountAllAccumulator implements GroupedAccumulator {
+ private final LongBigArray countValues = new LongBigArray(0L);
+
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(GroupedCountAccumulator.class);
+
+ @Override
+ public long getEstimatedSize() {
+ return INSTANCE_SIZE + countValues.sizeOf();
+ }
+
+ @Override
+ public void setGroupCount(long groupCount) {
+ countValues.ensureCapacity(groupCount);
+ }
+
+ @Override
+ public void addInput(int[] groupIds, Column[] arguments, AggregationMask
mask) {
+ int positionCount = mask.getSelectedPositionCount();
+
+ if (mask.isSelectAll()) {
+ for (int i = 0; i < positionCount; i++) {
+ countValues.increment(groupIds[i]);
+ }
+ } else {
+ int[] selectedPositions = mask.getSelectedPositions();
+ int position;
+ for (int i = 0; i < positionCount; i++) {
+ position = selectedPositions[i];
+ countValues.increment(groupIds[position]);
+ }
+ }
+ }
+
+ @Override
+ public void addIntermediate(int[] groupIds, Column argument) {
+ for (int i = 0; i < groupIds.length; i++) {
+ countValues.add(groupIds[i], argument.getLong(i));
+ }
+ }
+
+ @Override
+ public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
+ columnBuilder.writeLong(countValues.get(groupId));
+ }
+
+ @Override
+ public void evaluateFinal(int groupId, ColumnBuilder columnBuilder) {
+ columnBuilder.writeLong(countValues.get(groupId));
+ }
+
+ @Override
+ public void prepareFinal() {}
+
+ @Override
+ public void reset() {
+ countValues.reset();
+ }
+}
