This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch fixFetchDeviceNotInCache
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/fixFetchDeviceNotInCache by
this push:
new 4210205ff42 add it
4210205ff42 is described below
commit 4210205ff42d506616c15b87464ecba45fcd71b3
Author: shuwenwei <[email protected]>
AuthorDate: Mon Jul 14 17:17:02 2025 +0800
add it
---
.../IoTDBTableViewQueryWithCachedDeviceIT.java | 144 +++++++++++++++++++++
.../relational/DeviceIteratorScanOperator.java | 3 +-
.../metadata/fetcher/TableDeviceSchemaFetcher.java | 41 +++---
.../relational/sql/ast/AbstractTraverseDevice.java | 1 +
4 files changed, 172 insertions(+), 17 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/recent/IoTDBTableViewQueryWithCachedDeviceIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/recent/IoTDBTableViewQueryWithCachedDeviceIT.java
new file mode 100644
index 00000000000..32835598cd2
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/recent/IoTDBTableViewQueryWithCachedDeviceIT.java
@@ -0,0 +1,144 @@
+/*
+ * 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.view.recent;
+
+import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.isession.ISession;
+import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.TableSessionBuilder;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareData;
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData;
+
+public class IoTDBTableViewQueryWithCachedDeviceIT {
+
+ protected static final String DATABASE_NAME = "test";
+
+ protected static String[] createTreeNonAlignedDataSqls = {
+ "CREATE TIMESERIES root.db.battery.b1.voltage INT32",
+ "CREATE TIMESERIES root.db.battery.b1.current FLOAT",
+ "INSERT INTO root.db.battery.b1(time, voltage, current) values (1, 1, 1)",
+ "CREATE TIMESERIES root.db.battery.b2.voltage INT32",
+ "CREATE TIMESERIES root.db.battery.b2.current FLOAT",
+ "INSERT INTO root.db.battery.b2(time, voltage, current) values (1, 2, 2)",
+ "CREATE TIMESERIES root.db.battery.b3.voltage INT32",
+ "CREATE TIMESERIES root.db.battery.b3.current FLOAT",
+ "INSERT INTO root.db.battery.b3(time, voltage, current) values (1, 3, 3)",
+ "CREATE TIMESERIES root.db.battery.b4.voltage INT32",
+ "CREATE TIMESERIES root.db.battery.b4.current FLOAT",
+ "INSERT INTO root.db.battery.b4(time, voltage, current) values (1, 4, 4)",
+ "FLUSH",
+ };
+
+ protected static String[] createTableSqls = {
+ "CREATE DATABASE " + DATABASE_NAME,
+ "USE " + DATABASE_NAME,
+ "CREATE VIEW view1 (battery TAG, voltage INT32 FIELD, current FLOAT FIELD)
as root.db.battery.**",
+ };
+
+ public static void main(String[] args)
+ throws IoTDBConnectionException, StatementExecutionException {
+ try (ISession session = new Session.Builder().build()) {
+ session.open();
+ for (String sql : createTreeNonAlignedDataSqls) {
+ session.executeNonQueryStatement(sql);
+ }
+ }
+ try (ITableSession session = new TableSessionBuilder().build()) {
+ for (String createTableSql : createTableSqls) {
+ session.executeNonQueryStatement(createTableSql);
+ }
+ }
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ EnvFactory.getEnv().getConfig().getCommonConfig().setSortBufferSize(128 *
1024);
+
EnvFactory.getEnv().getConfig().getCommonConfig().setMaxTsBlockSizeInByte(4 *
1024);
+ EnvFactory.getEnv().initClusterEnvironment();
+ prepareData(createTreeNonAlignedDataSqls);
+ prepareTableData(createTableSqls);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void test1() throws IoTDBConnectionException,
StatementExecutionException {
+ int count1 = 0;
+ int count2 = 0;
+ try (ITableSession session =
EnvFactory.getEnv().getTableSessionConnection()) {
+ session.executeNonQueryStatement("use " + DATABASE_NAME);
+ SessionDataSet sessionDataSet =
+ session.executeQueryStatement(
+ "select * from view1 where battery = 'b1' or battery = 'b2' or
battery = 'b3'");
+ while (sessionDataSet.hasNext()) {
+ sessionDataSet.next();
+ count1++;
+ }
+ }
+ EnvFactory.getEnv().shutdownAllDataNodes();
+ for (DataNodeWrapper dataNodeWrapper :
EnvFactory.getEnv().getDataNodeWrapperList()) {
+ EnvFactory.getEnv()
+ .ensureNodeStatus(
+ Collections.singletonList(dataNodeWrapper),
+ Collections.singletonList(NodeStatus.Unknown));
+ }
+ EnvFactory.getEnv().startAllDataNodes();
+ for (DataNodeWrapper dataNodeWrapper :
EnvFactory.getEnv().getDataNodeWrapperList()) {
+ EnvFactory.getEnv()
+ .ensureNodeStatus(
+ Collections.singletonList(dataNodeWrapper),
+ Collections.singletonList(NodeStatus.Running));
+ }
+
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ session.executeNonQueryStatement(
+ "INSERT INTO root.db.battery.b3(time, voltage, current) aligned
values (2, 1, 1)");
+ }
+
+ try (ITableSession session =
EnvFactory.getEnv().getTableSessionConnection()) {
+ session.executeNonQueryStatement("use " + DATABASE_NAME);
+ SessionDataSet sessionDataSet =
+ session.executeQueryStatement(
+ "select * from view1 where (battery = 'b1' or battery = 'b2' or
battery = 'b3') and time = 1");
+ while (sessionDataSet.hasNext()) {
+ sessionDataSet.next();
+ count2++;
+ }
+ }
+ Assert.assertEquals(count1, count2);
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/DeviceIteratorScanOperator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/DeviceIteratorScanOperator.java
index 1689fd2fd84..f088e448fd7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/DeviceIteratorScanOperator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/DeviceIteratorScanOperator.java
@@ -34,7 +34,6 @@ import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import java.util.List;
-import java.util.NoSuchElementException;
import java.util.Set;
public class DeviceIteratorScanOperator extends AbstractDataSourceOperator {
@@ -134,7 +133,7 @@ public class DeviceIteratorScanOperator extends
AbstractDataSourceOperator {
@Override
public TsBlock next() throws Exception {
if (!hasNext()) {
- throw new NoSuchElementException();
+ return null;
}
if (!currentDeviceInit) {
return null;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
index a89c94ea61e..d5672db3e31 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
@@ -205,6 +205,7 @@ public class TableDeviceSchemaFetcher {
attributeColumns,
queryContext,
mayContainDuplicateDevice,
+ false,
false)) {
fetchMissingDeviceSchemaForQuery(
database, tableInstance, attributeColumns, statement,
deviceEntryMap, queryContext);
@@ -232,7 +233,8 @@ public class TableDeviceSchemaFetcher {
final List<String> attributeColumns,
final MPPQueryContext queryContext,
final AtomicBoolean mayContainDuplicateDevice,
- final boolean isDirectDeviceQuery) {
+ final boolean isDirectDeviceQuery,
+ final boolean fetchAllIfSomeDeviceNotInCache) {
final Pair<List<Expression>, List<Expression>> separatedExpression =
SchemaPredicateUtil.separateTagDeterminedPredicate(
expressionList, tableInstance, queryContext, isDirectDeviceQuery);
@@ -307,25 +309,34 @@ public class TableDeviceSchemaFetcher {
index2FilterMapList.size()
- tagSingleMatchIndexList.size()
+ tagSingleMatchPredicateNotInCache.size());
- int idx1 = 0;
- int idx2 = 0;
- for (int i = 0; i < index2FilterMapList.size(); i++) {
- if (idx1 >= tagSingleMatchIndexList.size() || i !=
tagSingleMatchIndexList.get(idx1)) {
- tagPredicateForFetch.add(
- index2FilterMapList.get(i).values().stream()
- .flatMap(Collection::stream)
- .collect(Collectors.toList()));
- } else {
- idx1++;
- if (idx2 < tagSingleMatchPredicateNotInCache.size()
- && i == tagSingleMatchPredicateNotInCache.get(idx2)) {
+ if (!fetchAllIfSomeDeviceNotInCache) {
+ int idx1 = 0;
+ int idx2 = 0;
+ for (int i = 0; i < index2FilterMapList.size(); i++) {
+ if (idx1 >= tagSingleMatchIndexList.size() || i !=
tagSingleMatchIndexList.get(idx1)) {
tagPredicateForFetch.add(
index2FilterMapList.get(i).values().stream()
.flatMap(Collection::stream)
.collect(Collectors.toList()));
- idx2++;
+ } else {
+ idx1++;
+ if (idx2 < tagSingleMatchPredicateNotInCache.size()
+ && i == tagSingleMatchPredicateNotInCache.get(idx2)) {
+ tagPredicateForFetch.add(
+ index2FilterMapList.get(i).values().stream()
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList()));
+ idx2++;
+ }
}
}
+ } else {
+ for (Map<Integer, List<SchemaFilter>> integerListMap :
index2FilterMapList) {
+ tagPredicateForFetch.add(
+ integerListMap.values().stream()
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList()));
+ }
}
statement.setTagDeterminedFilterList(tagPredicateForFetch);
statement.setTagFuzzyPredicate(compactedTagFuzzyPredicate);
@@ -562,7 +573,7 @@ public class TableDeviceSchemaFetcher {
deviceEntryList.add(deviceEntry);
// Only cache those exact device query
// Fetch paths is null iff there are fuzzy queries related to id columns
- if (Objects.nonNull(statement.getPartitionKeyList())) {
+ if (false && Objects.nonNull(statement.getPartitionKeyList())) {
cache.putAttributes(statement.getDatabase(), deviceID, attributeMap);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractTraverseDevice.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractTraverseDevice.java
index 2493418b671..7714fee3487 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractTraverseDevice.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AbstractTraverseDevice.java
@@ -149,6 +149,7 @@ public abstract class AbstractTraverseDevice extends
Statement {
attributeColumns,
context,
new AtomicBoolean(false),
+ true,
true);
}
