Skip to content

Commit b5ef04e

Browse files
JackieTien97JackieTien97
authored
[To dev/1.3] Fix schema cache pollution from LAST queries on virtual regions (#18697)
1 parent a42ea12 commit b5ef04e

3 files changed

Lines changed: 367 additions & 2 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/db/it/last/IoTDBLastQueryVirtualRegionIT.java‎

Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.iotdb.db.it.last;
20+
21+
import org.apache.iotdb.it.env.EnvFactory;
22+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
23+
import org.apache.iotdb.itbase.category.ClusterIT;
24+
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
25+
26+
import org.junit.AfterClass;
27+
import org.junit.BeforeClass;
28+
import org.junit.Test;
29+
import org.junit.experimental.categories.Category;
30+
import org.junit.runner.RunWith;
31+
32+
import java.sql.Connection;
33+
import java.sql.ResultSet;
34+
import java.sql.Statement;
35+
36+
import static org.apache.iotdb.db.it.utils.TestUtils.assertResultSetEqual;
37+
38+
@RunWith(IoTDBTestRunner.class)
39+
@Category({LocalStandaloneIT.class, ClusterIT.class})
40+
public class IoTDBLastQueryVirtualRegionIT {
41+
42+
private static final String[] LAST_HEADER = {"Time", "Timeseries", "Value", "DataType"};
43+
44+
@BeforeClass
45+
public static void setUp() throws Exception {
46+
EnvFactory.getEnv().getConfig().getCommonConfig().setEnableLastCache(true);
47+
EnvFactory.getEnv().initClusterEnvironment();
48+
}
49+
50+
@AfterClass
51+
public static void tearDown() throws Exception {
52+
EnvFactory.getEnv().cleanClusterEnvironment();
53+
}
54+
55+
@Test
56+
public void testNonAlignedDeviceWithoutDataPartition() throws Exception {
57+
testDeviceWithoutDataPartition(false);
58+
}
59+
60+
@Test
61+
public void testAlignedDeviceWithoutDataPartition() throws Exception {
62+
testDeviceWithoutDataPartition(true);
63+
}
64+
65+
private void testDeviceWithoutDataPartition(boolean aligned) throws Exception {
66+
String prefix = aligned ? "root.last_virtual_aligned" : "root.last_virtual_normal";
67+
String populatedDatabase = prefix + "_data";
68+
String emptyDatabase = prefix + "_empty";
69+
String populatedDevice = populatedDatabase + ".d";
70+
String emptyDevice = emptyDatabase + ".d";
71+
// A separate database guarantees that the empty device has no data partition, even if
72+
// its series partition slot happens to collide with that of the populated device.
73+
try (Connection connection =
74+
EnvFactory.getEnv()
75+
.getConnectionWithSpecifiedDataNode(EnvFactory.getEnv().getDataNodeWrapper(0));
76+
Statement statement = connection.createStatement()) {
77+
statement.execute("create database " + populatedDatabase);
78+
statement.execute("create database " + emptyDatabase);
79+
statement.execute("create timeseries " + populatedDevice + ".s1 INT32");
80+
statement.execute("insert into " + populatedDevice + "(time,s1) values(1,11)");
81+
if (aligned) {
82+
statement.execute("create aligned timeseries " + emptyDevice + "(s1 INT32,s2 INT32)");
83+
} else {
84+
statement.execute("create timeseries " + emptyDevice + ".s1 INT32");
85+
statement.execute("create timeseries " + emptyDevice + ".s2 INT32");
86+
}
87+
88+
String mixedLastQuery =
89+
"select last * from "
90+
+ populatedDatabase
91+
+ ".**, "
92+
+ emptyDatabase
93+
+ ".** order by timeseries";
94+
String[] populatedLast = {"1," + populatedDevice + ".s1,11,INT32,"};
95+
String[] emptyDeviceHeader = {"Time", emptyDevice + ".s1", emptyDevice + ".s2"};
96+
for (int i = 0; i < 2; i++) {
97+
statement.execute("clear schema cache on cluster");
98+
// Wildcard schema fetching leaves the device schema cache cold. The mixed LAST query
99+
// must still execute the populated branch and the empty virtual-region branch.
100+
assertQuery(statement, mixedLastQuery, LAST_HEADER, populatedLast);
101+
assertQuery(
102+
statement, "select s1,s2 from " + emptyDevice, emptyDeviceHeader, new String[0]);
103+
assertQuery(statement, "select last s1,s2 from " + emptyDevice, LAST_HEADER, new String[0]);
104+
// Repeat with the schema cache warmed by the exact query.
105+
assertQuery(statement, mixedLastQuery, LAST_HEADER, populatedLast);
106+
assertQuery(
107+
statement, "select s1,s2 from " + emptyDevice, emptyDeviceHeader, new String[0]);
108+
}
109+
110+
statement.execute(
111+
"insert into "
112+
+ emptyDevice
113+
+ "(time,s1,s2)"
114+
+ (aligned ? " aligned" : "")
115+
+ " values(2,21,22)");
116+
assertQuery(
117+
statement,
118+
"select last s1,s2 from " + emptyDevice + " order by timeseries",
119+
LAST_HEADER,
120+
new String[] {
121+
"2," + emptyDevice + ".s1,21,INT32,", "2," + emptyDevice + ".s2,22,INT32,"
122+
});
123+
assertQuery(
124+
statement,
125+
"select s1,s2 from " + emptyDevice,
126+
emptyDeviceHeader,
127+
new String[] {"2,21,22,"});
128+
}
129+
}
130+
131+
private void assertQuery(Statement statement, String sql, String[] header, String[] rows)
132+
throws Exception {
133+
try (ResultSet resultSet = statement.executeQuery(sql)) {
134+
assertResultSetEqual(resultSet, String.join(",", header) + ",", rows);
135+
}
136+
}
137+
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -255,6 +255,7 @@
255255
import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransformer;
256256
import org.apache.iotdb.db.queryengine.transformation.dag.column.leaf.LeafColumnTransformer;
257257
import org.apache.iotdb.db.queryengine.transformation.dag.udf.UDTFContext;
258+
import org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion;
258259
import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSourceType;
259260
import org.apache.iotdb.db.utils.columngenerator.ColumnGenerator;
260261
import org.apache.iotdb.db.utils.columngenerator.ColumnGeneratorType;
@@ -3047,8 +3048,15 @@ public Operator visitLastQueryScan(LastQueryScanNode node, LocalExecutionPlanCon
30473048
@Override
30483049
public Operator visitLastQuery(LastQueryNode node, LocalExecutionPlanContext context) {
30493050
Filter globalTimeFilter = context.getGlobalTimeFilter();
3050-
context.setNeedUpdateLastCache(LastQueryUtil.needUpdateCache(globalTimeFilter));
3051-
context.setNeedUpdateNullEntry(LastQueryUtil.needUpdateNullEntry(globalTimeFilter));
3051+
// Virtual regions only provide empty data sources. Their placeholder database must never be
3052+
// cached as the owner of a real device, including during last-cache initialization below.
3053+
boolean canUpdateLastCache =
3054+
!(((DataDriverContext) context.getDriverContext()).getDataRegion()
3055+
instanceof VirtualDataRegion)
3056+
&& LastQueryUtil.needUpdateCache(globalTimeFilter);
3057+
context.setNeedUpdateLastCache(canUpdateLastCache);
3058+
context.setNeedUpdateNullEntry(
3059+
canUpdateLastCache && LastQueryUtil.needUpdateNullEntry(globalTimeFilter));
30523060

30533061
List<Operator> operatorList =
30543062
node.getChildren().stream()

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LastQueryCacheTest.java‎

Lines changed: 220 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,220 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.iotdb.db.queryengine.plan.planner;
20+
21+
import org.apache.iotdb.commons.path.MeasurementPath;
22+
import org.apache.iotdb.commons.path.PartialPath;
23+
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
24+
import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
25+
import org.apache.iotdb.db.queryengine.common.QueryId;
26+
import org.apache.iotdb.db.queryengine.common.SessionInfo;
27+
import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
28+
import org.apache.iotdb.db.queryengine.execution.driver.DataDriverContext;
29+
import org.apache.iotdb.db.queryengine.execution.fragment.DataNodeQueryContext;
30+
import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
31+
import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine;
32+
import org.apache.iotdb.db.queryengine.execution.operator.Operator;
33+
import org.apache.iotdb.db.queryengine.execution.operator.source.DataSourceOperator;
34+
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
35+
import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
36+
import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DeviceLastCache;
37+
import org.apache.iotdb.db.queryengine.plan.planner.plan.TimePredicate;
38+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
39+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.last.LastQueryNode;
40+
import org.apache.iotdb.db.storageengine.dataregion.IDataRegionForQuery;
41+
import org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion;
42+
import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
43+
44+
import org.apache.tsfile.enums.TSDataType;
45+
import org.apache.tsfile.read.common.block.TsBlock;
46+
import org.apache.tsfile.read.filter.basic.Filter;
47+
import org.apache.tsfile.read.filter.factory.TimeFilterApi;
48+
import org.junit.After;
49+
import org.junit.Before;
50+
import org.junit.Test;
51+
import org.junit.runner.RunWith;
52+
import org.junit.runners.Parameterized;
53+
54+
import java.time.ZoneId;
55+
import java.util.Arrays;
56+
import java.util.Collection;
57+
import java.util.Collections;
58+
import java.util.HashMap;
59+
60+
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
61+
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationOperatorTest.TEST_TIME_SLICE;
62+
import static org.junit.Assert.assertEquals;
63+
import static org.junit.Assert.assertNull;
64+
import static org.junit.Assert.assertSame;
65+
import static org.junit.Assert.assertTrue;
66+
import static org.mockito.Mockito.mock;
67+
import static org.mockito.Mockito.spy;
68+
import static org.mockito.Mockito.when;
69+
70+
@RunWith(Parameterized.class)
71+
public class LastQueryCacheTest {
72+
73+
private static final String DATABASE = "root.last_query_cache";
74+
private final boolean aligned;
75+
private final Filter timeFilter;
76+
private final boolean updateSchema;
77+
private final boolean updateNullEntry;
78+
private final DataNodeSchemaCache cache = DataNodeSchemaCache.getInstance();
79+
private MeasurementPath path;
80+
81+
@Parameterized.Parameters(name = "aligned={0}, filter={1}")
82+
public static Collection<Object[]> parameters() {
83+
return Arrays.asList(
84+
new Object[] {false, null, true, true},
85+
new Object[] {true, null, true, true},
86+
new Object[] {false, TimeFilterApi.gt(10), true, false},
87+
new Object[] {true, TimeFilterApi.gt(10), true, false},
88+
new Object[] {false, TimeFilterApi.gtEq(10), true, false},
89+
new Object[] {true, TimeFilterApi.gtEq(10), true, false},
90+
new Object[] {false, TimeFilterApi.lt(10), false, false},
91+
new Object[] {true, TimeFilterApi.lt(10), false, false});
92+
}
93+
94+
public LastQueryCacheTest(
95+
boolean aligned, Filter timeFilter, boolean updateSchema, boolean updateNullEntry) {
96+
this.aligned = aligned;
97+
this.timeFilter = timeFilter;
98+
this.updateSchema = updateSchema;
99+
this.updateNullEntry = updateNullEntry;
100+
}
101+
102+
@Before
103+
public void setUp() throws Exception {
104+
cache.cleanUp();
105+
path = new MeasurementPath(DATABASE + ".d.s", TSDataType.INT32);
106+
path.setUnderAlignedEntity(aligned);
107+
}
108+
109+
@After
110+
public void tearDown() {
111+
cache.cleanUp();
112+
}
113+
114+
@Test
115+
public void testVirtualRegionDoesNotPopulateColdCache() throws Exception {
116+
LocalExecutionPlanContext context = createContext(VirtualDataRegion.getInstance());
117+
try (Operator operator = createOperator(context)) {
118+
// Initialization alone used to cache the virtual database as the device's owner.
119+
assertTrue(cache.getMatchedNormalSchema(path).isEmpty());
120+
consumeEmptyResult(operator, context);
121+
assertTrue(cache.getMatchedNormalSchema(path).isEmpty());
122+
assertNull(cache.getLastCache(path));
123+
}
124+
putSchema();
125+
assertEquals(DATABASE, cache.getMatchedNormalSchema(path).getBelongedDatabase(path));
126+
}
127+
128+
@Test
129+
public void testVirtualRegionDoesNotUpdateWarmCache() throws Exception {
130+
putSchema();
131+
cache.declareLastCache(DATABASE, path);
132+
LocalExecutionPlanContext context = createContext(VirtualDataRegion.getInstance());
133+
try (Operator operator = createOperator(context)) {
134+
consumeEmptyResult(operator, context);
135+
assertEquals(DATABASE, cache.getMatchedNormalSchema(path).getBelongedDatabase(path));
136+
// An empty virtual scan must not turn a pending cache entry into a cached empty result.
137+
assertNull(cache.getLastCache(path));
138+
}
139+
}
140+
141+
@Test
142+
public void testRealRegionRetainsEmptyResultCachePolicy() throws Exception {
143+
IDataRegionForQuery dataRegion = mock(IDataRegionForQuery.class);
144+
when(dataRegion.getDatabaseName()).thenReturn(DATABASE);
145+
LocalExecutionPlanContext context = createContext(dataRegion);
146+
try (Operator operator = createOperator(context)) {
147+
if (updateSchema) {
148+
assertEquals(DATABASE, cache.getMatchedNormalSchema(path).getBelongedDatabase(path));
149+
} else {
150+
assertTrue(cache.getMatchedNormalSchema(path).isEmpty());
151+
}
152+
consumeEmptyResult(operator, context);
153+
if (updateNullEntry) {
154+
assertSame(DeviceLastCache.EMPTY_TIME_VALUE_PAIR, cache.getLastCache(path));
155+
} else {
156+
assertNull(cache.getLastCache(path));
157+
}
158+
}
159+
}
160+
161+
private void putSchema() {
162+
ClusterSchemaTree tree = new ClusterSchemaTree();
163+
tree.appendSingleMeasurementPath(path);
164+
tree.setDatabases(Collections.singleton(DATABASE));
165+
cache.put(tree);
166+
}
167+
168+
private LocalExecutionPlanContext createContext(IDataRegionForQuery dataRegion) {
169+
QueryId queryId = new QueryId("last_query_cache_test");
170+
FragmentInstanceId instanceId =
171+
new FragmentInstanceId(new PlanFragmentId(queryId, 0), "instance");
172+
DataNodeQueryContext queryContext = new DataNodeQueryContext(1);
173+
HashMap<QueryId, DataNodeQueryContext> queryContexts = new HashMap<>();
174+
queryContexts.put(queryId, queryContext);
175+
FragmentInstanceContext instanceContext =
176+
spy(
177+
createFragmentInstanceContext(
178+
instanceId,
179+
new FragmentInstanceStateMachine(instanceId, Runnable::run),
180+
new SessionInfo(1, "test", ZoneId.systemDefault(), "127.0.0.1"),
181+
dataRegion,
182+
(TimePredicate) null,
183+
queryContexts,
184+
false));
185+
when(instanceContext.getGlobalTimeFilter()).thenReturn(timeFilter);
186+
return new LocalExecutionPlanContext(new TypeProvider(), instanceContext, queryContext);
187+
}
188+
189+
private Operator createOperator(LocalExecutionPlanContext context) {
190+
LastQueryNode node = new LastQueryNode(new PlanNodeId("last"), null, false);
191+
node.addDeviceLastQueryScanNode(
192+
new PlanNodeId("scan"),
193+
new PartialPath(path.getDevicePath().getNodes()),
194+
aligned,
195+
Collections.singletonList(path.getMeasurementSchema()),
196+
null,
197+
false);
198+
return node.accept(new OperatorTreeGenerator(), context);
199+
}
200+
201+
private void consumeEmptyResult(Operator operator, LocalExecutionPlanContext context)
202+
throws Exception {
203+
context
204+
.getDriverContext()
205+
.getOperatorContexts()
206+
.forEach(operatorContext -> operatorContext.setMaxRunTime(TEST_TIME_SLICE));
207+
for (DataSourceOperator source :
208+
((DataDriverContext) context.getDriverContext()).getSourceOperators()) {
209+
source.initQueryDataSource(
210+
new QueryDataSource(Collections.emptyList(), Collections.emptyList()));
211+
}
212+
int iterations = 0;
213+
while (operator.hasNext()) {
214+
assertTrue("Empty LAST scan did not finish", iterations++ < 10);
215+
assertTrue(operator.isBlocked().isDone());
216+
TsBlock block = operator.next();
217+
assertTrue(block == null || block.isEmpty());
218+
}
219+
}
220+
}

0 commit comments

Comments
 (0)