This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 81be9cee0ff Load: Optimized the partial path split logic in
modifications coverage judgment (#16212) && Refactored tsFile load IT && Fixed
the bug that loading non-exist table which is deleted in file may cause NPE
(#16217)
81be9cee0ff is described below
commit 81be9cee0ff312425a4d284e0507bbcccc20ca05
Author: Caideyipi <[email protected]>
AuthorDate: Thu Aug 21 14:38:13 2025 +0800
Load: Optimized the partial path split logic in modifications coverage
judgment (#16212) && Refactored tsFile load IT && Fixed the bug that loading
non-exist table which is deleted in file may cause NPE (#16217)
* fix
* fix
* 1-l
* ci-fix
* refactor
* may-final
---
.../org/apache/iotdb/it/utils/TsFileGenerator.java | 28 ++
.../iotdb/it/utils/TsFileTableGenerator.java | 17 +-
.../org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 155 ++---------
.../relational/it/db/it/IoTDBLoadTsFileIT.java | 299 +++++++++++++++++++++
.../analyze/load/LoadTsFileTableSchemaCache.java | 19 +-
.../analyze/load/LoadTsFileTreeSchemaCache.java | 24 +-
.../load/TreeSchemaAutoCreatorAndVerifier.java | 2 +-
.../db/storageengine/load/LoadTsFileManager.java | 22 +-
.../apache/iotdb/db/utils/ModificationUtils.java | 59 ++--
.../apache/iotdb/commons/path/PatternTreeMap.java | 4 +
10 files changed, 429 insertions(+), 200 deletions(-)
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/utils/TsFileGenerator.java
b/integration-test/src/main/java/org/apache/iotdb/it/utils/TsFileGenerator.java
index 896745ad648..bdc16201611 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/utils/TsFileGenerator.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/utils/TsFileGenerator.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.it.utils;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
@@ -30,6 +31,7 @@ import org.apache.tsfile.read.common.Path;
import org.apache.tsfile.write.TsFileWriter;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -225,6 +227,32 @@ public class TsFileGenerator implements AutoCloseable {
tablet.addValue(row, column, String.format("test point %d",
random.nextInt()));
}
+ public void generateDeletion(final String device) throws IOException,
IllegalPathException {
+ try (final ModificationFile modificationFile =
+ new ModificationFile(ModificationFile.getExclusiveMods(tsFile),
false)) {
+ modificationFile.write(
+ new TreeDeletionEntry(
+ new MeasurementPath(device,
IoTDBConstant.ONE_LEVEL_PATH_WILDCARD),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+ device2TimeSet.remove(device);
+ device2MeasurementSchema.remove(device);
+ }
+ }
+
+ public void generateDeletion(final String device, final MeasurementSchema
measurement)
+ throws IOException, IllegalPathException {
+ try (final ModificationFile modificationFile =
+ new ModificationFile(ModificationFile.getExclusiveMods(tsFile),
false)) {
+ modificationFile.write(
+ new TreeDeletionEntry(
+ new MeasurementPath(device, measurement.getMeasurementName()),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+ device2MeasurementSchema.get(device).remove(measurement);
+ }
+ }
+
public void generateDeletion(final String device, final int number)
throws IOException, IllegalPathException {
try (final ModificationFile modificationFile =
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/utils/TsFileTableGenerator.java
b/integration-test/src/main/java/org/apache/iotdb/it/utils/TsFileTableGenerator.java
index 6d86a5079f5..7c2bd1623de 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/utils/TsFileTableGenerator.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/utils/TsFileTableGenerator.java
@@ -19,10 +19,15 @@
package org.apache.iotdb.it.utils;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry;
+
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.exception.write.WriteProcessException;
import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.write.TsFileWriter;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
@@ -173,11 +178,13 @@ public class TsFileTableGenerator implements
AutoCloseable {
tablet.addValue(row, column, String.format("test point %d",
random.nextInt()));
}
- public long getTotalNumber() {
- return table2TimeSet.entrySet().stream()
- .mapToInt(
- entry -> entry.getValue().size() *
table2MeasurementSchema.get(entry.getKey()).size())
- .sum();
+ public void generateDeletion(final String table) throws IOException {
+ try (final ModificationFile modificationFile =
+ new ModificationFile(ModificationFile.getExclusiveMods(tsFile),
false)) {
+ modificationFile.write(
+ new TableDeletionEntry(
+ new DeletionPredicate(table), new TimeRange(Long.MIN_VALUE,
Long.MAX_VALUE)));
+ }
}
@Override
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 9c9003c2bff..a89becf1aa5 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
@@ -21,16 +21,14 @@ package org.apache.iotdb.db.it;
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
+import org.apache.iotdb.db.it.utils.TestUtils;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.it.utils.TsFileGenerator;
-import org.apache.iotdb.it.utils.TsFileTableGenerator;
import org.apache.iotdb.itbase.category.ClusterIT;
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
-import org.apache.iotdb.itbase.env.BaseEnv;
import org.apache.iotdb.jdbc.IoTDBSQLException;
-import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.common.Path;
@@ -107,8 +105,8 @@ public class IoTDBLoadTsFileIT {
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("CREATE DATABASE " + SchemaConfig.DATABASE_0);
+ statement.execute("CREATE DATABASE " + SchemaConfig.DATABASE_1);
statement.execute(convert2SQL(SchemaConfig.DEVICE_0,
SchemaConfig.MEASUREMENT_00));
statement.execute(convert2SQL(SchemaConfig.DEVICE_0,
SchemaConfig.MEASUREMENT_01));
@@ -163,8 +161,8 @@ public class IoTDBLoadTsFileIT {
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));
+ statement.execute(String.format("delete database %s",
SchemaConfig.DATABASE_0));
+ statement.execute(String.format("delete database %s",
SchemaConfig.DATABASE_1));
} catch (final IoTDBSQLException ignored) {
}
}
@@ -702,9 +700,11 @@ public class IoTDBLoadTsFileIT {
generator.registerTimeseries(
SchemaConfig.DEVICE_3,
Collections.singletonList(SchemaConfig.MEASUREMENT_30));
generator.registerAlignedTimeseries(
- SchemaConfig.DEVICE_4,
Collections.singletonList(SchemaConfig.MEASUREMENT_40));
+ SchemaConfig.DEVICE_4,
+ new ArrayList<>(Arrays.asList(SchemaConfig.MEASUREMENT_30,
SchemaConfig.MEASUREMENT_40)));
generator.generateData(SchemaConfig.DEVICE_2, 100, PARTITION_INTERVAL /
10_000, false);
generator.generateData(SchemaConfig.DEVICE_3, 100, PARTITION_INTERVAL /
10_000, false);
+ generator.generateDeletion(SchemaConfig.DEVICE_3);
generator.generateData(SchemaConfig.DEVICE_4, 100, PARTITION_INTERVAL /
10_000, true);
generator.generateDeletion(SchemaConfig.DEVICE_2, 2);
generator.generateDeletion(SchemaConfig.DEVICE_4, 2);
@@ -712,6 +712,7 @@ public class IoTDBLoadTsFileIT {
generator.generateData(SchemaConfig.DEVICE_4, 100, PARTITION_INTERVAL /
10_000, true);
generator.generateDeletion(SchemaConfig.DEVICE_2, 2);
generator.generateDeletion(SchemaConfig.DEVICE_4, 2);
+ generator.generateDeletion(SchemaConfig.DEVICE_4,
SchemaConfig.MEASUREMENT_30);
writtenPoint2 = generator.getTotalNumber();
}
@@ -731,6 +732,10 @@ public class IoTDBLoadTsFileIT {
Assert.fail("This ResultSet is empty.");
}
}
+
+ TestUtils.assertSingleResultSetEqual(
+ TestUtils.executeQueryWithRetry(statement, "count timeSeries
root.sg.**"),
+ Collections.singletonMap("count(timeseries)", "18"));
}
}
@@ -891,139 +896,9 @@ public class IoTDBLoadTsFileIT {
return pairs;
}
- @Test
- public void testLoadWithEmptyDatabaseForTableModel() throws Exception {
- final int lineCount = 10000;
-
- final List<Pair<MeasurementSchema, MeasurementSchema>> measurementSchemas =
- generateMeasurementSchemasForDataTypeConvertion();
- final List<ColumnCategory> columnCategories =
- generateTabletColumnCategory(0, measurementSchemas.size());
-
- final File file = new File(tmpDir, "1-0-0-0.tsfile");
-
- final List<IMeasurementSchema> schemaList =
- measurementSchemas.stream().map(pair ->
pair.right).collect(Collectors.toList());
-
- try (final TsFileTableGenerator generator = new
TsFileTableGenerator(file)) {
- generator.registerTable(SchemaConfig.TABLE_0, schemaList,
columnCategories);
-
- generator.generateData(SchemaConfig.TABLE_0, lineCount,
PARTITION_INTERVAL / 10_000);
- }
-
- // Prepare normal user
- try (final Connection adminCon =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
- final Statement adminStmt = adminCon.createStatement()) {
- adminStmt.execute("create user test 'password123456'");
- adminStmt.execute(
- String.format(
- "grant create, insert on %s.%s to user test",
- SchemaConfig.DATABASE_0, SchemaConfig.TABLE_0));
-
- // auto-create table
- adminStmt.execute(String.format("create database if not exists %s",
SchemaConfig.DATABASE_0));
- }
-
- try (final Connection connection =
- EnvFactory.getEnv().getConnection("test", "password123456",
BaseEnv.TABLE_SQL_DIALECT);
- final Statement statement = connection.createStatement()) {
- statement.execute(String.format("use %s", SchemaConfig.DATABASE_0));
- statement.execute(String.format("load '%s'", file.getAbsolutePath()));
- }
-
- try (final Connection adminCon =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
- final Statement adminStmt = adminCon.createStatement()) {
- adminStmt.execute(String.format("use %s", SchemaConfig.DATABASE_0));
- try (final ResultSet resultSet =
- adminStmt.executeQuery(String.format("select count(*) from %s",
SchemaConfig.TABLE_0))) {
- if (resultSet.next()) {
- Assert.assertEquals(lineCount, resultSet.getLong(1));
- } else {
- Assert.fail("This ResultSet is empty.");
- }
- }
- }
- }
-
- @Test
- @Ignore("Load with conversion is currently banned")
- public void testLoadWithConvertOnTypeMismatchForTableModel() throws
Exception {
- final int lineCount = 10000;
-
- List<Pair<MeasurementSchema, MeasurementSchema>> measurementSchemas =
- generateMeasurementSchemasForDataTypeConvertion();
- List<ColumnCategory> columnCategories =
- generateTabletColumnCategory(0, measurementSchemas.size());
-
- final File file = new File(tmpDir, "1-0-0-0.tsfile");
-
- List<MeasurementSchema> schemaList1 =
- measurementSchemas.stream().map(pair ->
pair.left).collect(Collectors.toList());
- List<IMeasurementSchema> schemaList2 =
- measurementSchemas.stream().map(pair ->
pair.right).collect(Collectors.toList());
-
- try (final TsFileTableGenerator generator = new
TsFileTableGenerator(file)) {
- generator.registerTable(SchemaConfig.TABLE_0, schemaList2,
columnCategories);
-
- generator.generateData(SchemaConfig.TABLE_0, lineCount,
PARTITION_INTERVAL / 10_000);
- }
-
- try (final Connection connection =
- EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
- final Statement statement = connection.createStatement()) {
- statement.execute(String.format("create database if not exists %s",
SchemaConfig.DATABASE_0));
- statement.execute(String.format("use %s", SchemaConfig.DATABASE_0));
- statement.execute(convert2TableSQL(SchemaConfig.TABLE_0, schemaList1,
columnCategories));
- statement.execute(
- String.format(
- "load '%s' with ('database'='%s')", file.getAbsolutePath(),
SchemaConfig.DATABASE_0));
- try (final ResultSet resultSet =
- statement.executeQuery(String.format("select count(*) from %s",
SchemaConfig.TABLE_0))) {
- if (resultSet.next()) {
- Assert.assertEquals(lineCount, resultSet.getLong(1));
- } else {
- Assert.fail("This ResultSet is empty.");
- }
- }
- }
- }
-
- private List<ColumnCategory> generateTabletColumnCategory(int tagNum, int
filedNum) {
- List<ColumnCategory> columnTypes = new ArrayList<>(tagNum + filedNum);
- for (int i = 0; i < tagNum; i++) {
- columnTypes.add(ColumnCategory.TAG);
- }
- for (int i = 0; i < filedNum; i++) {
- columnTypes.add(ColumnCategory.FIELD);
- }
- return columnTypes;
- }
-
- private String convert2TableSQL(
- final String tableName,
- final List<MeasurementSchema> schemaList,
- final List<ColumnCategory> columnCategoryList) {
- List<String> columns = new ArrayList<>();
- for (int i = 0; i < schemaList.size(); i++) {
- final MeasurementSchema measurement = schemaList.get(i);
- columns.add(
- String.format(
- "%s %s %s",
- measurement.getMeasurementName(),
- measurement.getType(),
- columnCategoryList.get(i).name()));
- }
- String tableCreation =
- String.format("create table %s(%s)", tableName, String.join(", ",
columns));
- LOGGER.info("schema execute: {}", tableCreation);
- return tableCreation;
- }
-
private static class SchemaConfig {
- private static final String DATABASE_0 = "root";
- 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";
+ private static final String DATABASE_0 = "root.sg.test_0";
+ private static final String DATABASE_1 = "root.sg.test_1";
// device 0, nonaligned, sg 0
private static final String DEVICE_0 = "root.sg.test_0.d_0";
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java
new file mode 100644
index 00000000000..5f88d50c9d9
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java
@@ -0,0 +1,299 @@
+/*
+ * 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.relational.it.db.it;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.it.utils.TsFileTableGenerator;
+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.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+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.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+public class IoTDBLoadTsFileIT {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(IoTDBLoadTsFileIT.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;
+
+ @Before
+ public void setUp() throws Exception {
+ tmpDir = new File(Files.createTempDirectory("load").toUri());
+
EnvFactory.getEnv().getConfig().getCommonConfig().setTimePartitionInterval(PARTITION_INTERVAL);
+
EnvFactory.getEnv().getConfig().getCommonConfig().setEnforceStrongPassword(false);
+ EnvFactory.getEnv()
+ .getConfig()
+ .getDataNodeConfig()
+ .setConnectionTimeoutInMS(connectionTimeoutInMS)
+
.setLoadTsFileAnalyzeSchemaMemorySizeInBytes(loadTsFileAnalyzeSchemaMemorySizeInBytes);
+ EnvFactory.getEnv().initClusterEnvironment();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+
+ if (!deleteDir()) {
+ LOGGER.error("Can not delete tmp dir for loading tsfile.");
+ }
+ }
+
+ private boolean deleteDir() {
+ for (final File file : tmpDir.listFiles()) {
+ if (!file.delete()) {
+ return false;
+ }
+ }
+ return tmpDir.delete();
+ }
+
+ private List<Pair<MeasurementSchema, MeasurementSchema>>
generateMeasurementSchemas() {
+ TSDataType[] dataTypes = {
+ TSDataType.STRING,
+ TSDataType.TEXT,
+ TSDataType.BLOB,
+ TSDataType.TIMESTAMP,
+ TSDataType.BOOLEAN,
+ TSDataType.DATE,
+ TSDataType.DOUBLE,
+ TSDataType.FLOAT,
+ TSDataType.INT32,
+ TSDataType.INT64
+ };
+ List<Pair<MeasurementSchema, MeasurementSchema>> pairs = new ArrayList<>();
+
+ for (TSDataType type : dataTypes) {
+ for (TSDataType dataType : dataTypes) {
+ String id = String.format("%s2%s", type.name(), dataType.name());
+ pairs.add(new Pair<>(new MeasurementSchema(id, type), new
MeasurementSchema(id, dataType)));
+ }
+ }
+ return pairs;
+ }
+
+ @Test
+ public void testLoadWithEmptyDatabaseForTableModel() throws Exception {
+ final int lineCount = 10000;
+
+ final List<Pair<MeasurementSchema, MeasurementSchema>> measurementSchemas =
+ generateMeasurementSchemas();
+ final List<ColumnCategory> columnCategories =
+ generateTabletColumnCategory(0, measurementSchemas.size());
+
+ final File file = new File(tmpDir, "1-0-0-0.tsfile");
+
+ final List<IMeasurementSchema> schemaList =
+ measurementSchemas.stream().map(pair ->
pair.right).collect(Collectors.toList());
+
+ try (final TsFileTableGenerator generator = new
TsFileTableGenerator(file)) {
+ generator.registerTable(SchemaConfig.TABLE_0, schemaList,
columnCategories);
+
+ generator.generateData(SchemaConfig.TABLE_0, lineCount,
PARTITION_INTERVAL / 10_000);
+ }
+
+ // Prepare normal user
+ try (final Connection adminCon =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement adminStmt = adminCon.createStatement()) {
+ adminStmt.execute("create user test 'password123456'");
+ adminStmt.execute(
+ String.format(
+ "grant create, insert on %s.%s to user test",
+ SchemaConfig.DATABASE_0, SchemaConfig.TABLE_0));
+
+ // auto-create table
+ adminStmt.execute(String.format("create database if not exists %s",
SchemaConfig.DATABASE_0));
+ }
+
+ try (final Connection connection =
+ EnvFactory.getEnv().getConnection("test", "password123456",
BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(String.format("use %s", SchemaConfig.DATABASE_0));
+ statement.execute(String.format("load '%s'", file.getAbsolutePath()));
+ }
+
+ try (final Connection adminCon =
EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement adminStmt = adminCon.createStatement()) {
+ adminStmt.execute(String.format("use %s", SchemaConfig.DATABASE_0));
+ try (final ResultSet resultSet =
+ adminStmt.executeQuery(String.format("select count(*) from %s",
SchemaConfig.TABLE_0))) {
+ if (resultSet.next()) {
+ Assert.assertEquals(lineCount, resultSet.getLong(1));
+ } else {
+ Assert.fail("This ResultSet is empty.");
+ }
+ }
+ }
+ }
+
+ @Test
+ @Ignore("Load with conversion is currently banned")
+ public void testLoadWithConvertOnTypeMismatchForTableModel() throws
Exception {
+ final int lineCount = 10000;
+
+ List<Pair<MeasurementSchema, MeasurementSchema>> measurementSchemas =
+ generateMeasurementSchemas();
+ List<ColumnCategory> columnCategories =
+ generateTabletColumnCategory(0, measurementSchemas.size());
+
+ final File file = new File(tmpDir, "1-0-0-0.tsfile");
+
+ List<MeasurementSchema> schemaList1 =
+ measurementSchemas.stream().map(pair ->
pair.left).collect(Collectors.toList());
+ List<IMeasurementSchema> schemaList2 =
+ measurementSchemas.stream().map(pair ->
pair.right).collect(Collectors.toList());
+
+ try (final TsFileTableGenerator generator = new
TsFileTableGenerator(file)) {
+ generator.registerTable(SchemaConfig.TABLE_0, schemaList2,
columnCategories);
+
+ generator.generateData(SchemaConfig.TABLE_0, lineCount,
PARTITION_INTERVAL / 10_000);
+ }
+
+ try (final Connection connection =
+ EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(String.format("create database if not exists %s",
SchemaConfig.DATABASE_0));
+ statement.execute(String.format("use %s", SchemaConfig.DATABASE_0));
+ statement.execute(convert2TableSQL(SchemaConfig.TABLE_0, schemaList1,
columnCategories));
+ statement.execute(
+ String.format(
+ "load '%s' with ('database'='%s')", file.getAbsolutePath(),
SchemaConfig.DATABASE_0));
+ try (final ResultSet resultSet =
+ statement.executeQuery(String.format("select count(*) from %s",
SchemaConfig.TABLE_0))) {
+ if (resultSet.next()) {
+ Assert.assertEquals(lineCount, resultSet.getLong(1));
+ } else {
+ Assert.fail("This ResultSet is empty.");
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testLoadWithTableMod() throws Exception {
+ final int lineCount = 10000;
+
+ List<Pair<MeasurementSchema, MeasurementSchema>> measurementSchemas =
+ generateMeasurementSchemas();
+ List<ColumnCategory> columnCategories =
+ generateTabletColumnCategory(0, measurementSchemas.size());
+
+ final File file = new File(tmpDir, "1-0-0-0.tsfile");
+
+ List<MeasurementSchema> schemaList1 =
+ measurementSchemas.stream().map(pair ->
pair.left).collect(Collectors.toList());
+
+ try (final TsFileTableGenerator generator = new
TsFileTableGenerator(file)) {
+ generator.registerTable(SchemaConfig.TABLE_0, new
ArrayList<>(schemaList1), columnCategories);
+ generator.generateData(SchemaConfig.TABLE_0, lineCount,
PARTITION_INTERVAL / 10_000);
+
+ generator.registerTable(SchemaConfig.TABLE_1, new
ArrayList<>(schemaList1), columnCategories);
+ generator.generateData(SchemaConfig.TABLE_1, lineCount,
PARTITION_INTERVAL / 10_000);
+ generator.generateDeletion(SchemaConfig.TABLE_1);
+ }
+
+ try (final Connection connection =
+ EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(String.format("create database if not exists %s",
SchemaConfig.DATABASE_0));
+ statement.execute(String.format("use %s", SchemaConfig.DATABASE_0));
+ statement.execute(
+ String.format(
+ "load '%s' with ('database'='%s')", file.getAbsolutePath(),
SchemaConfig.DATABASE_0));
+ try (final ResultSet resultSet =
+ statement.executeQuery(String.format("select count(*) from %s",
SchemaConfig.TABLE_0))) {
+ if (resultSet.next()) {
+ Assert.assertEquals(lineCount, resultSet.getLong(1));
+ } else {
+ Assert.fail("This ResultSet is empty.");
+ }
+ }
+
+ try (final ResultSet resultSet = statement.executeQuery("show tables")) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertFalse(resultSet.next());
+ }
+ }
+ }
+
+ private List<ColumnCategory> generateTabletColumnCategory(int tagNum, int
filedNum) {
+ List<ColumnCategory> columnTypes = new ArrayList<>(tagNum + filedNum);
+ for (int i = 0; i < tagNum; i++) {
+ columnTypes.add(ColumnCategory.TAG);
+ }
+ for (int i = 0; i < filedNum; i++) {
+ columnTypes.add(ColumnCategory.FIELD);
+ }
+ return columnTypes;
+ }
+
+ private String convert2TableSQL(
+ final String tableName,
+ final List<MeasurementSchema> schemaList,
+ final List<ColumnCategory> columnCategoryList) {
+ List<String> columns = new ArrayList<>();
+ for (int i = 0; i < schemaList.size(); i++) {
+ final MeasurementSchema measurement = schemaList.get(i);
+ columns.add(
+ String.format(
+ "%s %s %s",
+ measurement.getMeasurementName(),
+ measurement.getType(),
+ columnCategoryList.get(i).name()));
+ }
+ String tableCreation =
+ String.format("create table %s(%s)", tableName, String.join(", ",
columns));
+ LOGGER.info("schema execute: {}", tableCreation);
+ return tableCreation;
+ }
+
+ private static class SchemaConfig {
+ private static final String DATABASE_0 = "root";
+ private static final String TABLE_0 = "test";
+ private static final String TABLE_1 = "test1";
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java
index 7ea5154714c..1fb6da702c3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.queryengine.plan.analyze.load;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PatternTreeMap;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.db.conf.IoTDBConfig;
@@ -46,6 +47,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex;
import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryBlock;
import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryManager;
import org.apache.iotdb.db.utils.ModificationUtils;
+import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
@@ -58,7 +60,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -101,7 +102,7 @@ public class LoadTsFileTableSchemaCache {
// tableName -> Pair<device column count, device column mapping>
private Map<String, Pair<Integer, Map<Integer, Integer>>>
tableIdColumnMapper = new HashMap<>();
- private Collection<ModEntry> currentModifications;
+ private PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer>
currentModifications;
private ITimeIndex currentTimeIndex;
private long batchTable2DevicesMemoryUsageSizeInBytes = 0;
@@ -120,7 +121,7 @@ public class LoadTsFileTableSchemaCache {
this.metadata = metadata;
this.context = context;
this.currentBatchTable2Devices = new HashMap<>();
- this.currentModifications = new ArrayList<>();
+ this.currentModifications = PatternTreeMapFactory.getModsPatternTreeMap();
this.needToCreateDatabase = needToCreateDatabase;
}
@@ -385,10 +386,12 @@ public class LoadTsFileTableSchemaCache {
TsFileResource resource, TsFileSequenceReader reader) throws IOException
{
clearModificationsAndTimeIndex();
- currentModifications =
ModificationFile.readAllModifications(resource.getTsFile(), false);
- for (final ModEntry modification : currentModifications) {
- currentModificationsMemoryUsageSizeInBytes +=
modification.serializedSize();
- }
+ ModificationFile.readAllModifications(resource.getTsFile(), false)
+ .forEach(
+ modification ->
+ currentModifications.append(modification.keyOfPatternTree(),
modification));
+
+ currentModificationsMemoryUsageSizeInBytes =
currentModifications.ramBytesUsed();
// If there are too many modifications, a larger memory block is needed to
avoid frequent
// flush.
@@ -438,7 +441,7 @@ public class LoadTsFileTableSchemaCache {
}
private void clearModificationsAndTimeIndex() {
- currentModifications.clear();
+ currentModifications = PatternTreeMapFactory.getModsPatternTreeMap();
currentTimeIndex = null;
block.reduceMemoryUsage(currentModificationsMemoryUsageSizeInBytes);
block.reduceMemoryUsage(currentTimeIndexMemoryUsageSizeInBytes);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTreeSchemaCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTreeSchemaCache.java
index 59068bde706..d0335d1b3bd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTreeSchemaCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTreeSchemaCache.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.queryengine.plan.analyze.load;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.path.PatternTreeMap;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.load.LoadRuntimeOutOfMemoryException;
@@ -32,6 +33,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex;
import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryBlock;
import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryManager;
import org.apache.iotdb.db.utils.ModificationUtils;
+import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.TimeseriesMetadata;
@@ -41,8 +43,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -76,7 +76,7 @@ public class LoadTsFileTreeSchemaCache {
private Map<IDeviceID, Boolean> tsFileDevice2IsAligned;
private Set<PartialPath> alreadySetDatabases;
- private Collection<ModEntry> currentModifications;
+ private PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer>
currentModifications;
private ITimeIndex currentTimeIndex;
private long batchDevice2TimeSeriesSchemasMemoryUsageSizeInBytes = 0;
@@ -94,7 +94,7 @@ public class LoadTsFileTreeSchemaCache {
this.currentBatchDevice2TimeSeriesSchemas = new HashMap<>();
this.tsFileDevice2IsAligned = new HashMap<>();
this.alreadySetDatabases = new HashSet<>();
- this.currentModifications = new ArrayList<>();
+ this.currentModifications = PatternTreeMapFactory.getModsPatternTreeMap();
}
public Map<IDeviceID, Set<MeasurementSchema>> getDevice2TimeSeries() {
@@ -154,10 +154,12 @@ public class LoadTsFileTreeSchemaCache {
TsFileResource resource, TsFileSequenceReader reader) throws IOException
{
clearModificationsAndTimeIndex();
- currentModifications =
ModificationFile.readAllModifications(resource.getTsFile(), true);
- for (final ModEntry modification : currentModifications) {
- currentModificationsMemoryUsageSizeInBytes +=
modification.serializedSize();
- }
+ ModificationFile.readAllModifications(resource.getTsFile(), true)
+ .forEach(
+ modification ->
+ currentModifications.append(modification.keyOfPatternTree(),
modification));
+
+ currentModificationsMemoryUsageSizeInBytes =
currentModifications.ramBytesUsed();
// If there are too many modifications, a larger memory block is needed to
avoid frequent
// flush.
@@ -196,9 +198,9 @@ public class LoadTsFileTreeSchemaCache {
return ModificationUtils.isDeviceDeletedByMods(currentModifications,
currentTimeIndex, device);
}
- public boolean isTimeseriesDeletedByMods(IDeviceID device,
TimeseriesMetadata timeseriesMetadata)
+ public boolean isTimeSeriesDeletedByMods(IDeviceID device,
TimeseriesMetadata timeseriesMetadata)
throws IllegalPathException {
- return ModificationUtils.isTimeseriesDeletedByMods(
+ return ModificationUtils.isTimeSeriesDeletedByMods(
currentModifications,
device,
timeseriesMetadata.getMeasurementId(),
@@ -235,7 +237,7 @@ public class LoadTsFileTreeSchemaCache {
}
public void clearModificationsAndTimeIndex() {
- currentModifications.clear();
+ currentModifications = PatternTreeMapFactory.getModsPatternTreeMap();
currentTimeIndex = null;
block.reduceMemoryUsage(currentModificationsMemoryUsageSizeInBytes);
block.reduceMemoryUsage(currentTimeIndexMemoryUsageSizeInBytes);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
index a15a3013db2..1b4e1f908d6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
@@ -128,7 +128,7 @@ public class TreeSchemaAutoCreatorAndVerifier {
for (final TimeseriesMetadata timeseriesMetadata : entry.getValue()) {
try {
- if (schemaCache.isTimeseriesDeletedByMods(device,
timeseriesMetadata)) {
+ if (schemaCache.isTimeSeriesDeletedByMods(device,
timeseriesMetadata)) {
continue;
}
} catch (IllegalPathException e) {
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 5876a884660..0d8a3ca85b6 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
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.file.SystemFileFactory;
+import org.apache.iotdb.commons.schema.table.TsTable;
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;
@@ -463,16 +464,17 @@ public class LoadTsFileManager {
chunkData.getDevice() != null ? chunkData.getDevice().getTableName()
: null;
if (tableName != null
&& !(tableName.startsWith(TREE_MODEL_DATABASE_PREFIX) ||
tableName.equals(ROOT))) {
- writer
- .getSchema()
- .getTableSchemaMap()
- .computeIfAbsent(
- tableName,
- t ->
- TableSchema.of(
- DataNodeTableCache.getInstance()
-
.getTable(partitionInfo.getDataRegion().getDatabaseName(), t))
- .toTsFileTableSchemaNoAttribute());
+ // If the table does not exist, it means that the table is all deleted
by mods
+ final TsTable table =
+ DataNodeTableCache.getInstance()
+ .getTable(partitionInfo.getDataRegion().getDatabaseName(),
tableName);
+ if (Objects.nonNull(table)) {
+ writer
+ .getSchema()
+ .getTableSchemaMap()
+ .computeIfAbsent(
+ tableName, t ->
TableSchema.of(table).toTsFileTableSchemaNoAttribute());
+ }
}
if (!Objects.equals(chunkData.getDevice(),
dataPartition2LastDevice.get(partitionInfo))) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java
index d5b477e61e4..95d90dfa63d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java
@@ -20,12 +20,16 @@
package org.apache.iotdb.db.utils;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.AlignedPath;
+import org.apache.iotdb.commons.path.PatternTreeMap;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionPathUtils;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.impl.SettleSelectorImpl;
import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
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.TreeDeletionEntry;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex;
+import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata;
import org.apache.tsfile.file.metadata.AlignedChunkMetadata;
@@ -35,7 +39,6 @@ import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.utils.Pair;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
@@ -248,18 +251,23 @@ public class ModificationUtils {
* There are some slight differences from that in {@link SettleSelectorImpl}.
*/
public static boolean isAllDeletedByMods(
- Collection<ModEntry> modifications, IDeviceID device, long startTime,
long endTime) {
- for (ModEntry modification : modifications) {
- if (modification.affectsAll(device)
- && modification.getTimeRange().contains(startTime, endTime)) {
- return true;
- }
- }
- return false;
+ final PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer>
modifications,
+ final IDeviceID device,
+ final long startTime,
+ final long endTime)
+ throws IllegalPathException {
+ final List<ModEntry> mods =
+ modifications.getOverlapped(
+ CompactionPathUtils.getPath(device,
AlignedPath.VECTOR_PLACEHOLDER));
+ return mods.stream()
+ .anyMatch(
+ modification ->
+ modification.getTimeRange().contains(startTime, endTime)
+ && (!device.isTableModel() ||
modification.affects(device)));
}
public static boolean isAllDeletedByMods(
- Collection<ModEntry> modifications, long startTime, long endTime) {
+ final List<ModEntry> modifications, final long startTime, final long
endTime) {
if (modifications == null || modifications.isEmpty()) {
return false;
}
@@ -271,20 +279,21 @@ public class ModificationUtils {
return false;
}
- public static boolean isTimeseriesDeletedByMods(
- Collection<ModEntry> modifications,
- IDeviceID device,
- String timeseriesId,
- long startTime,
- long endTime) {
- for (ModEntry modification : modifications) {
- if (modification.affects(device)
- && modification.affects(timeseriesId)
- && modification.getTimeRange().contains(startTime, endTime)) {
- return true;
- }
- }
- return false;
+ public static boolean isTimeSeriesDeletedByMods(
+ final PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer>
modifications,
+ final IDeviceID device,
+ final String measurement,
+ final long startTime,
+ final long endTime)
+ throws IllegalPathException {
+ final List<ModEntry> mods =
+ modifications.getOverlapped(CompactionPathUtils.getPath(device,
measurement));
+ return mods.stream()
+ .anyMatch(
+ modification ->
+ modification.getTimeRange().contains(startTime, endTime)
+ && (!device.isTableModel()
+ || modification.affects(device) &&
modification.affects(measurement)));
}
private static void doModifyChunkMetaData(ModEntry modification,
IChunkMetadata metaData) {
@@ -403,7 +412,7 @@ public class ModificationUtils {
}
public static boolean isDeviceDeletedByMods(
- final Collection<ModEntry> currentModifications,
+ final PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer>
currentModifications,
final ITimeIndex currentTimeIndex,
final IDeviceID device)
throws IllegalPathException {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/path/PatternTreeMap.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/path/PatternTreeMap.java
index 8e3bbdf5f9a..2dafb8b99a1 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/path/PatternTreeMap.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/path/PatternTreeMap.java
@@ -68,6 +68,10 @@ public class PatternTreeMap<V, VSerializer extends
PathPatternNode.Serializer<V>
return rootMap.computeIfAbsent(rootName, r -> new PathPatternNode<>(r,
supplier, serializer));
}
+ public boolean isEmpty() {
+ return rootMap.values().stream().allMatch(node -> node.isLeaf() &&
node.getValues().isEmpty());
+ }
+
/**
* Append key and value to PatternTreeMap.
*
