This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 76d573a4e3d Fix deletion recovery and table last-cache correctness
(#19081)
76d573a4e3d is described below
commit 76d573a4e3d83068d4eb4bbb1e4ed97040463eb8
Author: Caideyipi <[email protected]>
AuthorDate: Mon Nov 2 18:19:38 2026 +0800
Fix deletion recovery and table last-cache correctness (#19081)
---
.../org/apache/iotdb/db/it/IoTDBRestartIT.java | 124 +++++-
.../iotdb/db/it/utils/DeletionTestUtils.java | 119 ++++++
.../relational/it/db/it/IoTDBDeletionTableIT.java | 467 ++++++++++++++++++++-
.../relational/it/db/it/IoTDBRestartTableIT.java | 153 +++++--
.../execution/fragment/DataNodeQueryContext.java | 2 +-
.../relational/LastQueryAggTableScanOperator.java | 78 ++--
.../planner/DataNodeTableOperatorGenerator.java | 29 +-
.../fetcher/cache/TableDeviceCacheEntry.java | 8 +
.../fetcher/cache/TableDeviceLastCache.java | 20 +-
.../fetcher/cache/TableDeviceSchemaCache.java | 14 +
.../wal/recover/file/TsFilePlanRedoer.java | 60 ++-
.../fetcher/cache/TableDeviceLastCacheTest.java | 33 ++
.../wal/recover/file/TsFilePlanRedoerTest.java | 124 ++++++
13 files changed, 1135 insertions(+), 96 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestartIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestartIT.java
index 571826307c6..302d059959a 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestartIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestartIT.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.it;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.it.utils.DeletionTestUtils;
import org.apache.iotdb.db.it.utils.TestUtils;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -30,6 +31,7 @@ import org.apache.iotdb.itbase.category.LocalStandaloneIT;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.external.commons.io.FileUtils;
+import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
import org.junit.Before;
@@ -41,16 +43,21 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.nio.file.Files;
+import java.nio.file.Path;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Collections;
+import java.util.List;
+import java.util.Map;
import static org.apache.iotdb.db.utils.constant.TestConstant.TIMESTAMP_STR;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@RunWith(IoTDBTestRunner.class)
@@ -61,7 +68,18 @@ public class IoTDBRestartIT {
@Before
public void setUp() throws Exception {
- EnvFactory.getEnv().getConfig().getCommonConfig().setWalMode("SYNC");
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setWalMode("SYNC")
+
.setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.simple.SimpleConsensus")
+ .setDataReplicationFactor(1)
+ .setDataRegionGroupExtensionPolicy("CUSTOM")
+ .setDefaultDataRegionGroupNumPerDatabase(1)
+ .setMemtableSizeThreshold(16 * 1024 * 1024)
+ .setEnableSeqSpaceCompaction(false)
+ .setEnableUnseqSpaceCompaction(false)
+ .setEnableCrossSpaceCompaction(false);
EnvFactory.getEnv()
.getConfig()
.getCommonConfig()
@@ -165,6 +183,110 @@ public class IoTDBRestartIT {
}
}
+ @Test
+ public void testDeleteAndRewriteSurviveForcedRestart() throws Exception {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE root.restart_delete");
+ statement.execute("CREATE TIMESERIES root.restart_delete.d1.s1 WITH
DATATYPE=INT64");
+ statement.execute("CREATE TIMESERIES root.restart_delete.d1.s2 WITH
DATATYPE=INT64");
+ statement.execute("CREATE TIMESERIES root.restart_delete.d2.s1 WITH
DATATYPE=INT64");
+ statement.execute("CREATE TIMESERIES root.restart_delete.d2.s2 WITH
DATATYPE=INT64");
+
+ // The first two rows are in a TsFile. The remaining rows are only
protected by the SYNC
+ // WAL when the delete is issued.
+ statement.execute("INSERT INTO root.restart_delete.d1(time,s1,s2)
VALUES(1,11,101)");
+ statement.execute("INSERT INTO root.restart_delete.d1(time,s1)
VALUES(2,12)");
+ statement.execute("INSERT INTO root.restart_delete.d2(time,s1,s2)
VALUES(2,22,202)");
+ statement.execute("FLUSH");
+ Path history =
+ DeletionTestUtils.assertSharedTsFile(
+ "root.restart_delete",
+
IDeviceID.Factory.DEFAULT_FACTORY.create("root.restart_delete.d1"),
+
IDeviceID.Factory.DEFAULT_FACTORY.create("root.restart_delete.d2"));
+ assertEquals(
+ Map.of(1L, 11L, 2L, 12L),
+ DeletionTestUtils.readTreePoints(history, "root.restart_delete.d1",
"s1"));
+ List<Path> closedFiles =
DeletionTestUtils.getClosedTsFiles("root.restart_delete", true);
+
+ statement.execute("INSERT INTO root.restart_delete.d1(time,s1)
VALUES(1,111)");
+ statement.execute("INSERT INTO root.restart_delete.d1(time,s1)
VALUES(3,13)");
+ statement.execute("INSERT INTO root.restart_delete.d1(time,s1,s2)
VALUES(4,14,104)");
+ assertEquals(closedFiles,
DeletionTestUtils.getClosedTsFiles("root.restart_delete", true));
+ assertTrue(DeletionTestUtils.getClosedTsFiles("root.restart_delete",
false).isEmpty());
+ assertEquals(
+ Map.of(1L, 11L, 2L, 12L),
+ DeletionTestUtils.readTreePoints(history, "root.restart_delete.d1",
"s1"));
+
+ statement.execute("DELETE FROM root.restart_delete.d1.s1 WHERE time <=
3");
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d1
ORDER BY time")) {
+ assertTrue(resultSet.next());
+ assertEquals(1, resultSet.getLong(1));
+ assertNull(resultSet.getObject(2));
+ assertEquals(101, resultSet.getLong(3));
+ assertTrue(resultSet.next());
+ assertEquals(4, resultSet.getLong(1));
+ assertEquals(14, resultSet.getLong(2));
+ assertEquals(104, resultSet.getLong(3));
+ assertFalse(resultSet.next());
+ }
+
+ // This rewrite is deliberately older than the surviving point. It must
survive the old
+ // delete record and remain visible after recovery, while s2 stays NULL.
+ statement.execute("INSERT INTO root.restart_delete.d1(time,s1)
VALUES(2,1202)");
+ assertEquals(closedFiles,
DeletionTestUtils.getClosedTsFiles("root.restart_delete", true));
+ assertTrue(DeletionTestUtils.getClosedTsFiles("root.restart_delete",
false).isEmpty());
+ try (ResultSet rows =
+ statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d1
ORDER BY time")) {
+ TestUtils.assertResultSetEqual(
+ rows,
+ "Time,root.restart_delete.d1.s1,root.restart_delete.d1.s2,",
+ new String[] {"1,null,101,", "2,1202,null,", "4,14,104,"});
+ }
+ }
+
+ TestUtils.stopForciblyAndRestartDataNodes();
+
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d1
ORDER BY time")) {
+ assertTrue(resultSet.next());
+ assertEquals(1, resultSet.getLong(1));
+ assertNull(resultSet.getObject(2));
+ assertEquals(101, resultSet.getLong(3));
+ assertTrue(resultSet.next());
+ assertEquals(2, resultSet.getLong(1));
+ assertEquals(1202, resultSet.getLong(2));
+ assertNull(resultSet.getObject(3));
+ assertTrue(resultSet.next());
+ assertEquals(4, resultSet.getLong(1));
+ assertEquals(14, resultSet.getLong(2));
+ assertEquals(104, resultSet.getLong(3));
+ assertFalse(resultSet.next());
+ }
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d2
ORDER BY time")) {
+ assertTrue(resultSet.next());
+ assertEquals(2, resultSet.getLong(1));
+ assertEquals(22, resultSet.getLong(2));
+ assertEquals(202, resultSet.getLong(3));
+ assertFalse(resultSet.next());
+ }
+ // Only after checking both the target and control do we test continued
writes and flush.
+ statement.execute("INSERT INTO root.restart_delete.d1(time,s1,s2)
VALUES(5,15,105)");
+ statement.execute("FLUSH");
+ try (ResultSet rows =
+ statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d1
ORDER BY time")) {
+ TestUtils.assertResultSetEqual(
+ rows,
+ "Time,root.restart_delete.d1.s1,root.restart_delete.d1.s2,",
+ new String[] {"1,null,101,", "2,1202,null,", "4,14,104,",
"5,15,105,"});
+ }
+ }
+ }
+
@Test
public void testRestartQueryLargerThanEndTime() throws SQLException {
try (Connection connection = EnvFactory.getEnv().getConnection();
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/DeletionTestUtils.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/DeletionTestUtils.java
new file mode 100644
index 00000000000..f8e94848d74
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/DeletionTestUtils.java
@@ -0,0 +1,119 @@
+/*
+ * 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.utils;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+
+import org.apache.tsfile.file.metadata.ChunkMetadata;
+import org.apache.tsfile.file.metadata.IChunkMetadata;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.TimeseriesMetadata;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.common.BatchData;
+import org.apache.tsfile.read.reader.chunk.ChunkReader;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
+import static org.junit.Assert.fail;
+
+/** Physical preconditions for deletion ITs; SQL assertions still verify the
visible data. */
+public final class DeletionTestUtils {
+ private DeletionTestUtils() {}
+
+ public static List<Path> getClosedTsFiles(String database, boolean sequence)
throws IOException {
+ List<Path> result = new ArrayList<>();
+ for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) {
+ Path directory =
+ Paths.get(
+ node.getDataPath(),
+ sequence ? IoTDBConstant.SEQUENCE_FOLDER_NAME :
IoTDBConstant.UNSEQUENCE_FOLDER_NAME,
+ database);
+ if (Files.exists(directory)) {
+ try (Stream<Path> files = Files.walk(directory)) {
+ result.addAll(
+ files
+ .filter(Files::isRegularFile)
+ .filter(file -> file.toString().endsWith(TSFILE_SUFFIX))
+ .filter(file -> Files.isRegularFile(Paths.get(file +
".resource")))
+ .sorted()
+ .collect(Collectors.toList()));
+ }
+ }
+ }
+ return result;
+ }
+
+ public static Path assertSharedTsFile(String database, IDeviceID... devices)
throws IOException {
+ for (Path file : getClosedTsFiles(database, true)) {
+ try (TsFileSequenceReader reader = new
TsFileSequenceReader(file.toString())) {
+ if (reader.getAllDevices().containsAll(Arrays.asList(devices))) {
+ return file;
+ }
+ }
+ }
+ fail("No closed TsFile contains all expected devices: " +
Arrays.toString(devices));
+ return null;
+ }
+
+ // Read the physical file without applying mods or merging it with newer
files.
+ public static Map<Long, Long> readTreePoints(Path file, String device,
String measurement)
+ throws IOException {
+ Map<Long, Long> points = new TreeMap<>();
+ try (TsFileSequenceReader reader = new
TsFileSequenceReader(file.toString())) {
+ List<TimeseriesMetadata> series =
+ reader
+ .getAllTimeseriesMetadata(true)
+ .get(IDeviceID.Factory.DEFAULT_FACTORY.create(device));
+ if (series != null) {
+ for (TimeseriesMetadata metadata : series) {
+ if (!measurement.equals(metadata.getMeasurementId())) {
+ continue;
+ }
+ for (IChunkMetadata chunk : metadata.getChunkMetadataList()) {
+ try (ChunkReader chunkReader =
+ new ChunkReader(reader.readMemChunk((ChunkMetadata) chunk))) {
+ while (chunkReader.hasNextSatisfiedPage()) {
+ BatchData page = chunkReader.nextPageData();
+ while (page.hasCurrent()) {
+ points.put(page.currentTime(), page.getLong());
+ page.next();
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ return points;
+ }
+}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
index 1ff1beebec2..0f29879caad 100644
---
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.relational.it.db.it;
import org.apache.iotdb.db.it.utils.TestUtils;
import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionConfig;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
@@ -119,6 +120,9 @@ public class IoTDBDeletionTableIT {
EnvFactory.getEnv()
.getConfig()
.getCommonConfig()
+
.setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.simple.SimpleConsensus")
+ .setDataReplicationFactor(1)
+ .setEnableLastCache(true)
.setPartitionInterval(1000)
.setMemtableSizeThreshold(10000);
// Adjust MemTable threshold size to make it flush automatically
@@ -1086,24 +1090,465 @@ public class IoTDBDeletionTableIT {
@Test
public void testSuccessfullyInvalidateCache() throws SQLException {
- prepareData(4, 1);
try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
Statement statement = connection.createStatement()) {
statement.execute("use test");
- statement.executeQuery(
- "SELECT last(time), last_by(s0,time), last_by(s1,time),
last_by(s2,time), last_by(s3,time), last_by(s4,time) FROM vehicle4 where
deviceId = 'd0'");
+ statement.execute(
+ "CREATE TABLE last_delete_cache(device_id STRING TAG, s1 INT32
FIELD, s2 INT32 FIELD)");
+ statement.execute(
+ "INSERT INTO last_delete_cache(time,device_id,s1,s2)
VALUES(1,'d0',11,101)");
+ statement.execute("INSERT INTO last_delete_cache(time,device_id,s1)
VALUES(2,'d0',12)");
+ statement.execute("INSERT INTO last_delete_cache(time,device_id,s2)
VALUES(3,'d0',103)");
+ statement.execute("INSERT INTO last_delete_cache(time,device_id,s1)
VALUES(4,'d0',14)");
+ statement.execute("FLUSH");
- // [1, 400] -> [1, 299]
- statement.execute("DELETE FROM vehicle4 WHERE time >= 300");
- try (ResultSet set = statement.executeQuery("SELECT s0 FROM vehicle4")) {
- int cnt = 0;
- while (set.next()) {
- cnt++;
+ assertLastValuesOnAllCoordinators(4L, 14, 103, 14, null, true);
+ assertLastCacheRows(
+ "1970-01-01T00:00:00.001Z,d0,11,101,",
+ "1970-01-01T00:00:00.002Z,d0,12,null,",
+ "1970-01-01T00:00:00.003Z,d0,null,103,",
+ "1970-01-01T00:00:00.004Z,d0,14,null,");
+
+ statement.execute("DELETE FROM last_delete_cache WHERE device_id='d0'
AND time >= 4");
+ // The sparse fields must fall back independently after their common
latest row is deleted.
+ // last ignores NULLs; last_by returns the value (including NULL) at the
latest row.
+ assertLastValuesOnAllCoordinators(3L, 12, 103, null, 103);
+ assertLastCacheRows(
+ "1970-01-01T00:00:00.001Z,d0,11,101,",
+ "1970-01-01T00:00:00.002Z,d0,12,null,",
+ "1970-01-01T00:00:00.003Z,d0,null,103,");
+
+ statement.execute("DELETE FROM last_delete_cache WHERE device_id='d0'
AND time = 1");
+ assertLastValuesOnAllCoordinators(3L, 12, 103, null, 103);
+ assertLastCacheRows(
+ "1970-01-01T00:00:00.002Z,d0,12,null,",
"1970-01-01T00:00:00.003Z,d0,null,103,");
+
+ statement.execute("DELETE FROM last_delete_cache WHERE device_id='d0'");
+ assertLastValuesOnAllCoordinators(null, null, null, null, null);
+ assertLastCacheRows();
+
+ // Rewriting an older timestamp after the full delete must become the
new last value; the
+ // deleted timestamp 4 must not reappear from the cache or the flushed
file.
+ statement.execute("INSERT INTO last_delete_cache(time,device_id,s1)
VALUES(2,'d0',120)");
+ assertLastCacheRows("1970-01-01T00:00:00.002Z,d0,120,null,");
+ assertLastValuesOnAllCoordinators(2L, 120, null, 120, null);
+ statement.execute("FLUSH");
+ assertLastValuesOnAllCoordinators(2L, 120, null, 120, null);
+ assertLastCacheRows("1970-01-01T00:00:00.002Z,d0,120,null,");
+ }
+ }
+
+ @Test
+ public void testRewrittenLastCacheWithTimeFilter() throws SQLException {
+ String query =
+ "SELECT
last(time),last(s1),last(s2),last(device_id),last(label),last_by(time,s1) "
+ + "FROM last_filter_cache WHERE device_id='d0'";
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute(
+ "CREATE TABLE last_filter_cache(device_id STRING TAG, label STRING
ATTRIBUTE, s1 INT32 FIELD, s2 INT32 FIELD)");
+ statement.execute(
+ "INSERT INTO last_filter_cache(time,device_id,label,s1)
VALUES(4,'d0','blue',14)");
+ statement.execute("DELETE FROM last_filter_cache WHERE device_id='d0'");
+ }
+ for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) {
+ try (Connection connection =
+ EnvFactory.getEnv()
+ .getConnection(
+ node,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ assertFilteredLastValues(statement, query, true);
+ }
+ }
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute("INSERT INTO last_filter_cache(time,device_id,s1)
VALUES(2,'d0',120)");
+ }
+ for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) {
+ try (Connection connection =
+ EnvFactory.getEnv()
+ .getConnection(
+ node,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ assertFilteredLastValues(statement, query, false);
+ assertFilteredLastValues(statement, query + " AND time > 2", true);
+ assertFilteredLastValues(statement, query + " AND time >= 3", true);
+ try (ResultSet rows = statement.executeQuery(query + " AND time > 2
GROUP BY device_id")) {
+ assertFalse(rows.next());
+ }
+ assertFilteredLastValues(statement, query + " AND time >= 2", false);
+ assertFilteredLastValues(statement, query + " AND time < 2", true);
+ assertFilteredLastValues(statement, query, false);
+ assertFilteredLastValues(statement, query + " AND time >= 2 GROUP BY
device_id", false);
+ }
+ }
+ }
+
+ @Test
+ public void testSparseLastCacheWithoutTime() throws SQLException {
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute(
+ "CREATE TABLE last_sparse_cache(device_id STRING TAG, s1 INT32
FIELD, s2 INT32 FIELD)");
+ statement.execute(
+ "INSERT INTO last_sparse_cache(time,device_id,s1)
VALUES(3,'d0',30),(1,'d1',10)");
+ }
+
+ // Multiple devices require partial aggregation. An empty s2 cache must
not discard s1.
+ assertCachedLastQueryOnAllCoordinators(
+ "SELECT last(s1) AS v1,last(s2) AS v2 FROM last_sparse_cache",
"v1,v2,", "30,null,");
+ }
+
+ @Test
+ public void testNullTagAndAttributeDoNotCacheDeviceAsEmpty() throws
SQLException {
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute(
+ "CREATE TABLE last_null_identity_cache(device_id STRING TAG, site
STRING TAG, "
+ + "label STRING ATTRIBUTE, s1 INT32 FIELD)");
+ statement.execute(
+ "INSERT INTO last_null_identity_cache(time,device_id,site,s1) "
+ + "VALUES(3,'d0',null,30),(1,'d1','',10)");
+ }
+ // Populate the FIELD cache without initializing the time cache, then read
NULL identities.
+ assertCachedLastQueryOnAllCoordinators(
+ "SELECT last(s1) AS v FROM last_null_identity_cache", "v,", "30,");
+ assertCachedLastQueryOnAllCoordinators(
+ "SELECT last(site) AS site,last(label) AS label "
+ + "FROM last_null_identity_cache WHERE device_id='d0'",
+ "site,label,",
+ "null,null,");
+ String query =
+ "SELECT last(time) AS t,last(s1) AS v,last(site) AS site,last(label)
AS label "
+ + "FROM last_null_identity_cache WHERE device_id='d0'";
+ assertCachedLastQueryOnAllCoordinators(
+ query, "t,v,site,label,", "1970-01-01T00:00:00.003Z,30,null,null,");
+ assertCachedLastQueryOnAllCoordinators(
+ query + " GROUP BY device_id", "t,v,site,label,",
"1970-01-01T00:00:00.003Z,30,null,null,");
+ assertCachedLastQueryOnAllCoordinators(
+ "SELECT last(time) AS t,last(s1) AS v,last(site) AS site,last(label)
AS label "
+ + "FROM last_null_identity_cache WHERE device_id='d1' GROUP BY
device_id",
+ "t,v,site,label,",
+ "1970-01-01T00:00:00.001Z,10,,null,");
+ }
+
+ @Test
+ public void testLastByTagAndAttributeUseSourceField() throws SQLException {
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute(
+ "CREATE TABLE last_by_source_cache(device_id STRING TAG, site STRING
TAG, "
+ + "label STRING ATTRIBUTE, s1 INT32 FIELD, s2 INT32 FIELD)");
+ statement.execute(
+ "INSERT INTO last_by_source_cache(time,device_id,site,label,s1) "
+ + "VALUES(10,'d0',null,'blue',30),(4,'d1','','green',40)");
+ }
+
+ String emptySourceQuery =
+ "SELECT last(time) AS t,last_by(device_id,s2) AS d,last_by(label,s2)
AS a,"
+ + "last_by(time,s2) AS st FROM last_by_source_cache WHERE
device_id='d0'";
+ assertCachedLastQueryOnAllCoordinators(
+ emptySourceQuery, "t,d,a,st,",
"1970-01-01T00:00:00.010Z,null,null,null,");
+ assertCachedLastQueryOnAllCoordinators(
+ emptySourceQuery + " GROUP BY device_id",
+ "t,d,a,st,",
+ "1970-01-01T00:00:00.010Z,null,null,null,");
+
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute(
+ "INSERT INTO last_by_source_cache(time,device_id,site,s2) "
+ + "VALUES(2,'d0',null,20),(3,'d1','',15)");
+ }
+ // s1 and s2 have different last times. The partial states must carry the
source field's time,
+ // including when the selected TAG is NULL or an empty string.
+ assertCachedLastQueryOnAllCoordinators(
+ "SELECT last(s1) AS v,last_by(device_id,s2) AS d,last_by(label,s2) AS
a,"
+ + "last_by(site,s2) AS site,last_by(site,s1) AS null_site "
+ + "FROM last_by_source_cache",
+ "v,d,a,site,null_site,",
+ "30,d1,green,,null,");
+ }
+
+ @Test
+ public void testFilteredLastCacheMissDoesNotHideOlderRows() throws
SQLException {
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute(
+ "CREATE TABLE last_filter_miss_cache(device_id STRING TAG, s1 INT32
FIELD, s2 INT32 FIELD)");
+ statement.execute(
+ "INSERT INTO last_filter_miss_cache(time,device_id,s1)
VALUES(2,'d0',120),(2,'d1',120)");
+ }
+ String[] queries = {
+ "SELECT last(time) AS t,last(s1) AS v1,last(s2) AS v2 "
+ + "FROM last_filter_miss_cache WHERE device_id='d0'",
+ "SELECT last(time) AS t,last_by(s1,time) AS v1,last_by(s2,time) AS v2 "
+ + "FROM last_filter_miss_cache WHERE device_id='d1'"
+ };
+ // Start with cache misses in both LAST_VALUES and LAST_ROW paths. Empty
filtered reads must
+ // leave the older rows available to subsequent unfiltered and grouped
queries.
+ for (String query : queries) {
+ assertCachedLastQueryOnAllCoordinators(
+ query + " AND time > 2", "t,v1,v2,", "null,null,null,");
+ assertCachedLastQueryOnAllCoordinators(
+ query, "t,v1,v2,", "1970-01-01T00:00:00.002Z,120,null,");
+ assertCachedLastQueryOnAllCoordinators(
+ query + " GROUP BY device_id", "t,v1,v2,",
"1970-01-01T00:00:00.002Z,120,null,");
+ }
+ }
+
+ @Test
+ public void testLastCacheRespectsTableTTL() throws SQLException {
+ String query = "SELECT last(s1) AS v FROM last_ttl_cache WHERE
device_id='d0'";
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute("CREATE TABLE last_ttl_cache(device_id STRING TAG, s1
INT32 FIELD)");
+ statement.execute("INSERT INTO last_ttl_cache(time,device_id,s1)
VALUES(1,'d0',11)");
+ }
+ assertCachedLastQueryOnAllCoordinators(query, "v,", "11,");
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute("ALTER TABLE last_ttl_cache SET PROPERTIES TTL=1");
+ }
+ assertCachedLastQueryOnAllCoordinators(query, "v,", "null,");
+ assertCachedLastQueryOnAllCoordinators(query + " AND time >= 1", "v,",
"null,");
+ assertCachedLastQueryOnAllCoordinators(query + " GROUP BY device_id",
"v,");
+ }
+
+ @Test
+ public void testExpiredLastCacheMissDoesNotHideRowsAfterTTLIsExtended()
throws SQLException {
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute(
+ "CREATE TABLE last_ttl_miss_cache(device_id STRING TAG, s1 INT32
FIELD, s2 INT32 FIELD)");
+ statement.execute(
+ "INSERT INTO last_ttl_miss_cache(time,device_id,s1)
VALUES(2,'d0',120),(2,'d1',120)");
+ statement.execute("ALTER TABLE last_ttl_miss_cache SET PROPERTIES
TTL=1");
+ }
+ String[] queries = {
+ "SELECT last(time) AS t,last(s1) AS v1,last(s2) AS v2 "
+ + "FROM last_ttl_miss_cache WHERE device_id='d0'",
+ "SELECT last(time) AS t,last_by(s1,time) AS v1,last_by(s2,time) AS v2 "
+ + "FROM last_ttl_miss_cache WHERE device_id='d1'"
+ };
+ // The first reads miss both LAST_VALUES and LAST_ROW caches while TTL
hides the rows.
+ for (String query : queries) {
+ assertCachedLastQueryOnAllCoordinators(query, "t,v1,v2,",
"null,null,null,");
+ assertCachedLastQueryOnAllCoordinators(query + " GROUP BY device_id",
"t,v1,v2,");
+ }
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute("ALTER TABLE last_ttl_miss_cache SET PROPERTIES
TTL='INF'");
+ // The rows remain in memory; extending TTL makes them visible without
another write.
+ try (ResultSet rows =
+ statement.executeQuery(
+ "SELECT time,device_id,s1,s2 FROM last_ttl_miss_cache ORDER BY
device_id,time")) {
+ TestUtils.assertResultSetEqual(
+ rows,
+ "time,device_id,s1,s2,",
+ new String[] {
+ "1970-01-01T00:00:00.002Z,d0,120,null,",
"1970-01-01T00:00:00.002Z,d1,120,null,"
+ });
+ }
+ }
+ // Read LAST_VALUES before LAST_ROW can scan again and repair its cached
empty entries.
+ String lastValuesAfterLastRowMiss =
+ "SELECT last(time) AS t,last(s1) AS v1,last(s2) AS v2 "
+ + "FROM last_ttl_miss_cache WHERE device_id='d1'";
+ assertCachedLastQueryOnAllCoordinators(
+ lastValuesAfterLastRowMiss, "t,v1,v2,",
"1970-01-01T00:00:00.002Z,120,null,");
+ assertCachedLastQueryOnAllCoordinators(
+ lastValuesAfterLastRowMiss + " GROUP BY device_id",
+ "t,v1,v2,",
+ "1970-01-01T00:00:00.002Z,120,null,");
+ for (String query : queries) {
+ assertCachedLastQueryOnAllCoordinators(
+ query, "t,v1,v2,", "1970-01-01T00:00:00.002Z,120,null,");
+ assertCachedLastQueryOnAllCoordinators(
+ query + " GROUP BY device_id", "t,v1,v2,",
"1970-01-01T00:00:00.002Z,120,null,");
+ }
+ }
+
+ @Test
+ public void testLastByTimeOverDeletedDeviceDoesNotBreakLastCache() throws
SQLException {
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute(
+ "CREATE TABLE last_by_time_empty_cache(device_id STRING TAG, s1
INT32 FIELD, s2 INT32 FIELD)");
+ statement.execute(
+ "INSERT INTO last_by_time_empty_cache(time,device_id,s1)
VALUES(4,'d0',40),(1,'d1',10)");
+ statement.execute("DELETE FROM last_by_time_empty_cache WHERE
device_id='d0'");
+ }
+ String query =
+ "SELECT last_by(time,time) AS t FROM last_by_time_empty_cache WHERE
device_id='d0'";
+ String withFields =
+ "SELECT last_by(time,time) AS t,last_by(s1,time) AS
v1,last_by(s2,time) AS v2 "
+ + "FROM last_by_time_empty_cache WHERE device_id='d0'";
+ // The device remains in schema after DELETE. An uninitialized time
accumulator must still
+ // produce matching cache-column and cache-value arrays on every cold read.
+ assertCachedLastQueryOnAllCoordinators(query, "t,", "null,");
+ assertCachedLastQueryOnAllCoordinators(query + " GROUP BY device_id",
"t,");
+ assertCachedLastQueryOnAllCoordinators(withFields, "t,v1,v2,",
"null,null,null,");
+ assertCachedLastQueryOnAllCoordinators(withFields + " GROUP BY device_id",
"t,v1,v2,");
+ assertCachedLastQueryOnAllCoordinators(
+ "SELECT last_by(time,time) AS t FROM last_by_time_empty_cache",
+ "t,",
+ "1970-01-01T00:00:00.001Z,");
+
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ statement.execute(
+ "INSERT INTO last_by_time_empty_cache(time,device_id,s1)
VALUES(2,'d0',20)");
+ }
+ assertCachedLastQueryOnAllCoordinators(query, "t,",
"1970-01-01T00:00:00.002Z,");
+ assertCachedLastQueryOnAllCoordinators(
+ withFields, "t,v1,v2,", "1970-01-01T00:00:00.002Z,20,null,");
+ assertCachedLastQueryOnAllCoordinators(
+ withFields + " GROUP BY device_id", "t,v1,v2,",
"1970-01-01T00:00:00.002Z,20,null,");
+ assertCachedLastQueryOnAllCoordinators(
+ "SELECT last(time) AS t,last(s1) AS v1,last(s2) AS v2 "
+ + "FROM last_by_time_empty_cache WHERE device_id='d0'",
+ "t,v1,v2,",
+ "1970-01-01T00:00:00.002Z,20,null,");
+ }
+
+ private static void assertCachedLastQueryOnAllCoordinators(
+ String query, String header, String... expected) throws SQLException {
+ for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) {
+ try (Connection connection =
+ EnvFactory.getEnv()
+ .getConnection(
+ node,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ // The first read may populate the cache; the second must have the
same result.
+ for (int i = 0; i < 2; i++) {
+ try (ResultSet rows = statement.executeQuery(query)) {
+ TestUtils.assertResultSetEqual(rows, header, expected);
+ }
+ }
+ }
+ }
+ }
+
+ private static void assertFilteredLastValues(Statement statement, String
query, boolean empty)
+ throws SQLException {
+ Object[] expected = {2L, 120L, null, "d0", "blue", 2L};
+ try (ResultSet rows = statement.executeQuery(query)) {
+ assertTrue(query, rows.next());
+ for (int i = 0; i < expected.length; i++) {
+ if (empty || expected[i] == null) {
+ assertNull(query, rows.getObject(i + 1));
+ } else if (expected[i] instanceof Long) {
+ assertEquals(query, (long) expected[i], rows.getLong(i + 1));
+ assertFalse(query, rows.wasNull());
+ } else {
+ assertEquals(query, expected[i], rows.getString(i + 1));
+ }
+ }
+ assertFalse(query, rows.next());
+ }
+ }
+
+ private static void assertLastCacheRows(String... expected) {
+ TestUtils.tableResultSetEqualTest(
+ "SELECT time,device_id,s1,s2 FROM last_delete_cache ORDER BY
device_id,time",
+ new String[] {"time", "device_id", "s1", "s2"},
+ expected,
+ "test");
+ }
+
+ private static void assertLastValuesOnAllCoordinators(
+ Long expectedTime, Integer s1, Integer s2, Integer byS1, Integer byS2)
throws SQLException {
+ assertLastValuesOnAllCoordinators(expectedTime, s1, s2, byS1, byS2, false);
+ }
+
+ private static void assertLastValuesOnAllCoordinators(
+ Long expectedTime,
+ Integer s1,
+ Integer s2,
+ Integer byS1,
+ Integer byS2,
+ boolean verifyCachePath)
+ throws SQLException {
+ // Mixing last(FIELD) with last_by(FIELD,time) disables both last-cache
optimizations.
+ String[] queries = {
+ "SELECT last(time),last(s1),last(s2) FROM last_delete_cache WHERE
device_id='d0'",
+ "SELECT last(time),last_by(s1,time),last_by(s2,time) "
+ + "FROM last_delete_cache WHERE device_id='d0'"
+ };
+ Number[][] expected = {{expectedTime, s1, s2}, {expectedTime, byS1, byS2}};
+ for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) {
+ // Direct connections keep each read on the coordinator whose cache was
warmed.
+ try (Connection connection =
+ EnvFactory.getEnv()
+ .getConnection(
+ node,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE test");
+ for (int query = 0; query < queries.length; query++) {
+ String context = node.getIpAndPortString() + ": " + queries[query];
+ try (ResultSet rows = statement.executeQuery(queries[query])) {
+ assertTrue(context, rows.next());
+ for (int column = 0; column < expected[query].length; column++) {
+ String columnContext = context + ", column " + (column + 1);
+ if (expected[query][column] == null) {
+ assertNull(columnContext, rows.getObject(column + 1));
+ } else {
+ long actual = rows.getLong(column + 1);
+ assertFalse(columnContext + " is NULL", rows.wasNull());
+ assertEquals(columnContext,
expected[query][column].longValue(), actual);
+ }
+ }
+ assertFalse(context, rows.next());
+ }
+ if (verifyCachePath) {
+ StringBuilder plan = new StringBuilder();
+ try (ResultSet rows = statement.executeQuery("EXPLAIN ANALYZE " +
queries[query])) {
+ while (rows.next()) {
+ plan.append(rows.getString(1)).append('\n');
+ }
+ }
+ assertTrue(plan.toString(),
plan.toString().contains("LastQueryAggTableScanOperator"));
+ }
+ }
+ if (expectedTime == null) {
+ try (ResultSet rows =
+ statement.executeQuery(
+ "SELECT
device_id,last(time),last_by(s1,time),last_by(s2,time) "
+ + "FROM last_delete_cache WHERE device_id='d0' GROUP BY
device_id")) {
+ assertFalse(rows.next());
+ }
}
- assertEquals(299, cnt);
}
}
- cleanData(4);
}
@Test
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java
index 127e998cb5d..d24c4737c25 100644
---
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.relational.it.db.it;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.it.utils.DeletionTestUtils;
import org.apache.iotdb.db.it.utils.TestUtils;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -28,6 +29,7 @@ import org.apache.iotdb.itbase.category.TableClusterIT;
import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.tsfile.file.metadata.IDeviceID;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
@@ -37,15 +39,20 @@ import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.nio.file.Path;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
import static org.apache.iotdb.db.utils.constant.TestConstant.TIMESTAMP_STR;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@RunWith(IoTDBTestRunner.class)
@@ -56,7 +63,18 @@ public class IoTDBRestartTableIT {
@Before
public void setUp() throws Exception {
- EnvFactory.getEnv().getConfig().getCommonConfig().setWalMode("SYNC");
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setWalMode("SYNC")
+
.setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.simple.SimpleConsensus")
+ .setDataReplicationFactor(1)
+ .setDataRegionGroupExtensionPolicy("CUSTOM")
+ .setDefaultDataRegionGroupNumPerDatabase(1)
+ .setMemtableSizeThreshold(16 * 1024 * 1024)
+ .setEnableSeqSpaceCompaction(false)
+ .setEnableUnseqSpaceCompaction(false)
+ .setEnableCrossSpaceCompaction(false);
EnvFactory.getEnv()
.getConfig()
.getCommonConfig()
@@ -121,50 +139,119 @@ public class IoTDBRestartTableIT {
}
}
- @Ignore // data deletion
@Test
- public void testRestartDelete() throws SQLException {
+ public void testRestartDelete() throws Exception {
try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
Statement statement = connection.createStatement()) {
+ connection.setClientInfo("time_zone", "+00:00");
+ statement.execute("create database test");
statement.execute("use \"test\"");
- statement.execute("insert into root.turbine.d1(time,s1) values(1,1)");
- statement.execute("insert into root.turbine.d1(time,s1) values(2,2)");
- statement.execute("insert into root.turbine.d1(time,s1) values(3,3)");
+ statement.execute(
+ "create table restart_delete(device_id string tag, site string tag, "
+ + "s1 int64 field, s2 boolean field)");
+ statement.execute(
+ "create table restart_keep(device_id string tag, site string tag, "
+ + "s1 int64 field, s2 boolean field)");
+ statement.execute(
+ "insert into restart_delete(time,device_id,site,s1,s2) "
+ + "values(1,'d1','north',11,true),(2,'d1','north',12,false),"
+ +
"(2,'d1','south',212,true),(2,'d1',null,52,true),(2,'d1','',62,false)");
+ statement.execute(
+ "insert into restart_keep(time,device_id,site,s1,s2) "
+ + "values(1,'d1','north',311,true),(2,'d1','north',312,false)");
+ statement.execute("flush");
+ // Identical TAG text in two tables, and NULL/empty trailing TAGs,
coexist in one file.
+ DeletionTestUtils.assertSharedTsFile(
+ "test",
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[]
{"restart_delete", "d1", "north"}),
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[]
{"restart_delete", "d1", "south"}),
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[]
{"restart_delete", "d1", null}),
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[]
{"restart_delete", "d1", ""}),
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[]
{"restart_keep", "d1", "north"}));
+ List<Path> closedFiles = DeletionTestUtils.getClosedTsFiles("test",
true);
+ statement.execute(
+ "insert into restart_delete(time,device_id,site,s1)
values(1,'d1','north',111)");
+ statement.execute(
+ "insert into restart_delete(time,device_id,site,s1,s2) "
+ + "values(3,'d1','north',13,true),(4,'d1','north',14,false)");
+ assertEquals(closedFiles, DeletionTestUtils.getClosedTsFiles("test",
true));
+ assertTrue(DeletionTestUtils.getClosedTsFiles("test", false).isEmpty());
+
+ statement.execute(
+ "delete from restart_delete where device_id='d1' and site='north'
and time <= 3");
+ assertRestartTableData(statement,
"1970-01-01T00:00:00.004Z,d1,north,14,false,");
+
+ // This older rewrite is issued after DELETE has completed and is
protected by the SYNC WAL.
+ statement.execute(
+ "insert into restart_delete(time,device_id,site,s1)
values(2,'d1','north',1202)");
+ assertEquals(closedFiles, DeletionTestUtils.getClosedTsFiles("test",
true));
+ assertTrue(DeletionTestUtils.getClosedTsFiles("test", false).isEmpty());
+ assertRestartTableData(
+ statement,
+ "1970-01-01T00:00:00.002Z,d1,north,1202,null,",
+ "1970-01-01T00:00:00.004Z,d1,north,14,false,");
}
- TestUtils.restartDataNodes();
+ TestUtils.stopForciblyAndRestartDataNodes();
try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
Statement statement = connection.createStatement()) {
+ connection.setClientInfo("time_zone", "+00:00");
statement.execute("use \"test\"");
- statement.execute("delete from root.turbine.d1.s1 where time<=1");
-
- ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.turbine.d1");
- assertNotNull(resultSet);
- String[] exp = new String[] {"2,2.0", "3,3.0"};
- int cnt = 0;
- try {
- while (resultSet.next()) {
- String result = resultSet.getString(TIMESTAMP_STR) + "," +
resultSet.getString(2);
- assertEquals(exp[cnt], result);
- cnt++;
- }
+ assertRestartTableData(
+ statement,
+ "1970-01-01T00:00:00.002Z,d1,north,1202,null,",
+ "1970-01-01T00:00:00.004Z,d1,north,14,false,");
+ // Full data deletion must also preserve the other identities sharing
the original file.
+ statement.execute("delete from restart_delete where device_id='d1' and
site='north'");
+ assertRestartTableData(statement);
+ statement.execute(
+ "insert into restart_delete(time,device_id,site,s1)
values(1,'d1','north',1101)");
+ statement.execute("flush");
+ }
- statement.execute("flush");
- statement.execute("delete from root.turbine.d1.s1 where time<=2");
+ TestUtils.stopForciblyAndRestartDataNodes();
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ connection.setClientInfo("time_zone", "+00:00");
+ statement.execute("use test");
+ assertRestartTableData(statement,
"1970-01-01T00:00:00.001Z,d1,north,1101,null,");
+ statement.execute(
+ "insert into restart_delete(time,device_id,site,s1,s2)
values(5,'d1','north',15,true)");
+ statement.execute("flush");
+ assertRestartTableData(
+ statement,
+ "1970-01-01T00:00:00.001Z,d1,north,1101,null,",
+ "1970-01-01T00:00:00.005Z,d1,north,15,true,");
+ }
+ }
- exp = new String[] {"3,3.0"};
- resultSet = statement.executeQuery("SELECT s1 FROM root.turbine.d1");
- assertNotNull(resultSet);
- cnt = 0;
- while (resultSet.next()) {
- String result = resultSet.getString(TIMESTAMP_STR) + "," +
resultSet.getString(2);
- assertEquals(exp[cnt], result);
- cnt++;
- }
- } finally {
- resultSet.close();
- }
+ private static void assertRestartTableData(Statement statement, String...
targetRows)
+ throws SQLException {
+ List<String> expected =
+ new ArrayList<>(
+ Arrays.asList(
+ "1970-01-01T00:00:00.002Z,d1,null,52,true,",
+ "1970-01-01T00:00:00.002Z,d1,,62,false,"));
+ expected.addAll(Arrays.asList(targetRows));
+ expected.add("1970-01-01T00:00:00.002Z,d1,south,212,true,");
+ try (ResultSet rows =
+ statement.executeQuery(
+ "select time,device_id,site,s1,s2 from restart_delete "
+ + "order by device_id,site nulls first,time")) {
+ TestUtils.assertResultSetEqual(
+ rows, "time,device_id,site,s1,s2,", expected.toArray(new String[0]));
+ }
+ try (ResultSet rows =
+ statement.executeQuery(
+ "select time,device_id,site,s1,s2 from restart_keep order by
device_id,site,time")) {
+ TestUtils.assertResultSetEqual(
+ rows,
+ "time,device_id,site,s1,s2,",
+ new String[] {
+ "1970-01-01T00:00:00.001Z,d1,north,311,true,",
+ "1970-01-01T00:00:00.002Z,d1,north,312,false,"
+ });
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java
index 3e3b8beee01..a1cd2af1de2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java
@@ -168,7 +168,7 @@ public class DataNodeQueryContext {
deviceCountAndMeasurementValues.get(tableName).get(deviceEntry).getRight();
// if a device hits cache each time, the values recorded in context will
be null
if (values != null) {
- TABLE_DEVICE_SCHEMA_CACHE.updateLastCacheIfExists(
+ TABLE_DEVICE_SCHEMA_CACHE.updateLastCacheFromQuery(
tableName.getDatabaseName(),
deviceEntry.getDeviceID(),
values.keySet().toArray(new String[0]),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
index 3c1e19a80d8..3753a717369 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
@@ -76,7 +76,6 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
private final int allDeviceCount;
private final boolean needUpdateCache;
- private final boolean needUpdateNullEntry;
private final List<Integer> hitCachesIndexes;
private final List<Pair<OptionalLong, TsPrimitiveType[]>>
lastRowCacheResults;
private final List<TimeValuePair[]> lastValuesCacheResults;
@@ -105,8 +104,6 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
this.cachedDeviceEntries = cachedDeviceEntries;
this.needUpdateCache =
LastQueryUtil.needUpdateCache(parameter.seriesScanOptions.getGlobalTimeFilter());
- this.needUpdateNullEntry =
-
LastQueryUtil.needUpdateNullEntry(parameter.seriesScanOptions.getGlobalTimeFilter());
this.hitCachesIndexes = hitCachesIndexes;
this.lastRowCacheResults = lastRowCacheResults;
this.lastValuesCacheResults = lastValuesCacheResults;
@@ -322,11 +319,15 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
private void buildResultUseLastValuesCache() {
TimeValuePair[] currentHitResult =
lastValuesCacheResults.get(currentHitCacheIndex);
- // if it is EMPTY_PRIMITIVE_TYPE, means there is no data in device
- TsPrimitiveType timeLastValue = currentHitResult[currentHitResult.length -
1].getValue();
+ // The time column uses PLACEHOLDER_NO_VALUE even when its timestamp
identifies a real row.
+ // Only an empty time entry means that the device has no data. Without a
time entry, an
+ // empty final FIELD must not discard other fields' partial results.
+ boolean deviceHasNoData =
+ currentHitResult.length > measurementCount
+ && currentHitResult[measurementCount] == PLACEHOLDER_EMPTY_COLUMN;
// when there is no data, no need to append result if the query is GROUP
BY or output of
// aggregator is partial (final operator will produce NULL result)
- if (timeLastValue == PLACEHOLDER_NO_VALUE
+ if (deviceHasNoData
&& (groupingKeySize != 0 ||
tableAggregators.get(0).getStep().isOutputPartial())) {
outputDeviceIndex++;
currentHitCacheIndex++;
@@ -350,7 +351,7 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
getNthIdColumnValue(
cachedDeviceEntries.get(currentHitCacheIndex),
aggColumnsIndexArray[columnIdx]);
if (aggregator.getAccumulator() instanceof LastDescAccumulator) {
- if (timeLastValue == PLACEHOLDER_NO_VALUE || id == null) {
+ if (deviceHasNoData || id == null) {
columnBuilder.appendNull();
} else {
if (aggregator.getStep().isOutputPartial()) {
@@ -368,11 +369,13 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
}
} else {
// last_by
- int measurementIdx =
aggColumnsIndexArray[aggregatorInputChannels.get(channel)];
- long lastTime =
-
lastValuesCacheResults.get(currentHitCacheIndex)[measurementIdx].getTimestamp();
+ int measurementIdx =
aggColumnsIndexArray[aggregatorInputChannels.get(channel + 1)];
+ TimeValuePair sourceLast = currentHitResult[measurementIdx];
+ long lastTime = sourceLast.getTimestamp();
- if (timeLastValue == PLACEHOLDER_NO_VALUE || id == null) {
+ if (sourceLast.getValue() == PLACEHOLDER_NO_VALUE) {
+ columnBuilder.appendNull();
+ } else if (id == null) {
if (aggregator.getStep().isOutputPartial()) {
columnBuilder.writeBinary(
new Binary(
@@ -403,7 +406,7 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
cachedDeviceEntries.get(currentHitCacheIndex)
.getAttributeColumnValues()[aggColumnsIndexArray[columnIdx]];
if (aggregator.getAccumulator() instanceof LastDescAccumulator) {
- if (timeLastValue == PLACEHOLDER_NO_VALUE || attribute == null) {
+ if (deviceHasNoData || attribute == null) {
columnBuilder.appendNull();
} else {
if (aggregator.getStep().isOutputPartial()) {
@@ -419,12 +422,14 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
}
}
} else {
- int measurementIdx =
aggColumnsIndexArray[aggregatorInputChannels.get(channel)];
- long lastTime =
-
lastValuesCacheResults.get(currentHitCacheIndex)[measurementIdx].getTimestamp();
+ int measurementIdx =
aggColumnsIndexArray[aggregatorInputChannels.get(channel + 1)];
+ TimeValuePair sourceLast = currentHitResult[measurementIdx];
+ long lastTime = sourceLast.getTimestamp();
// last_by
- if (timeLastValue == PLACEHOLDER_NO_VALUE || attribute == null) {
+ if (sourceLast.getValue() == PLACEHOLDER_NO_VALUE) {
+ columnBuilder.appendNull();
+ } else if (attribute == null) {
if (aggregator.getStep().isOutputPartial()) {
columnBuilder.writeBinary(
new Binary(
@@ -452,7 +457,7 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
case TIME:
if (aggregator.getAccumulator() instanceof LastDescAccumulator) {
// for last(time) aggregation
- if (timeLastValue == PLACEHOLDER_NO_VALUE) {
+ if (deviceHasNoData) {
columnBuilder.appendNull();
} else {
if (aggregator.getStep().isOutputPartial()) {
@@ -477,14 +482,7 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
if (tsPrimitiveType == PLACEHOLDER_NO_VALUE) {
// there is no data
- if (aggregator.getStep().isOutputPartial()) {
- columnBuilder.writeBinary(
- new Binary(
- serializeTimeValueWithNull(
- getTSDataType(schema.getType()), lastTime, true,
false, null)));
- } else {
- columnBuilder.appendNull();
- }
+ columnBuilder.appendNull();
} else {
if (aggregator.getStep().isOutputPartial()) {
// output: xDataType, yLastTime, xIsNull, isOrderTimeNull,
xResult
@@ -547,6 +545,10 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
boolean hasSetLastTime = false;
for (int i = 0; i < tableAggregators.size(); i++) {
TableAggregator tableAggregator = tableAggregators.get(i);
+ if (!canUpdateLastCache(tableAggregator)) {
+ channel += tableAggregator.getChannelCount();
+ continue;
+ }
ColumnSchema schema =
aggColumnSchemas.get(aggregatorInputChannels.get(channel));
switch (schema.getColumnCategory()) {
@@ -574,6 +576,8 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
new TimeValuePair(
lastByAccumulator.getLastTimeOfY(),
new
TsPrimitiveType.TsLong(lastByAccumulator.getLastTimeOfY())));
+ } else {
+ updateTimeValuePairList.add(PLACEHOLDER_EMPTY_COLUMN);
}
}
}
@@ -618,6 +622,10 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
List<TimeValuePair> updateTimeValuePairList = new ArrayList<>();
boolean hasSetLastTime = false;
for (TableAggregator tableAggregator : tableAggregators) {
+ if (!canUpdateLastCache(tableAggregator)) {
+ channel += tableAggregator.getChannelCount();
+ continue;
+ }
ColumnSchema schema =
aggColumnSchemas.get(aggregatorInputChannels.get(channel));
switch (schema.getColumnCategory()) {
@@ -625,16 +633,19 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
case TAG:
case ATTRIBUTE:
if (!hasSetLastTime && tableAggregator.getAccumulator() instanceof
LastDescAccumulator) {
- hasSetLastTime = true;
LastDescAccumulator lastAccumulator =
(LastDescAccumulator) tableAggregator.getAccumulator();
- updateMeasurementList.add("");
if (lastAccumulator.hasInitResult()) {
+ hasSetLastTime = true;
+ updateMeasurementList.add("");
updateTimeValuePairList.add(
new TimeValuePair(
lastAccumulator.getMaxTime(),
new
TsPrimitiveType.TsLong(lastAccumulator.getMaxTime())));
- } else {
+ } else if (schema.getColumnCategory() ==
TsTableColumnCategory.TIME) {
+ // A NULL TAG or ATTRIBUTE does not imply that the device has no
rows.
+ hasSetLastTime = true;
+ updateMeasurementList.add("");
currentDeviceEntry = deviceEntries.get(currentDeviceIndex);
updateTimeValuePairList.add(PLACEHOLDER_EMPTY_COLUMN);
}
@@ -667,6 +678,15 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
checkIfUpdated(updateMeasurementList, updateTimeValuePairList);
}
+ private boolean canUpdateLastCache(TableAggregator aggregator) {
+ // TTL is added to the time filter when the scan's data source is
initialized.
+ // An empty result under a lower time bound does not prove that older data
is absent.
+ return
LastQueryUtil.needUpdateNullEntry(seriesScanOptions.getGlobalTimeFilter())
+ || (aggregator.getAccumulator() instanceof LastAccumulator
+ ? ((LastAccumulator) aggregator.getAccumulator()).hasInitResult()
+ : ((LastByDescAccumulator)
aggregator.getAccumulator()).hasInitResult());
+ }
+
private void checkIfUpdated(
List<String> updateMeasurementList, List<TimeValuePair>
updateTimeValuePairList) {
if (!updateMeasurementList.isEmpty()) {
@@ -675,7 +695,7 @@ public class LastQueryAggTableScanOperator extends
AbstractAggTableScanOperator
boolean deviceInMultiRegion =
deviceCountMap != null &&
deviceCountMap.containsKey(currentDeviceEntry);
if (!deviceInMultiRegion) {
- TABLE_DEVICE_SCHEMA_CACHE.updateLastCacheIfExists(
+ TABLE_DEVICE_SCHEMA_CACHE.updateLastCacheFromQuery(
dbName,
currentDeviceEntry.getDeviceID(),
updateMeasurementList.toArray(new String[0]),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java
index 2fae9aa5370..01584fe2307 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java
@@ -214,7 +214,7 @@ import static
org.apache.iotdb.db.queryengine.execution.operator.source.relation
import static
org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils.convertPredicateToFilter;
import static
org.apache.iotdb.db.queryengine.plan.planner.OperatorTreeGenerator.isFilterGtOrGe;
import static
org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions.updateFilterUsingTTL;
-import static
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache.PLACEHOLDER_NO_VALUE;
+import static
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache.PLACEHOLDER_EMPTY_COLUMN;
import static
org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache.PLACEHOLDER_STALE_VALUE;
import static org.apache.tsfile.read.common.type.TimestampType.TIMESTAMP;
@@ -1911,23 +1911,18 @@ public class DataNodeTableOperatorGenerator
"",
parameter.getMeasurementColumnNames());
boolean allHitCache = true;
- if (lastByResult.isPresent() &&
lastByResult.get().getLeft().isPresent()) {
+ // Check the aligned row's time even when the query has no FIELD
columns.
+ if (lastByResult.isPresent()
+ && lastByResult.get().getLeft().isPresent()
+ && (updateTimeFilter == null
+ ||
updateTimeFilter.satisfy(lastByResult.get().getLeft().getAsLong(), null))) {
for (int j = 0; j < lastByResult.get().getRight().length; j++) {
TsPrimitiveType tsPrimitiveType =
lastByResult.get().getRight()[j];
if (tsPrimitiveType == null
// Known-null at the aligned row time can still hit cache.
Only miss or stale
// target
// values need to fall back to scan for correctness.
- || tsPrimitiveType == PLACEHOLDER_STALE_VALUE
- || (updateTimeFilter != null
- && !LastQueryUtil.satisfyFilter(
- updateTimeFilter,
- new TimeValuePair(
- lastByResult.get().getLeft().getAsLong(),
tsPrimitiveType)))) {
- // the process logic is different from tree model which
examine if
- //
`isFilterGtOrGe(seriesScanOptions.getGlobalTimeFilter())`, set
- // `lastByResult.get().getRight()[j] = EMPTY_PRIMITIVE_TYPE`,
- // but it should skip in table model
+ || tsPrimitiveType == PLACEHOLDER_STALE_VALUE) {
allHitCache = false;
break;
}
@@ -2014,18 +2009,18 @@ public class DataNodeTableOperatorGenerator
targetColumns);
boolean allHitCache = true;
if (lastResult != null) {
- for (TimeValuePair timeValuePair : lastResult) {
+ for (int k = 0; k < lastResult.length; k++) {
+ TimeValuePair timeValuePair = lastResult[k];
if (timeValuePair == null || timeValuePair.getValue() == null)
{
allHitCache = false;
break;
}
if (updateTimeFilter != null
- && !LastQueryUtil.satisfyFilter(
-
parameter.getSeriesScanOptions().getGlobalTimeFilter(), timeValuePair)) {
+ && !LastQueryUtil.satisfyFilter(updateTimeFilter,
timeValuePair)) {
if (isFilterGtOrGe(updateTimeFilter)) {
- // it means there is no data meets Filter
- timeValuePair.setValue(PLACEHOLDER_NO_VALUE);
+ // No row satisfies this query. Keep the shared cached
pair unchanged.
+ lastResult[k] = PLACEHOLDER_EMPTY_COLUMN;
} else {
allHitCache = false;
break;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java
index 394cd509571..c7f67307a70 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java
@@ -224,6 +224,14 @@ public class TableDeviceCacheEntry {
return Objects.nonNull(lastCache.get()) ? result : 0;
}
+ int tryUpdateLastCacheFromQuery(
+ final String[] measurements, final TimeValuePair[] timeValuePairs) {
+ final TableDeviceLastCache cache = lastCache.get();
+ final int result =
+ Objects.nonNull(cache) ? cache.tryUpdateFromQuery(measurements,
timeValuePairs) : 0;
+ return Objects.nonNull(lastCache.get()) ? result : 0;
+ }
+
int tryUpdateLastCache(final String[] measurements, final TimeValuePair[]
timeValuePairs) {
return tryUpdateLastCache(measurements, timeValuePairs, false);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
index 5c79457bada..f36c200207d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
@@ -144,6 +144,22 @@ public class TableDeviceLastCache {
final @Nullable IMeasurementSchema[] measurementSchemas,
final @Nonnull TimeValuePair[] timeValuePairs,
final boolean invalidateNull) {
+ return tryUpdate(measurements, measurementSchemas, timeValuePairs,
invalidateNull, true);
+ }
+
+ int tryUpdateFromQuery(
+ final @Nonnull String[] measurements, final @Nonnull TimeValuePair[]
timeValuePairs) {
+ // A FIELD's last non-NULL value can be older than the device's latest
row. Queries must
+ // provide an explicit time entry when they can establish the actual
last-row time.
+ return tryUpdate(measurements, null, timeValuePairs, false, false);
+ }
+
+ private int tryUpdate(
+ final @Nonnull String[] measurements,
+ final @Nullable IMeasurementSchema[] measurementSchemas,
+ final @Nonnull TimeValuePair[] timeValuePairs,
+ final boolean invalidateNull,
+ final boolean updateLastTimeFromMeasurements) {
int diff = 0;
long lastTime = Long.MIN_VALUE;
@@ -172,7 +188,9 @@ public class TableDeviceLastCache {
diff += tryUpdateCachedLast(measurement, timeValuePair);
}
}
- tryUpdateLastTime(lastTime);
+ if (updateLastTimeFromMeasurements) {
+ tryUpdateLastTime(lastTime);
+ }
return diff;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
index 1c11148ffc4..17f691d8542 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
@@ -341,6 +341,20 @@ public class TableDeviceSchemaCache {
false);
}
+ /** Update query results without inferring the device's last-row time from
FIELD timestamps. */
+ public void updateLastCacheFromQuery(
+ final String database,
+ final IDeviceID deviceId,
+ final String[] measurements,
+ final TimeValuePair[] timeValuePairs) {
+ dualKeyCache.update(
+ new TableId(database, deviceId.getTableName()),
+ deviceId,
+ null,
+ entry -> entry.tryUpdateLastCacheFromQuery(measurements,
timeValuePairs),
+ false);
+ }
+
/**
* Update the last cache in writing or the second push of last cache query.
If a measurement is
* with all {@code null}s or is a tag/attribute column, its {@link
TimeValuePair}[] shall be
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
index c4550898a05..f4c6c193b55 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
@@ -19,7 +19,9 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.recover.file;
+import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.MeasurementPath;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
@@ -35,13 +37,16 @@ import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalIn
import org.apache.iotdb.db.service.metrics.WritingMetrics;
import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
import org.apache.iotdb.db.storageengine.dataregion.memtable.PrimitiveMemTable;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate;
import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
import
org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.FullExactMatch;
import
org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.utils.Pair;
import java.io.IOException;
@@ -49,6 +54,8 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import static
org.apache.iotdb.commons.conf.IoTDBConstant.ONE_LEVEL_PATH_WILDCARD;
+
/**
* This class helps redo wal logs into a TsFile. Notice: You should update
time map in {@link
* TsFileResource} before using this class to avoid duplicated insertion and
this class doesn't
@@ -79,16 +86,63 @@ public class TsFilePlanRedoer {
new TreeDeletionEntry(
path, deleteDataNode.getDeleteStartTime(),
deleteDataNode.getDeleteEndTime());
recoveryMemTable.delete(deletionEntry, ignoreAllNullRows);
- deletionEntries.add(deletionEntry);
+ deletionEntries.addAll(deletionsForFlushedData(deletionEntry));
+ }
+ if (!deletionEntries.isEmpty()) {
+ tsFileResource.getModFileForWrite().write(deletionEntries);
}
- tsFileResource.getModFileForWrite().write(deletionEntries);
}
void redoDelete(RelationalDeleteDataNode node) throws IOException {
+ List<ModEntry> deletionEntries = new ArrayList<>();
for (TableDeletionEntry modEntry : node.getModEntries()) {
recoveryMemTable.delete(modEntry, ignoreAllNullRows);
+ deletionEntries.addAll(deletionsForFlushedData(modEntry));
+ }
+ if (!deletionEntries.isEmpty()) {
+ tsFileResource.getModFileForWrite().write(deletionEntries);
+ }
+ }
+
+ private List<ModEntry> deletionsForFlushedData(ModEntry deletion) throws
IOException {
+ // WAL inserts are replayed into memory and flushed after all deletions
have been replayed.
+ // A file-wide deletion would also hide later inserts. The resource still
describes only the
+ // data already on disk, and redoInsert skips that data using each
device's end time.
+ List<ModEntry> deletions = new ArrayList<>();
+ for (IDeviceID deviceID : tsFileResource.getDevices()) {
+ long startTime = tsFileResource.getStartTime(deviceID).get();
+ long endTime = tsFileResource.getEndTime(deviceID).get();
+ if (startTime > endTime || !deletion.affects(deviceID, startTime,
endTime)) {
+ continue;
+ }
+ TimeRange timeRange =
+ new TimeRange(deletion.getStartTime(),
Math.min(deletion.getEndTime(), endTime));
+ if (deletion instanceof TreeDeletionEntry) {
+ try {
+ PartialPath pathPattern = ((TreeDeletionEntry)
deletion).getPathPattern();
+ deletions.add(
+ new TreeDeletionEntry(
+ new MeasurementPath(
+ deviceID,
+ pathPattern.endWithMultiLevelWildcard()
+ ? ONE_LEVEL_PATH_WILDCARD
+ : pathPattern.getMeasurement()),
+ timeRange));
+ } catch (IllegalPathException e) {
+ throw new IOException(e);
+ }
+ } else {
+ TableDeletionEntry tableDeletion = (TableDeletionEntry) deletion;
+ deletions.add(
+ new TableDeletionEntry(
+ new DeletionPredicate(
+ tableDeletion.getTableName(),
+ new FullExactMatch(deviceID),
+ tableDeletion.getPredicate().getMeasurementNames()),
+ timeRange));
+ }
}
- tsFileResource.getModFileForWrite().write(node.getModEntries());
+ return deletions;
}
void redoInsert(InsertNode node) throws WriteProcessException {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
index 4ae3228013e..3270e60157b 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
@@ -32,6 +32,39 @@ import java.util.concurrent.atomic.AtomicInteger;
public class TableDeviceLastCacheTest {
+ @Test
+ public void testQueryFieldTimesDoNotInitializeAlignedTime() {
+ final TableDeviceLastCache cache = new TableDeviceLastCache(false);
+ cache.initOrInvalidate(null, null, new String[] {"", "s1"}, false);
+ final TimeValuePair historicalValue = new TimeValuePair(2L, new
TsPrimitiveType.TsInt(20));
+
+ cache.tryUpdateFromQuery(new String[] {"s1"}, new TimeValuePair[]
{historicalValue});
+
+ Assert.assertEquals(historicalValue, cache.getTimeValuePair("s1"));
+ Assert.assertNull(cache.getTimeValuePair(""));
+ Assert.assertFalse(cache.getLastRow("",
Collections.singletonList("s1")).isPresent());
+
+ cache.tryUpdateFromQuery(
+ new String[] {"", "s1"},
+ new TimeValuePair[] {
+ new TimeValuePair(7L, TableDeviceLastCache.PLACEHOLDER_NO_VALUE),
+ new TimeValuePair(7L, TableDeviceLastCache.PLACEHOLDER_NO_VALUE)
+ });
+
+ Assert.assertEquals(historicalValue, cache.getTimeValuePair("s1"));
+ Optional<Pair<OptionalLong, TsPrimitiveType[]>> result =
+ cache.getLastRow("", Collections.singletonList("s1"));
+ Assert.assertTrue(result.isPresent());
+ Assert.assertEquals(OptionalLong.of(7L), result.get().getLeft());
+ Assert.assertArrayEquals(
+ new TsPrimitiveType[] {TableDeviceLastCache.PLACEHOLDER_NO_VALUE},
result.get().getRight());
+
+ final TimeValuePair newerFieldValue = new TimeValuePair(8L, new
TsPrimitiveType.TsInt(80));
+ cache.tryUpdateFromQuery(new String[] {"s1"}, new TimeValuePair[]
{newerFieldValue});
+ Assert.assertEquals(newerFieldValue, cache.getTimeValuePair("s1"));
+ Assert.assertEquals(7L, cache.getTimeValuePair("").getTimestamp());
+ }
+
@Test
public void testLazyUpdateOnlyComposesCachedMeasurements() {
final TableDeviceLastCache cache = new TableDeviceLastCache(false);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java
index a58171017ac..aa4450ce13e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java
@@ -30,11 +30,16 @@ import
org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate;
+import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.NOP;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.TsFileUtilsForRecoverTest;
import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest;
@@ -49,6 +54,7 @@ import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.read.reader.IPointReader;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
@@ -629,6 +635,124 @@ public class TsFilePlanRedoerTest {
assertTrue(modsFile.exists());
}
+ @Test
+ public void testRedoDeleteWithoutFlushedDataDoesNotCreateMods() throws
Exception {
+ File file = new File(FILE_NAME);
+ assertTrue(file.getParentFile().isDirectory() ||
file.getParentFile().mkdirs());
+ tsFileResource = new TsFileResource(file);
+ TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource);
+
+ planRedoer.redoDelete(
+ new DeleteDataNode(
+ new PlanNodeId(""),
+ Collections.singletonList(new MeasurementPath(SG_NAME + ".**")),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+ planRedoer.redoDelete(
+ new RelationalDeleteDataNode(
+ new PlanNodeId(""),
+ Collections.singletonList(
+ new TableDeletionEntry(
+ new DeletionPredicate("table1"),
+ new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE))),
+ null));
+
+ assertFalse(ModificationFile.getExclusiveMods(file).exists());
+ }
+
+ @Test
+ public void testRedoTreeDeleteDoesNotMaskLaterInserts() throws Exception {
+ File file = new File(FILE_NAME);
+ generateCompleteFile(file);
+ tsFileResource = new TsFileResource(file);
+ tsFileResource.updateStartTime(DEVICE1_NAME, 1);
+ tsFileResource.updateEndTime(DEVICE1_NAME, 2);
+ tsFileResource.updateStartTime(DEVICE2_NAME, 3);
+ tsFileResource.updateEndTime(DEVICE2_NAME, 4);
+ IDeviceID nestedDevice = IDeviceID.Factory.DEFAULT_FACTORY.create(SG_NAME
+ ".d1.child");
+ tsFileResource.updateStartTime(nestedDevice, 1);
+ tsFileResource.updateEndTime(nestedDevice, 1);
+ TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource);
+
+ planRedoer.redoDelete(
+ new DeleteDataNode(
+ new PlanNodeId(""),
+ Collections.singletonList(new MeasurementPath(SG_NAME + ".**.s1")),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+
+ List<ModEntry> mods = tsFileResource.getModFileForWrite().getAllMods();
+ for (IDeviceID device : Arrays.asList(DEVICE1_NAME, DEVICE2_NAME,
nestedDevice)) {
+ long endTime = tsFileResource.getEndTime(device).get();
+ assertTrue(mods.stream().anyMatch(mod -> mod.affects(device, endTime,
endTime)));
+ assertFalse(mods.stream().anyMatch(mod -> mod.affects(device, endTime +
1, endTime + 1)));
+ }
+ assertFalse(mods.stream().anyMatch(mod -> mod.affects(DEVICE3_NAME)));
+ assertTrue(mods.stream().allMatch(mod -> mod.affects("s1") &&
!mod.affects("s2")));
+
+ planRedoer.redoDelete(
+ new DeleteDataNode(
+ new PlanNodeId(""),
+ Collections.singletonList(new MeasurementPath(SG_NAME + ".**")),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+ List<ModEntry> allMods = tsFileResource.getModFileForWrite().getAllMods();
+ assertTrue(allMods.stream().anyMatch(mod -> mod.affects(nestedDevice) &&
mod.affects("s2")));
+ assertFalse(allMods.stream().anyMatch(mod -> mod.affects(nestedDevice, 2,
2)));
+ assertFalse(
+ allMods.stream()
+ .anyMatch(
+ mod ->
+ mod.affects(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(SG_NAME +
".d1.unflushed"))));
+ }
+
+ @Test
+ public void testRedoTableDeleteDoesNotMaskLaterInserts() throws Exception {
+ File file = new File(FILE_NAME);
+ generateCompleteFile(file);
+ tsFileResource = new TsFileResource(file);
+ List<IDeviceID> devices =
+ Arrays.asList(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1",
"d1", "north"}),
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1",
"d1", null}),
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1",
"d1", ""}));
+ for (int i = 0; i < devices.size(); i++) {
+ tsFileResource.updateStartTime(devices.get(i), 1);
+ tsFileResource.updateEndTime(devices.get(i), i + 2);
+ }
+ IDeviceID otherTable =
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table2", "d1",
"north"});
+ tsFileResource.updateStartTime(otherTable, 1);
+ tsFileResource.updateEndTime(otherTable, 10);
+ TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource);
+
+ planRedoer.redoDelete(
+ new RelationalDeleteDataNode(
+ new PlanNodeId(""),
+ Collections.singletonList(
+ new TableDeletionEntry(
+ new DeletionPredicate("table1", new NOP(),
Collections.singletonList("s1")),
+ new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE))),
+ null));
+
+ List<ModEntry> mods = tsFileResource.getModFileForWrite().getAllMods();
+ for (IDeviceID device : devices) {
+ long endTime = tsFileResource.getEndTime(device).get();
+ assertTrue(mods.stream().anyMatch(mod -> mod.affects(device, endTime,
endTime)));
+ assertFalse(mods.stream().anyMatch(mod -> mod.affects(device, endTime +
1, endTime + 1)));
+ }
+ assertFalse(mods.stream().anyMatch(mod -> mod.affects(otherTable)));
+ assertFalse(
+ mods.stream()
+ .anyMatch(
+ mod ->
+ mod.affects(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(
+ new String[] {"table1", "d2", "north"}))));
+ assertTrue(mods.stream().allMatch(mod -> mod.affects("s1") &&
!mod.affects("s2")));
+ }
+
@Test
public void testRedoRelationalInsertTablet()
throws IOException,
