This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/deletion-recovery-last-cache in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 93dec7aec4d59f83525a15a632c6a626e63ebccb Author: Caideyipi <[email protected]> AuthorDate: Mon Nov 2 17:04:52 2026 +0800 Fix deletion recovery and table last-cache correctness --- .../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,
