This is an automated email from the ASF dual-hosted git repository.
JackieTien97 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 56b09f28dec Fix schema cache pollution from LAST queries on virtual
regions (#18694)
56b09f28dec is described below
commit 56b09f28decc17266b446d2dfa8838ca09cb1cb2
Author: Jackie Tien <[email protected]>
AuthorDate: Tue Sep 22 17:16:40 2026 +0800
Fix schema cache pollution from LAST queries on virtual regions (#18694)
---
.../db/it/last/IoTDBLastQueryVirtualRegionIT.java | 137 ++++++++++++
.../plan/planner/OperatorTreeGenerator.java | 12 +-
.../plan/planner/LastQueryCacheTest.java | 236 +++++++++++++++++++++
3 files changed, 383 insertions(+), 2 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/last/IoTDBLastQueryVirtualRegionIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/last/IoTDBLastQueryVirtualRegionIT.java
new file mode 100644
index 00000000000..fceb119ae18
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/last/IoTDBLastQueryVirtualRegionIT.java
@@ -0,0 +1,137 @@
+/*
+ * 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.it.last;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.assertResultSetEqual;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBLastQueryVirtualRegionIT {
+
+ private static final String[] LAST_HEADER = {"Time", "Timeseries", "Value",
"DataType"};
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().getConfig().getCommonConfig().setEnableLastCache(true);
+ EnvFactory.getEnv().initClusterEnvironment();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void testNonAlignedDeviceWithoutDataPartition() throws Exception {
+ testDeviceWithoutDataPartition(false);
+ }
+
+ @Test
+ public void testAlignedDeviceWithoutDataPartition() throws Exception {
+ testDeviceWithoutDataPartition(true);
+ }
+
+ private void testDeviceWithoutDataPartition(boolean aligned) throws
Exception {
+ String prefix = aligned ? "root.last_virtual_aligned" :
"root.last_virtual_normal";
+ String populatedDatabase = prefix + "_data";
+ String emptyDatabase = prefix + "_empty";
+ String populatedDevice = populatedDatabase + ".d";
+ String emptyDevice = emptyDatabase + ".d";
+ // A separate database guarantees that the empty device has no data
partition, even if
+ // its series partition slot happens to collide with that of the populated
device.
+ try (Connection connection =
+ EnvFactory.getEnv()
+
.getConnectionWithSpecifiedDataNode(EnvFactory.getEnv().getDataNodeWrapper(0));
+ Statement statement = connection.createStatement()) {
+ statement.execute("create database " + populatedDatabase);
+ statement.execute("create database " + emptyDatabase);
+ statement.execute("create timeseries " + populatedDevice + ".s1 INT32");
+ statement.execute("insert into " + populatedDevice + "(time,s1)
values(1,11)");
+ if (aligned) {
+ statement.execute("create aligned timeseries " + emptyDevice + "(s1
INT32,s2 INT32)");
+ } else {
+ statement.execute("create timeseries " + emptyDevice + ".s1 INT32");
+ statement.execute("create timeseries " + emptyDevice + ".s2 INT32");
+ }
+
+ String mixedLastQuery =
+ "select last * from "
+ + populatedDatabase
+ + ".**, "
+ + emptyDatabase
+ + ".** order by timeseries";
+ String[] populatedLast = {"1," + populatedDevice + ".s1,11,INT32,"};
+ String[] emptyDeviceHeader = {"Time", emptyDevice + ".s1", emptyDevice +
".s2"};
+ for (int i = 0; i < 2; i++) {
+ statement.execute("clear schema cache on cluster");
+ // Wildcard schema fetching leaves the device schema cache cold. The
mixed LAST query
+ // must still execute the populated branch and the empty
virtual-region branch.
+ assertQuery(statement, mixedLastQuery, LAST_HEADER, populatedLast);
+ assertQuery(
+ statement, "select s1,s2 from " + emptyDevice, emptyDeviceHeader,
new String[0]);
+ assertQuery(statement, "select last s1,s2 from " + emptyDevice,
LAST_HEADER, new String[0]);
+ // Repeat with the schema cache warmed by the exact query.
+ assertQuery(statement, mixedLastQuery, LAST_HEADER, populatedLast);
+ assertQuery(
+ statement, "select s1,s2 from " + emptyDevice, emptyDeviceHeader,
new String[0]);
+ }
+
+ statement.execute(
+ "insert into "
+ + emptyDevice
+ + "(time,s1,s2)"
+ + (aligned ? " aligned" : "")
+ + " values(2,21,22)");
+ assertQuery(
+ statement,
+ "select last s1,s2 from " + emptyDevice + " order by timeseries",
+ LAST_HEADER,
+ new String[] {
+ "2," + emptyDevice + ".s1,21,INT32,", "2," + emptyDevice +
".s2,22,INT32,"
+ });
+ assertQuery(
+ statement,
+ "select s1,s2 from " + emptyDevice,
+ emptyDeviceHeader,
+ new String[] {"2,21,22,"});
+ }
+ }
+
+ private void assertQuery(Statement statement, String sql, String[] header,
String[] rows)
+ throws Exception {
+ try (ResultSet resultSet = statement.executeQuery(sql)) {
+ assertResultSetEqual(resultSet, String.join(",", header) + ",", rows);
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
index dc12fab2a7f..b6a57db5445 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
@@ -243,6 +243,7 @@ import
org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem;
import org.apache.iotdb.db.queryengine.plan.statement.literal.Literal;
import org.apache.iotdb.db.queryengine.transformation.dag.udf.UDTFContext;
+import org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion;
import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSourceType;
import org.apache.iotdb.db.utils.columngenerator.ColumnGenerator;
import org.apache.iotdb.db.utils.columngenerator.ColumnGeneratorType;
@@ -3007,8 +3008,15 @@ public class OperatorTreeGenerator implements
PlanVisitor<Operator, LocalExecuti
@Override
public Operator visitLastQuery(LastQueryNode node, LocalExecutionPlanContext
context) {
Filter globalTimeFilter = context.getGlobalTimeFilter();
-
context.setNeedUpdateLastCache(LastQueryUtil.needUpdateCache(globalTimeFilter));
-
context.setNeedUpdateNullEntry(LastQueryUtil.needUpdateNullEntry(globalTimeFilter));
+ // Virtual regions only provide empty data sources. Their placeholder
database must never be
+ // cached as the owner of a real device, including during last-cache
initialization below.
+ boolean canUpdateLastCache =
+ !(((DataDriverContext) context.getDriverContext()).getDataRegion()
+ instanceof VirtualDataRegion)
+ && LastQueryUtil.needUpdateCache(globalTimeFilter);
+ context.setNeedUpdateLastCache(canUpdateLastCache);
+ context.setNeedUpdateNullEntry(
+ canUpdateLastCache &&
LastQueryUtil.needUpdateNullEntry(globalTimeFilter));
List<Operator> operatorList =
node.getChildren().stream()
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LastQueryCacheTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LastQueryCacheTest.java
new file mode 100644
index 00000000000..865c882c980
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LastQueryCacheTest.java
@@ -0,0 +1,236 @@
+/*
+ * 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.plan.planner;
+
+import org.apache.iotdb.calc.execution.operator.Operator;
+import org.apache.iotdb.commons.audit.UserEntity;
+import org.apache.iotdb.commons.path.MeasurementPath;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.queryengine.common.SessionInfo;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
+import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
+import org.apache.iotdb.db.queryengine.execution.driver.DataDriverContext;
+import org.apache.iotdb.db.queryengine.execution.fragment.DataNodeQueryContext;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.DataSourceOperator;
+import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
+import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
+import
org.apache.iotdb.db.queryengine.plan.analyze.cache.partition.PartitionCache;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.TimePredicate;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryNode;
+import
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache;
+import
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceSchemaCacheManager;
+import org.apache.iotdb.db.storageengine.dataregion.IDataRegionForQuery;
+import org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.filter.basic.Filter;
+import org.apache.tsfile.read.filter.factory.TimeFilterApi;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.lang.reflect.Field;
+import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+
+import static
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
+import static
org.apache.iotdb.db.queryengine.execution.operator.AggregationOperatorTest.TEST_TIME_SLICE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+@RunWith(Parameterized.class)
+public class LastQueryCacheTest {
+
+ private static final String DATABASE = "root.last_query_cache";
+ private final boolean aligned;
+ private final Filter timeFilter;
+ private final boolean updateSchema;
+ private final boolean updateNullEntry;
+ private final TreeDeviceSchemaCacheManager cache =
TreeDeviceSchemaCacheManager.getInstance();
+ private MeasurementPath path;
+ private PartitionCache partitionCache;
+
+ @Parameterized.Parameters(name = "aligned={0}, filter={1}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(
+ new Object[] {false, null, true, true},
+ new Object[] {true, null, true, true},
+ new Object[] {false, TimeFilterApi.gt(10), true, false},
+ new Object[] {true, TimeFilterApi.gt(10), true, false},
+ new Object[] {false, TimeFilterApi.gtEq(10), true, false},
+ new Object[] {true, TimeFilterApi.gtEq(10), true, false},
+ new Object[] {false, TimeFilterApi.lt(10), false, false},
+ new Object[] {true, TimeFilterApi.lt(10), false, false});
+ }
+
+ public LastQueryCacheTest(
+ boolean aligned, Filter timeFilter, boolean updateSchema, boolean
updateNullEntry) {
+ this.aligned = aligned;
+ this.timeFilter = timeFilter;
+ this.updateSchema = updateSchema;
+ this.updateNullEntry = updateNullEntry;
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ cache.cleanUp();
+ path = new MeasurementPath(DATABASE + ".d.s", TSDataType.INT32);
+ path.setUnderAlignedEntity(aligned);
+ // Seed database options locally so cache declarations do not contact a
ConfigNode.
+ Field field =
ClusterPartitionFetcher.class.getDeclaredField("partitionCache");
+ field.setAccessible(true);
+ partitionCache = (PartitionCache)
field.get(ClusterPartitionFetcher.getInstance());
+ partitionCache.updateDatabaseCache(
+ new HashSet<>(Arrays.asList(DATABASE,
VirtualDataRegion.getInstance().getDatabaseName())));
+ }
+
+ @After
+ public void tearDown() {
+ cache.cleanUp();
+ partitionCache.removeFromDatabaseCache();
+ }
+
+ @Test
+ public void testVirtualRegionDoesNotPopulateColdCache() throws Exception {
+ LocalExecutionPlanContext context =
createContext(VirtualDataRegion.getInstance());
+ try (Operator operator = createOperator(context)) {
+ // Initialization alone used to cache the virtual database as the
device's owner.
+ assertTrue(cache.getMatchedNormalSchema(path).isEmpty());
+ consumeEmptyResult(operator, context);
+ assertTrue(cache.getMatchedNormalSchema(path).isEmpty());
+ assertNull(cache.getLastCache(path));
+ }
+ putSchema();
+ assertEquals(DATABASE,
cache.getMatchedNormalSchema(path).getBelongedDatabase(path));
+ }
+
+ @Test
+ public void testVirtualRegionDoesNotUpdateWarmCache() throws Exception {
+ putSchema();
+ cache.declareLastCache(DATABASE, path);
+ LocalExecutionPlanContext context =
createContext(VirtualDataRegion.getInstance());
+ try (Operator operator = createOperator(context)) {
+ consumeEmptyResult(operator, context);
+ assertEquals(DATABASE,
cache.getMatchedNormalSchema(path).getBelongedDatabase(path));
+ // An empty virtual scan must not turn a pending cache entry into a
cached empty result.
+ assertNull(cache.getLastCache(path));
+ }
+ }
+
+ @Test
+ public void testRealRegionRetainsEmptyResultCachePolicy() throws Exception {
+ IDataRegionForQuery dataRegion = mock(IDataRegionForQuery.class);
+ when(dataRegion.getDatabaseName()).thenReturn(DATABASE);
+ LocalExecutionPlanContext context = createContext(dataRegion);
+ try (Operator operator = createOperator(context)) {
+ if (updateSchema) {
+ assertEquals(DATABASE,
cache.getMatchedNormalSchema(path).getBelongedDatabase(path));
+ } else {
+ assertTrue(cache.getMatchedNormalSchema(path).isEmpty());
+ }
+ consumeEmptyResult(operator, context);
+ if (updateNullEntry) {
+ assertEquals(TableDeviceLastCache.PLACEHOLDER_EMPTY_COLUMN,
cache.getLastCache(path));
+ } else {
+ assertNull(cache.getLastCache(path));
+ }
+ }
+ }
+
+ private void putSchema() {
+ ClusterSchemaTree tree = new ClusterSchemaTree();
+ tree.appendSingleMeasurementPath(path);
+ tree.setDatabases(Collections.singleton(DATABASE));
+ cache.put(tree);
+ }
+
+ private LocalExecutionPlanContext createContext(IDataRegionForQuery
dataRegion) {
+ QueryId queryId = new QueryId("last_query_cache_test");
+ FragmentInstanceId instanceId =
+ new FragmentInstanceId(new PlanFragmentId(queryId, 0), "instance");
+ DataNodeQueryContext queryContext = new DataNodeQueryContext(1);
+ HashMap<QueryId, DataNodeQueryContext> queryContexts = new HashMap<>();
+ queryContexts.put(queryId, queryContext);
+ FragmentInstanceContext instanceContext =
+ spy(
+ createFragmentInstanceContext(
+ instanceId,
+ new FragmentInstanceStateMachine(instanceId, Runnable::run),
+ new SessionInfo(
+ 1, new UserEntity(666, "test", "127.0.0.1"),
ZoneId.systemDefault()),
+ dataRegion,
+ (TimePredicate) null,
+ queryContexts,
+ Long.MAX_VALUE,
+ false,
+ false));
+ when(instanceContext.getGlobalTimeFilter()).thenReturn(timeFilter);
+ return new LocalExecutionPlanContext(new TypeProvider(), instanceContext,
queryContext);
+ }
+
+ private Operator createOperator(LocalExecutionPlanContext context) {
+ LastQueryNode node = new LastQueryNode(new PlanNodeId("last"), null,
false);
+ node.addDeviceLastQueryScanNode(
+ new PlanNodeId("scan"),
+ new PartialPath(path.getDevicePath().getNodes()),
+ aligned,
+ Collections.singletonList(path.getMeasurementSchema()),
+ null,
+ false,
+ null);
+ return node.accept(new OperatorTreeGenerator(), context);
+ }
+
+ private void consumeEmptyResult(Operator operator, LocalExecutionPlanContext
context)
+ throws Exception {
+ context
+ .getDriverContext()
+ .getOperatorContexts()
+ .forEach(operatorContext ->
operatorContext.setMaxRunTime(TEST_TIME_SLICE));
+ for (DataSourceOperator source :
+ ((DataDriverContext) context.getDriverContext()).getSourceOperators())
{
+ source.initQueryDataSource(
+ new QueryDataSource(Collections.emptyList(),
Collections.emptyList()));
+ }
+ int iterations = 0;
+ while (operator.hasNext()) {
+ assertTrue("Empty LAST scan did not finish", iterations++ < 10);
+ assertTrue(operator.isBlocked().isDone());
+ TsBlock block = operator.next();
+ assertTrue(block == null || block.isEmpty());
+ }
+ }
+}
