This is an automated email from the ASF dual-hosted git repository. jiangtian pushed a commit to branch update_last_cache_in_load in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 25a6812cdb1afd68c005591f3260dfda8ada3740 Author: Tian Jiang <[email protected]> AuthorDate: Tue May 27 14:32:33 2025 +0800 add load update last cache strategies --- .../it/env/cluster/config/MppDataNodeConfig.java | 6 + .../it/env/remote/config/RemoteDataNodeConfig.java | 5 + .../apache/iotdb/itbase/env/DataNodeConfig.java | 2 + .../apache/iotdb/db/it/IoTDBLoadLastCacheIT.java | 413 +++++++++++++++++++++ .../org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 84 ----- .../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 + .../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 6 + .../pipeconsensus/PipeConsensusReceiver.java | 2 +- .../fetcher/cache/LastCacheLoadStrategy.java | 30 ++ .../fetcher/cache/TableDeviceCacheEntry.java | 13 +- .../fetcher/cache/TableDeviceLastCache.java | 11 + .../fetcher/cache/TableDeviceSchemaCache.java | 26 +- .../fetcher/cache/TreeDeviceNormalSchema.java | 4 + .../scheduler/load/LoadTsFileDispatcherImpl.java | 3 +- .../db/storageengine/dataregion/DataRegion.java | 78 +++- .../db/storageengine/load/LoadTsFileManager.java | 2 +- pom.xml | 2 +- 17 files changed, 603 insertions(+), 95 deletions(-) diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java index 5f51e486dd8..469881785e1 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java @@ -107,4 +107,10 @@ public class MppDataNodeConfig extends MppBaseConfig implements DataNodeConfig { setProperty("mqtt_payload_formatter", String.valueOf(mqttPayloadFormatter)); return this; } + + @Override + public DataNodeConfig setLoadLastCacheStrategy(String strategyName) { + setProperty("last_cache_operation_on_load", strategyName); + return this; + } } diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java index c273daba49e..f1f5146e060 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java @@ -68,4 +68,9 @@ public class RemoteDataNodeConfig implements DataNodeConfig { public DataNodeConfig setMqttPayloadFormatter(String mqttPayloadFormatter) { return this; } + + @Override + public DataNodeConfig setLoadLastCacheStrategy(String strategyName) { + return this; + } } diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java index b8c44423bf8..ce4993eeb32 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java @@ -41,4 +41,6 @@ public interface DataNodeConfig { DataNodeConfig setEnableMQTTService(boolean enableMQTTService); DataNodeConfig setMqttPayloadFormatter(String mqttPayloadFormatter); + + DataNodeConfig setLoadLastCacheStrategy(String strategyName); } diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadLastCacheIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadLastCacheIT.java new file mode 100644 index 00000000000..f05683056f0 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadLastCacheIT.java @@ -0,0 +1,413 @@ +/* + * 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; + +import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.LastCacheLoadStrategy; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.utils.TsFileGenerator; +import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; +import org.apache.iotdb.jdbc.IoTDBSQLException; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.ColumnSchema; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.v4.ITsFileWriter; +import org.apache.tsfile.write.v4.TsFileWriterBuilder; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.nio.file.Files; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.TimeUnit; + +@RunWith(Parameterized.class) +@Category({LocalStandaloneIT.class, ClusterIT.class}) +public class IoTDBLoadLastCacheIT { + + private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBLoadLastCacheIT.class); + private static final long PARTITION_INTERVAL = 10 * 1000L; + private static final int connectionTimeoutInMS = (int) TimeUnit.SECONDS.toMillis(300); + private static final long loadTsFileAnalyzeSchemaMemorySizeInBytes = 10 * 1024L; + + private File tmpDir; + private final LastCacheLoadStrategy lastCacheLoadStrategy; + + @Parameters(name = "loadLastCacheStrategy={0}") + public static Collection<Object[]> data() { + return Arrays.asList( + new Object[][] { + {LastCacheLoadStrategy.CLEAN_ALL}, + {LastCacheLoadStrategy.UPDATE}, + {LastCacheLoadStrategy.UPDATE_NO_BLOB}, + {LastCacheLoadStrategy.CLEAN_DEVICE} + }); + } + + public IoTDBLoadLastCacheIT(LastCacheLoadStrategy lastCacheLoadStrategy) { + this.lastCacheLoadStrategy = lastCacheLoadStrategy; + } + + @Before + public void setUp() throws Exception { + tmpDir = new File(Files.createTempDirectory("load").toUri()); + EnvFactory.getEnv().getConfig().getCommonConfig().setTimePartitionInterval(PARTITION_INTERVAL); + EnvFactory.getEnv() + .getConfig() + .getDataNodeConfig() + .setConnectionTimeoutInMS(connectionTimeoutInMS) + .setLoadTsFileAnalyzeSchemaMemorySizeInBytes(loadTsFileAnalyzeSchemaMemorySizeInBytes); + EnvFactory.getEnv() + .getConfig() + .getDataNodeConfig() + .setLoadLastCacheStrategy(lastCacheLoadStrategy.name()); + EnvFactory.getEnv().initClusterEnvironment(); + } + + @After + public void tearDown() throws Exception { + deleteSG(); + EnvFactory.getEnv().cleanClusterEnvironment(); + + if (!deleteDir()) { + LOGGER.error("Can not delete tmp dir for loading tsfile."); + } + } + + private void registerSchema() throws SQLException { + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + + statement.execute("CREATE DATABASE " + SchemaConfig.STORAGE_GROUP_0); + statement.execute("CREATE DATABASE " + SchemaConfig.STORAGE_GROUP_1); + + statement.execute(convert2SQL(SchemaConfig.DEVICE_0, SchemaConfig.MEASUREMENT_00)); + statement.execute(convert2SQL(SchemaConfig.DEVICE_0, SchemaConfig.MEASUREMENT_01)); + statement.execute(convert2SQL(SchemaConfig.DEVICE_0, SchemaConfig.MEASUREMENT_02)); + statement.execute(convert2SQL(SchemaConfig.DEVICE_0, SchemaConfig.MEASUREMENT_03)); + + statement.execute( + convert2AlignedSQL( + SchemaConfig.DEVICE_1, + Arrays.asList( + SchemaConfig.MEASUREMENT_10, + SchemaConfig.MEASUREMENT_11, + SchemaConfig.MEASUREMENT_12, + SchemaConfig.MEASUREMENT_13, + SchemaConfig.MEASUREMENT_14, + SchemaConfig.MEASUREMENT_15, + SchemaConfig.MEASUREMENT_16, + SchemaConfig.MEASUREMENT_17))); + + statement.execute(convert2SQL(SchemaConfig.DEVICE_2, SchemaConfig.MEASUREMENT_20)); + + statement.execute(convert2SQL(SchemaConfig.DEVICE_3, SchemaConfig.MEASUREMENT_30)); + + statement.execute( + convert2AlignedSQL( + SchemaConfig.DEVICE_4, Collections.singletonList(SchemaConfig.MEASUREMENT_40))); + } + } + + private String convert2SQL(final String device, final MeasurementSchema schema) { + final String sql = + String.format( + "create timeseries %s %s", + new Path(device, schema.getMeasurementName(), true).getFullPath(), + schema.getType().name()); + LOGGER.info("schema execute: {}", sql); + return sql; + } + + private String convert2AlignedSQL(final String device, final List<IMeasurementSchema> schemas) { + StringBuilder sql = new StringBuilder(String.format("create aligned timeseries %s(", device)); + for (int i = 0; i < schemas.size(); i++) { + final IMeasurementSchema schema = schemas.get(i); + sql.append(String.format("%s %s", schema.getMeasurementName(), schema.getType().name())); + sql.append(i == schemas.size() - 1 ? ")" : ","); + } + LOGGER.info("schema execute: {}.", sql); + return sql.toString(); + } + + private void deleteSG() throws SQLException { + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + + statement.execute(String.format("delete database %s", SchemaConfig.STORAGE_GROUP_0)); + statement.execute(String.format("delete database %s", SchemaConfig.STORAGE_GROUP_1)); + } catch (final IoTDBSQLException ignored) { + } + } + + private boolean deleteDir() { + for (final File file : Objects.requireNonNull(tmpDir.listFiles())) { + if (!file.delete()) { + return false; + } + } + return tmpDir.delete(); + } + + @Test + public void testTreeModelLoadWithLastCache() throws Exception { + registerSchema(); + + final String device = SchemaConfig.DEVICE_0; + final String measurement = SchemaConfig.MEASUREMENT_00.getMeasurementName(); + + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + + statement.execute( + String.format("insert into %s(timestamp, %s) values(100, 100)", device, measurement)); + + try (final ResultSet resultSet = + statement.executeQuery(String.format("select last %s from %s", measurement, device))) { + if (resultSet.next()) { + final String lastValue = resultSet.getString(ColumnHeaderConstant.VALUE); + Assert.assertEquals("100", lastValue); + } else { + Assert.fail("This ResultSet is empty."); + } + } + } + + final File file1 = new File(tmpDir, "1-0-0-0.tsfile"); + final File file2 = new File(tmpDir, "2-0-0-0.tsfile"); + // device 0, device 1, sg 0 + try (final TsFileGenerator generator = new TsFileGenerator(file1)) { + generator.registerTimeseries( + SchemaConfig.DEVICE_0, + Arrays.asList( + SchemaConfig.MEASUREMENT_00, + SchemaConfig.MEASUREMENT_01, + SchemaConfig.MEASUREMENT_02, + SchemaConfig.MEASUREMENT_03, + SchemaConfig.MEASUREMENT_04, + SchemaConfig.MEASUREMENT_05, + SchemaConfig.MEASUREMENT_06, + SchemaConfig.MEASUREMENT_07)); + generator.registerAlignedTimeseries( + SchemaConfig.DEVICE_1, + Arrays.asList( + SchemaConfig.MEASUREMENT_10, + SchemaConfig.MEASUREMENT_11, + SchemaConfig.MEASUREMENT_12, + SchemaConfig.MEASUREMENT_13, + SchemaConfig.MEASUREMENT_14, + SchemaConfig.MEASUREMENT_15, + SchemaConfig.MEASUREMENT_16, + SchemaConfig.MEASUREMENT_17)); + generator.generateData(SchemaConfig.DEVICE_0, 10000, PARTITION_INTERVAL / 10_000, false); + generator.generateData(SchemaConfig.DEVICE_1, 10000, PARTITION_INTERVAL / 10_000, true); + } + + // device 2, device 3, device4, sg 1 + try (final TsFileGenerator generator = new TsFileGenerator(file2)) { + generator.registerTimeseries( + SchemaConfig.DEVICE_2, Collections.singletonList(SchemaConfig.MEASUREMENT_20)); + generator.registerTimeseries( + SchemaConfig.DEVICE_3, Collections.singletonList(SchemaConfig.MEASUREMENT_30)); + generator.registerAlignedTimeseries( + SchemaConfig.DEVICE_4, Collections.singletonList(SchemaConfig.MEASUREMENT_40)); + generator.generateData(SchemaConfig.DEVICE_2, 10000, PARTITION_INTERVAL / 10_000, false); + generator.generateData(SchemaConfig.DEVICE_3, 10000, PARTITION_INTERVAL / 10_000, false); + generator.generateData(SchemaConfig.DEVICE_4, 10000, PARTITION_INTERVAL / 10_000, true); + } + + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + + statement.execute(String.format("load \"%s\" sglevel=2", tmpDir.getAbsolutePath())); + + try (final ResultSet resultSet = + statement.executeQuery(String.format("select last %s from %s", measurement, device))) { + if (resultSet.next()) { + final String lastTime = resultSet.getString(ColumnHeaderConstant.TIME); + Assert.assertEquals(String.valueOf(PARTITION_INTERVAL), lastTime); + } else { + Assert.fail("This ResultSet is empty."); + } + } + } + } + + @Test + public void testTableModelLoadWithLastCache() throws Exception { + registerSchema(); + + final String database = SchemaConfig.DATABASE_0; + final String table = SchemaConfig.TABLE_0; + final String measurement = SchemaConfig.MEASUREMENT_00.getMeasurementName(); + + try (final Connection connection = EnvFactory.getEnv().getTableConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE IF NOT EXISTS " + database); + statement.execute("USE " + database); + statement.execute( + "CREATE TABLE IF NOT EXISTS " + + table + + " (device_id STRING TAG," + + measurement + + " INT32" + + ")"); + + statement.execute( + String.format( + "insert into %s(time, device_id, %s) values(100, 'd0', 100)", table, measurement)); + + try (final ResultSet resultSet = + statement.executeQuery(String.format("select last(%s) from %s", measurement, table))) { + if (resultSet.next()) { + final String lastValue = resultSet.getString("_col0"); + Assert.assertEquals("100", lastValue); + } else { + Assert.fail("This ResultSet is empty."); + } + } + } + + final File file1 = new File(tmpDir, "1-0-0-0.tsfile"); + TableSchema tableSchema = + new TableSchema( + table, + Arrays.asList( + new ColumnSchema("device_id", TSDataType.STRING, ColumnCategory.TAG), + new ColumnSchema( + SchemaConfig.MEASUREMENT_00.getMeasurementName(), + SchemaConfig.MEASUREMENT_00.getType(), + ColumnCategory.FIELD))); + try (ITsFileWriter tsFileWriter = + new TsFileWriterBuilder().file(file1).tableSchema(tableSchema).build()) { + Tablet tablet = + new Tablet( + Arrays.asList("device_id", SchemaConfig.MEASUREMENT_00.getMeasurementName()), + Arrays.asList(TSDataType.STRING, SchemaConfig.MEASUREMENT_00.getType())); + tablet.addTimestamp(0, PARTITION_INTERVAL); + tablet.addValue(0, 0, "d0"); + tablet.addValue(0, 1, 10000); + tsFileWriter.write(tablet); + } + + try (final Connection connection = EnvFactory.getEnv().getTableConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("USE " + database); + statement.execute( + String.format( + "load '%s' with ('database-name'='%s')", tmpDir.getAbsolutePath(), database)); + + try (final ResultSet resultSet = + statement.executeQuery(String.format("select last(%s) from %s", measurement, table))) { + if (resultSet.next()) { + final String lastTime = resultSet.getString("_col0"); + Assert.assertEquals(String.valueOf(PARTITION_INTERVAL), lastTime); + } else { + Assert.fail("This ResultSet is empty."); + } + } + } + } + + private static class SchemaConfig { + + private static final String DATABASE_0 = "db"; + private static final String TABLE_0 = "test"; + private static final String STORAGE_GROUP_0 = "root.sg.test_0"; + private static final String STORAGE_GROUP_1 = "root.sg.test_1"; + + // device 0, nonaligned, sg 0 + private static final String DEVICE_0 = "root.sg.test_0.d_0"; + private static final MeasurementSchema MEASUREMENT_00 = + new MeasurementSchema("sensor_00", TSDataType.INT32, TSEncoding.RLE); + private static final MeasurementSchema MEASUREMENT_01 = + new MeasurementSchema("sensor_01", TSDataType.INT64, TSEncoding.RLE); + private static final MeasurementSchema MEASUREMENT_02 = + new MeasurementSchema("sensor_02", TSDataType.DOUBLE, TSEncoding.GORILLA); + private static final MeasurementSchema MEASUREMENT_03 = + new MeasurementSchema("sensor_03", TSDataType.TEXT, TSEncoding.PLAIN); + private static final MeasurementSchema MEASUREMENT_04 = + new MeasurementSchema("sensor_04", TSDataType.TIMESTAMP, TSEncoding.RLE); + private static final MeasurementSchema MEASUREMENT_05 = + new MeasurementSchema("sensor_05", TSDataType.DATE, TSEncoding.RLE); + private static final MeasurementSchema MEASUREMENT_06 = + new MeasurementSchema("sensor_06", TSDataType.BLOB, TSEncoding.PLAIN); + private static final MeasurementSchema MEASUREMENT_07 = + new MeasurementSchema("sensor_07", TSDataType.STRING, TSEncoding.PLAIN); + + // device 1, aligned, sg 0 + private static final String DEVICE_1 = "root.sg.test_0.a_1"; + private static final MeasurementSchema MEASUREMENT_10 = + new MeasurementSchema("sensor_10", TSDataType.INT32, TSEncoding.RLE); + private static final MeasurementSchema MEASUREMENT_11 = + new MeasurementSchema("sensor_11", TSDataType.INT64, TSEncoding.RLE); + private static final MeasurementSchema MEASUREMENT_12 = + new MeasurementSchema("sensor_12", TSDataType.DOUBLE, TSEncoding.GORILLA); + private static final MeasurementSchema MEASUREMENT_13 = + new MeasurementSchema("sensor_13", TSDataType.TEXT, TSEncoding.PLAIN); + private static final MeasurementSchema MEASUREMENT_14 = + new MeasurementSchema("sensor_14", TSDataType.TIMESTAMP, TSEncoding.RLE); + private static final MeasurementSchema MEASUREMENT_15 = + new MeasurementSchema("sensor_15", TSDataType.DATE, TSEncoding.RLE); + private static final MeasurementSchema MEASUREMENT_16 = + new MeasurementSchema("sensor_16", TSDataType.BLOB, TSEncoding.PLAIN); + private static final MeasurementSchema MEASUREMENT_17 = + new MeasurementSchema("sensor_17", TSDataType.STRING, TSEncoding.PLAIN); + + // device 2, non aligned, sg 1 + private static final String DEVICE_2 = "root.sg.test_1.d_2"; + private static final MeasurementSchema MEASUREMENT_20 = + new MeasurementSchema("sensor_20", TSDataType.INT32, TSEncoding.RLE); + + // device 3, non aligned, sg 1 + private static final String DEVICE_3 = "root.sg.test_1.d_3"; + private static final MeasurementSchema MEASUREMENT_30 = + new MeasurementSchema("sensor_30", TSDataType.INT32, TSEncoding.RLE); + + // device 4, aligned, sg 1 + private static final String DEVICE_4 = "root.sg.test_1.a_4"; + private static final MeasurementSchema MEASUREMENT_40 = + new MeasurementSchema("sensor_40", TSDataType.INT32, TSEncoding.RLE); + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java index 94b1941ec56..45d3adc67fe 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java @@ -588,90 +588,6 @@ public class IoTDBLoadTsFileIT { } } - @Test - public void testLoadWithLastCache() throws Exception { - registerSchema(); - - final String device = SchemaConfig.DEVICE_0; - final String measurement = SchemaConfig.MEASUREMENT_00.getMeasurementName(); - - try (final Connection connection = EnvFactory.getEnv().getConnection(); - final Statement statement = connection.createStatement()) { - - statement.execute( - String.format("insert into %s(timestamp, %s) values(100, 100)", device, measurement)); - - try (final ResultSet resultSet = - statement.executeQuery(String.format("select last %s from %s", measurement, device))) { - if (resultSet.next()) { - final String lastValue = resultSet.getString(ColumnHeaderConstant.VALUE); - Assert.assertEquals("100", lastValue); - } else { - Assert.fail("This ResultSet is empty."); - } - } - } - - final File file1 = new File(tmpDir, "1-0-0-0.tsfile"); - final File file2 = new File(tmpDir, "2-0-0-0.tsfile"); - // device 0, device 1, sg 0 - try (final TsFileGenerator generator = new TsFileGenerator(file1)) { - generator.registerTimeseries( - SchemaConfig.DEVICE_0, - Arrays.asList( - SchemaConfig.MEASUREMENT_00, - SchemaConfig.MEASUREMENT_01, - SchemaConfig.MEASUREMENT_02, - SchemaConfig.MEASUREMENT_03, - SchemaConfig.MEASUREMENT_04, - SchemaConfig.MEASUREMENT_05, - SchemaConfig.MEASUREMENT_06, - SchemaConfig.MEASUREMENT_07)); - generator.registerAlignedTimeseries( - SchemaConfig.DEVICE_1, - Arrays.asList( - SchemaConfig.MEASUREMENT_10, - SchemaConfig.MEASUREMENT_11, - SchemaConfig.MEASUREMENT_12, - SchemaConfig.MEASUREMENT_13, - SchemaConfig.MEASUREMENT_14, - SchemaConfig.MEASUREMENT_15, - SchemaConfig.MEASUREMENT_16, - SchemaConfig.MEASUREMENT_17)); - generator.generateData(SchemaConfig.DEVICE_0, 10000, PARTITION_INTERVAL / 10_000, false); - generator.generateData(SchemaConfig.DEVICE_1, 10000, PARTITION_INTERVAL / 10_000, true); - } - - // device 2, device 3, device4, sg 1 - try (final TsFileGenerator generator = new TsFileGenerator(file2)) { - generator.registerTimeseries( - SchemaConfig.DEVICE_2, Collections.singletonList(SchemaConfig.MEASUREMENT_20)); - generator.registerTimeseries( - SchemaConfig.DEVICE_3, Collections.singletonList(SchemaConfig.MEASUREMENT_30)); - generator.registerAlignedTimeseries( - SchemaConfig.DEVICE_4, Collections.singletonList(SchemaConfig.MEASUREMENT_40)); - generator.generateData(SchemaConfig.DEVICE_2, 10000, PARTITION_INTERVAL / 10_000, false); - generator.generateData(SchemaConfig.DEVICE_3, 10000, PARTITION_INTERVAL / 10_000, false); - generator.generateData(SchemaConfig.DEVICE_4, 10000, PARTITION_INTERVAL / 10_000, true); - } - - try (final Connection connection = EnvFactory.getEnv().getConnection(); - final Statement statement = connection.createStatement()) { - - statement.execute(String.format("load \"%s\" sglevel=2", tmpDir.getAbsolutePath())); - - try (final ResultSet resultSet = - statement.executeQuery(String.format("select last %s from %s", measurement, device))) { - if (resultSet.next()) { - final String lastTime = resultSet.getString(ColumnHeaderConstant.TIME); - Assert.assertEquals(String.valueOf(PARTITION_INTERVAL), lastTime); - } else { - Assert.fail("This ResultSet is empty."); - } - } - } - } - @Test public void testLoadWithOnNonStandardTsFileName() throws Exception { final File file1 = new File(tmpDir, "1-0-0-0.tsfile"); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index 4673e9c6637..adc0b70cd31 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java @@ -30,6 +30,7 @@ import org.apache.iotdb.db.audit.AuditLogOperation; import org.apache.iotdb.db.audit.AuditLogStorage; import org.apache.iotdb.db.exception.LoadConfigurationException; import org.apache.iotdb.db.protocol.thrift.impl.ClientRPCServiceImpl; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.LastCacheLoadStrategy; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.CrossCompactionPerformer; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.InnerSeqCompactionPerformer; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.InnerUnseqCompactionPerformer; @@ -1148,6 +1149,8 @@ public class IoTDBConfig { private CompressionType WALCompressionAlgorithm = CompressionType.LZ4; + private LastCacheLoadStrategy lastCacheLoadStrategy = LastCacheLoadStrategy.UPDATE; + IoTDBConfig() {} public int getMaxLogEntriesNumPerBatch() { @@ -4068,4 +4071,12 @@ public class IoTDBConfig { public void setWALCompressionAlgorithm(CompressionType WALCompressionAlgorithm) { this.WALCompressionAlgorithm = WALCompressionAlgorithm; } + + public LastCacheLoadStrategy getLastCacheLoadStrategy() { + return lastCacheLoadStrategy; + } + + public void setLastCacheLoadStrategy(LastCacheLoadStrategy lastCacheLoadStrategy) { + this.lastCacheLoadStrategy = lastCacheLoadStrategy; + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index 0ea2f5a097d..c927c0f4cfb 100755 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -36,6 +36,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TRatisConfig; import org.apache.iotdb.consensus.config.PipeConsensusConfig; import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; import org.apache.iotdb.db.exception.query.QueryProcessException; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.LastCacheLoadStrategy; import org.apache.iotdb.db.service.metrics.IoTDBInternalLocalReporter; import org.apache.iotdb.db.storageengine.StorageEngine; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.CrossCompactionPerformer; @@ -2261,6 +2262,11 @@ public class IoTDBDescriptor { properties.getProperty( "load_disk_select_strategy_for_pipe_and_iotv2", ILoadDiskSelector.LoadDiskSelectorType.INHERIT_LOAD.getValue())); + + conf.setLastCacheLoadStrategy( + LastCacheLoadStrategy.valueOf( + properties.getProperty( + "last_cache_operation_on_load", LastCacheLoadStrategy.UPDATE.name()))); } private void loadLoadTsFileHotModifiedProp(TrimProperties properties) throws IOException { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java index 9e5a41afc41..843cc1765c9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java @@ -719,7 +719,7 @@ public class PipeConsensusReceiver { StorageEngine.getInstance().getDataRegion(((DataRegionId) consensusGroupId)); if (region != null) { TsFileResource resource = generateTsFileResource(filePath, progressIndex); - region.loadNewTsFile(resource, true, false); + region.loadNewTsFile(resource, true, false, true); } else { // Data region is null indicates that dr has been removed or migrated. In those cases, there // is no need to replicate data. we just return success to avoid leader keeping retry diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/LastCacheLoadStrategy.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/LastCacheLoadStrategy.java new file mode 100644 index 00000000000..d9738e7d304 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/LastCacheLoadStrategy.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache; + +public enum LastCacheLoadStrategy { + // when a TsFile is loaded, read its data to update LastCache + UPDATE, + // similar to UPDATE, but will invalidate cache of Blob series instead of updating them + UPDATE_NO_BLOB, + // when a TsFile is loaded, clean its included device in LastCache + CLEAN_DEVICE, + // when a TsFile is loaded, clean all LastCache + CLEAN_ALL +} 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 a2bf23f8ef7..382ff1bfd5e 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 @@ -143,6 +143,9 @@ public class TableDeviceCacheEntry { final boolean isAligned, final String[] measurements, final IMeasurementSchema[] schemas) { + if (schemas == null) { + return 0; + } // Safe here because tree schema is invalidated by the whole entry final int result = (deviceSchema.compareAndSet(null, new TreeDeviceNormalSchema(database, isAligned)) @@ -191,12 +194,18 @@ public class TableDeviceCacheEntry { return Objects.nonNull(lastCache.get()) ? result : 0; } - int tryUpdateLastCache(final String[] measurements, final TimeValuePair[] timeValuePairs) { + int tryUpdateLastCache( + final String[] measurements, final TimeValuePair[] timeValuePairs, boolean invalidateNull) { final TableDeviceLastCache cache = lastCache.get(); - final int result = Objects.nonNull(cache) ? cache.tryUpdate(measurements, timeValuePairs) : 0; + final int result = + Objects.nonNull(cache) ? cache.tryUpdate(measurements, timeValuePairs, invalidateNull) : 0; return Objects.nonNull(lastCache.get()) ? result : 0; } + int tryUpdateLastCache(final String[] measurements, final TimeValuePair[] timeValuePairs) { + return tryUpdateLastCache(measurements, timeValuePairs, false); + } + int invalidateLastCache(final String measurement, final boolean isTableModel) { final TableDeviceLastCache cache = lastCache.get(); final int result = Objects.nonNull(cache) ? cache.invalidate(measurement, isTableModel) : 0; 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 d2cdde1d586..16f5e208f9d 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 @@ -134,13 +134,24 @@ public class TableDeviceLastCache { int tryUpdate( final @Nonnull String[] measurements, final @Nonnull TimeValuePair[] timeValuePairs) { + return tryUpdate(measurements, timeValuePairs, false); + } + + int tryUpdate( + final @Nonnull String[] measurements, + final @Nonnull TimeValuePair[] timeValuePairs, + final boolean invalidateNull) { final AtomicInteger diff = new AtomicInteger(0); long lastTime = Long.MIN_VALUE; for (int i = 0; i < measurements.length; ++i) { if (Objects.isNull(timeValuePairs[i])) { + if (invalidateNull) { + measurement2CachedLastMap.remove(measurements[i]); + } continue; } + final int finalI = i; if (lastTime < timeValuePairs[i].getTimestamp()) { lastTime = timeValuePairs[i].getTimestamp(); 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 466a74dcdc7..56577448859 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 @@ -260,20 +260,42 @@ public class TableDeviceSchemaCache { * @param deviceId {@link IDeviceID} * @param measurements the fetched measurements * @param timeValuePairs the {@link TimeValuePair}s with indexes corresponding to the measurements + * @param invalidateNull when true invalidate cache entries where timeValuePairs[i] == null; when + * false ignore cache entries where timeValuePairs[i] == null */ public void updateLastCacheIfExists( final String database, final IDeviceID deviceId, final String[] measurements, - final TimeValuePair[] timeValuePairs) { + final TimeValuePair[] timeValuePairs, + boolean invalidateNull) { dualKeyCache.update( new TableId(database, deviceId.getTableName()), deviceId, null, - entry -> entry.tryUpdateLastCache(measurements, timeValuePairs), + entry -> entry.tryUpdateLastCache(measurements, timeValuePairs, invalidateNull), 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 + * {@code null}. For correctness, this will put the cache lazily and only update the existing last + * caches of measurements. + * + * @param database the device's database, without "root" + * @param deviceId {@link IDeviceID} + * @param measurements the fetched measurements + * @param timeValuePairs the {@link TimeValuePair}s with indexes corresponding to the measurements + */ + public void updateLastCacheIfExists( + final String database, + final IDeviceID deviceId, + final String[] measurements, + final TimeValuePair[] timeValuePairs) { + updateLastCacheIfExists(database, deviceId, measurements, timeValuePairs, false); + } + /** * Get the last {@link TimeValuePair} of a measurement, the measurement shall never be "time". * diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceNormalSchema.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceNormalSchema.java index 9b44b7feead..548c7d1b5ac 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceNormalSchema.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceNormalSchema.java @@ -59,6 +59,10 @@ public class TreeDeviceNormalSchema implements IDeviceSchema { public int update(final String[] measurements, final IMeasurementSchema[] schemas) { int diff = 0; + if (schemas == null) { + return diff; + } + final int length = measurements.length; for (int i = 0; i < length; ++i) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java index 5227f230007..0bd56aa8d56 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java @@ -174,7 +174,8 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher { .loadNewTsFile( tsFileResource, ((LoadSingleTsFileNode) planNode).isDeleteAfterLoad(), - isGeneratedByPipe); + isGeneratedByPipe, + false); } catch (LoadFileException e) { LOGGER.warn("Load TsFile Node {} error.", planNode, e); TSStatus resultStatus = new TSStatus(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 1002d66fecf..2d749622ce6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -112,6 +112,7 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ArrayDeviceTimeIndex; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndexCacheRecorder; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex; @@ -150,7 +151,9 @@ import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.fileSystem.FSFactoryProducer; import org.apache.tsfile.fileSystem.FSType; import org.apache.tsfile.fileSystem.fsFactory.FSFactory; +import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.filter.basic.Filter; +import org.apache.tsfile.read.reader.TsFileLastReader; import org.apache.tsfile.utils.FSUtils; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.write.writer.RestorableTsFileIOWriter; @@ -191,6 +194,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; import static org.apache.iotdb.commons.conf.IoTDBConstant.FILE_NAME_SEPARATOR; +import static org.apache.iotdb.commons.utils.PathUtils.isTableModelDatabase; import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.SEQUENCE_TSFILE; import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.UNSEQUENCE_TSFILE; import static org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource.BROKEN_SUFFIX; @@ -2992,7 +2996,8 @@ public class DataRegion implements IDataRegionForQuery { public void loadNewTsFile( final TsFileResource newTsFileResource, final boolean deleteOriginFile, - final boolean isGeneratedByPipe) + final boolean isGeneratedByPipe, + final boolean isFromConsensus) throws LoadFileException { final File tsfileToBeInserted = newTsFileResource.getTsFile(); final long newFilePartitionId = newTsFileResource.getTimePartitionWithCheck(); @@ -3064,6 +3069,7 @@ public class DataRegion implements IDataRegionForQuery { false); } + onTsFileLoaded(newTsFileResource, isFromConsensus); logger.info("TsFile {} is successfully loaded in unsequence list.", newFileName); } catch (final DiskSpaceInsufficientException e) { logger.error( @@ -3071,11 +3077,77 @@ public class DataRegion implements IDataRegionForQuery { tsfileToBeInserted.getAbsolutePath(), tsfileToBeInserted.getParentFile().getName()); throw new LoadFileException(e); + } catch (Exception e) { + throw new LoadFileException(e); } finally { writeUnlock(); // TODO: do more precise control - if (CommonDescriptor.getInstance().getConfig().isLastCacheEnable()) { - TreeDeviceSchemaCacheManager.getInstance().cleanUp(); + } + } + + private void onTsFileLoaded(TsFileResource newTsFileResource, boolean isFromConsensus) + throws Exception { + if (CommonDescriptor.getInstance().getConfig().isLastCacheEnable() && !isFromConsensus) { + switch (IoTDBDescriptor.getInstance().getConfig().getLastCacheLoadStrategy()) { + case UPDATE_NO_BLOB: + updateLastCache(newTsFileResource, true); + break; + case UPDATE: + updateLastCache(newTsFileResource, false); + break; + case CLEAN_ALL: + // The inner cache is shared by TreeDeviceSchemaCacheManager and + // TableDeviceSchemaCacheManager, + // so cleaning either of them is enough + TreeDeviceSchemaCacheManager.getInstance().cleanUp(); + break; + case CLEAN_DEVICE: + boolean isTableModel = isTableModelDatabase(databaseName); + ITimeIndex timeIndex = newTsFileResource.getTimeIndex(); + if (timeIndex instanceof ArrayDeviceTimeIndex) { + ArrayDeviceTimeIndex deviceTimeIndex = (ArrayDeviceTimeIndex) timeIndex; + deviceTimeIndex + .getDevices() + .forEach( + deviceID -> + TableDeviceSchemaCache.getInstance() + .invalidateLastCache(isTableModel ? databaseName : null, deviceID)); + } else { + TreeDeviceSchemaCacheManager.getInstance().invalidateDatabaseLastCache(databaseName); + } + break; + default: + logger.warn( + "Unrecognized LastCacheLoadStrategy: {}, fall back to CLEAN_ALL", + IoTDBDescriptor.getInstance().getConfig().getLastCacheLoadStrategy()); + TreeDeviceSchemaCacheManager.getInstance().cleanUp(); + break; + } + } + } + + @SuppressWarnings("java:S112") + private void updateLastCache(TsFileResource newTsFileResource, boolean ignoreBlob) + throws Exception { + boolean isTableModel = isTableModelDatabase(databaseName); + + try (TsFileLastReader lastReader = + new TsFileLastReader(newTsFileResource.getTsFilePath(), true, ignoreBlob)) { + while (lastReader.hasNext()) { + Pair<IDeviceID, List<Pair<String, TimeValuePair>>> nextDevice = lastReader.next(); + IDeviceID deviceID = nextDevice.left; + String[] measurements = nextDevice.right.stream().map(Pair::getLeft).toArray(String[]::new); + TimeValuePair[] timeValuePairs = + nextDevice.right.stream().map(Pair::getRight).toArray(TimeValuePair[]::new); + if (isTableModel) { + TableDeviceSchemaCache.getInstance() + .updateLastCacheIfExists(databaseName, deviceID, measurements, timeValuePairs); + } else { + // we do not update schema here, so aligned is not relevant + TreeDeviceSchemaCacheManager.getInstance() + .updateLastCacheIfExists( + databaseName, deviceID, measurements, timeValuePairs, false, null); + } } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java index 74bf838a8d3..d6962e2ae35 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java @@ -517,7 +517,7 @@ public class LoadTsFileManager { final DataRegion dataRegion = entry.getKey().getDataRegion(); final TsFileResource tsFileResource = dataPartition2Resource.get(entry.getKey()); endTsFileResource(writer, tsFileResource, progressIndex); - dataRegion.loadNewTsFile(tsFileResource, true, isGeneratedByPipe); + dataRegion.loadNewTsFile(tsFileResource, true, isGeneratedByPipe, false); // Metrics dataRegion diff --git a/pom.xml b/pom.xml index ce623c83375..d34727fb491 100644 --- a/pom.xml +++ b/pom.xml @@ -175,7 +175,7 @@ <thrift.version>0.14.1</thrift.version> <xz.version>1.9</xz.version> <zstd-jni.version>1.5.6-3</zstd-jni.version> - <tsfile.version>2.1.0-250515-SNAPSHOT</tsfile.version> + <tsfile.version>2.1.0-250521-SNAPSHOT</tsfile.version> </properties> <!-- if we claim dependencies in dependencyManagement, then we do not claim
