This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch fix-pipe-batch
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/fix-pipe-batch by this push:
new c8a8e236b56 Reapply "Load: Enhanced the table auto-creation logic
(#15265)"
c8a8e236b56 is described below
commit c8a8e236b56b43e1ab1edda49ef23b3b03e8a28b
Author: Steve Yurong Su <[email protected]>
AuthorDate: Thu Apr 24 13:00:53 2025 +0800
Reapply "Load: Enhanced the table auto-creation logic (#15265)"
This reverts commit 6b20d82dea92b078b54c847c9824f19ca27379ad.
---
.../manual/basic/IoTDBPipeWithLoadIT.java | 130 +++++++++++++++++++++
.../plan/analyze/load/LoadTsFileAnalyzer.java | 66 ++---------
.../analyze/load/LoadTsFileTableSchemaCache.java | 78 ++++++++++++-
.../plan/relational/sql/ast/LoadTsFile.java | 10 +-
.../apache/iotdb/db/utils/ModificationUtils.java | 23 ++--
5 files changed, 232 insertions(+), 75 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java
index 36ff0d61811..38177fee0ff 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java
@@ -41,6 +41,7 @@ import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import java.sql.Connection;
+import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;
@@ -380,4 +381,133 @@ public class IoTDBPipeWithLoadIT extends
AbstractPipeTableModelDualManualIT {
handleFailure);
}
}
+
+ @Test
+ public void testLoadAutoCreateWithTableDeletion() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ final Consumer<String> handleFailure =
+ o -> {
+ TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
+ TestUtils.executeNonQueryWithRetry(receiverEnv, "flush");
+ };
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("extractor.realtime.mode", "file");
+ extractorAttributes.put("user", "root");
+
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
+ try (final Connection connection =
senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("create database if not exists db");
+ statement.execute("use db");
+ statement.execute(
+ "create table if not exists t1(tag1 STRING TAG, tag2 STRING TAG,
s1 TEXT FIELD, s2 INT32 FIELD)");
+ statement.execute("INSERT INTO t1(time,tag1,tag2,s1,s2) values(1,
'd1', 'd2', 'red', 1)");
+ statement.execute("INSERT INTO t1(time,tag1,tag2,s1,s2) values(2,
'd1', 'd2', 'blue', 2)");
+ statement.execute("flush");
+ statement.execute("drop table t1");
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
client.startPipe("p1").getCode());
+
+ // Ensure the deleted table won't be created
+ // Now the database will also be created at receiver
+ try (final Connection connection =
receiverEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.executeQuery("describe db.t1");
+ fail();
+ } catch (final SQLException ignore) {
+ // Expected
+ }
+ }
+ }
+
+ @Test
+ public void testLoadAutoCreateWithoutInsertPermission() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ final Consumer<String> handleFailure =
+ o -> {
+ TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
+ TestUtils.executeNonQueryWithRetry(receiverEnv, "flush");
+ };
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("extractor.realtime.mode", "file");
+ extractorAttributes.put("user", "root");
+
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+ connectorAttributes.put("connector.user", "user01");
+ connectorAttributes.put("connector.password", "1234");
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
+ try (final Connection connection =
receiverEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("create user user01 '1234'");
+ statement.execute("grant create on any to user user01");
+ } catch (final Exception e) {
+ fail(e.getMessage());
+ }
+
+ try (final Connection connection =
senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("create database if not exists db");
+ statement.execute("use db");
+ statement.execute(
+ "create table if not exists t1(tag1 STRING TAG, tag2 STRING TAG,
s1 TEXT FIELD, s2 INT32 FIELD)");
+ statement.execute("INSERT INTO t1(time,tag1,tag2,s1,s2) values(1,
'd1', 'd2', 'red', 1)");
+ statement.execute("INSERT INTO t1(time,tag1,tag2,s1,s2) values(2,
'd1', 'd2', 'blue', 2)");
+ statement.execute("flush");
+ } catch (final Exception e) {
+ fail(e.getMessage());
+ }
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
client.startPipe("p1").getCode());
+
+ // Ensure the table without insert privilege won't be created
+ // Now the database will also be created at receiver
+ try (final Connection connection =
receiverEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.executeQuery("describe db.t1");
+ fail();
+ } catch (final SQLException ignore) {
+ // Expected
+ }
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
index 0b69e91360d..fb8afc5ca58 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
@@ -23,7 +23,6 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.utils.RetryUtils;
-import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.load.LoadAnalyzeException;
import org.apache.iotdb.db.exception.load.LoadAnalyzeTypeMismatchException;
@@ -37,16 +36,11 @@ import
org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
-import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
-import
org.apache.iotdb.db.queryengine.plan.execution.config.executor.ClusterConfigTaskExecutor;
-import
org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.CreateDBTask;
import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
-import
org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LoadTsFile;
import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
-import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
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.utils.TsFileResourceUtils;
@@ -56,7 +50,6 @@ import org.apache.iotdb.db.utils.TimestampPrecisionUtils;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
-import com.google.common.util.concurrent.ListenableFuture;
import org.apache.commons.io.FileUtils;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.encrypt.EncryptParameter;
@@ -84,7 +77,6 @@ import java.util.Optional;
import static org.apache.iotdb.commons.utils.FileUtils.copyFileWithMD5Check;
import static org.apache.iotdb.commons.utils.FileUtils.moveFileWithMD5Check;
import static
org.apache.iotdb.db.queryengine.plan.execution.config.TableConfigTaskVisitor.DATABASE_NOT_SPECIFIED;
-import static
org.apache.iotdb.db.queryengine.plan.execution.config.TableConfigTaskVisitor.validateDatabaseName;
import static
org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileCostMetricsSet.ANALYSIS;
import static
org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileCostMetricsSet.ANALYSIS_ASYNC_MOVE;
@@ -522,12 +514,13 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
final Map<IDeviceID, List<TimeseriesMetadata>> device2TimeseriesMetadata
=
timeseriesMetadataIterator.next();
- if (isAutoCreateSchemaOrVerifySchemaEnabled) {
- getOrCreateTreeSchemaVerifier().autoCreateAndVerify(reader,
device2TimeseriesMetadata);
- }
-
if (!tsFileResource.resourceFileExists()) {
TsFileResourceUtils.updateTsFileResource(device2TimeseriesMetadata,
tsFileResource);
+
getOrCreateTableSchemaCache().setCurrentTimeIndex(tsFileResource.getTimeIndex());
+ }
+
+ if (isAutoCreateSchemaOrVerifySchemaEnabled) {
+ getOrCreateTreeSchemaVerifier().autoCreateAndVerify(reader,
device2TimeseriesMetadata);
}
// TODO: how to get the correct write point count when
@@ -572,32 +565,21 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
}
}
- autoCreateTableDatabaseIfAbsent(databaseForTableData);
-
getOrCreateTableSchemaCache().setDatabase(databaseForTableData);
+ getOrCreateTableSchemaCache().setTableSchemaMap(tableSchemaMap);
getOrCreateTableSchemaCache().setCurrentModificationsAndTimeIndex(tsFileResource,
reader);
- for (Map.Entry<String, org.apache.tsfile.file.metadata.TableSchema>
name2Schema :
- tableSchemaMap.entrySet()) {
- final
org.apache.iotdb.db.queryengine.plan.relational.metadata.TableSchema fileSchema
=
- org.apache.iotdb.db.queryengine.plan.relational.metadata.TableSchema
- .fromTsFileTableSchema(name2Schema.getKey(),
name2Schema.getValue());
- getOrCreateTableSchemaCache().createTable(fileSchema, context, metadata);
- accessControl.checkCanInsertIntoTable(
- context.getSession().getUserName(),
- new QualifiedObjectName(databaseForTableData, name2Schema.getKey()));
- }
-
while (timeseriesMetadataIterator.hasNext()) {
final Map<IDeviceID, List<TimeseriesMetadata>> device2TimeseriesMetadata
=
timeseriesMetadataIterator.next();
- for (IDeviceID deviceId : device2TimeseriesMetadata.keySet()) {
- getOrCreateTableSchemaCache().autoCreateAndVerify(deviceId);
- }
-
if (!tsFileResource.resourceFileExists()) {
TsFileResourceUtils.updateTsFileResource(device2TimeseriesMetadata,
tsFileResource);
+
getOrCreateTableSchemaCache().setCurrentTimeIndex(tsFileResource.getTimeIndex());
+ }
+
+ for (IDeviceID deviceId : device2TimeseriesMetadata.keySet()) {
+ getOrCreateTableSchemaCache().autoCreateAndVerify(deviceId);
}
writePointCount += getWritePointCount(device2TimeseriesMetadata);
@@ -639,35 +621,11 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
private LoadTsFileTableSchemaCache getOrCreateTableSchemaCache() {
if (tableSchemaCache == null) {
- tableSchemaCache = new LoadTsFileTableSchemaCache(metadata, context);
+ tableSchemaCache = new LoadTsFileTableSchemaCache(metadata, context,
isAutoCreateDatabase);
}
return tableSchemaCache;
}
- private void autoCreateTableDatabaseIfAbsent(final String database) throws
LoadAnalyzeException {
- validateDatabaseName(database);
- if (DataNodeTableCache.getInstance().isDatabaseExist(database)) {
- return;
- }
-
- accessControl.checkCanCreateDatabase(context.getSession().getUserName(),
database);
- final CreateDBTask task =
- new CreateDBTask(new TDatabaseSchema(database).setIsTableModel(true),
true);
- try {
- final ListenableFuture<ConfigTaskResult> future =
- task.execute(ClusterConfigTaskExecutor.getInstance());
- final ConfigTaskResult result = future.get();
- if (result.getStatusCode().getStatusCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- throw new LoadAnalyzeException(
- String.format(
- "Auto create database failed: %s, status code: %s",
- database, result.getStatusCode()));
- }
- } catch (final Exception e) {
- throw new LoadAnalyzeException("Auto create database failed because: " +
e.getMessage());
- }
- }
-
private void addTsFileResource(TsFileResource tsFileResource) {
if (isTableModelStatement) {
loadTsFileTableStatement.addTsFileResource(tsFileResource);
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 cfca453a3ed..1df3a90ae6f 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
@@ -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.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.load.LoadAnalyzeException;
@@ -28,10 +29,16 @@ import
org.apache.iotdb.db.exception.load.LoadAnalyzeTypeMismatchException;
import org.apache.iotdb.db.exception.load.LoadRuntimeOutOfMemoryException;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.plan.Coordinator;
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import
org.apache.iotdb.db.queryengine.plan.execution.config.executor.ClusterConfigTaskExecutor;
+import
org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.CreateDBTask;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import
org.apache.iotdb.db.queryengine.plan.relational.metadata.ITableDeviceSchemaValidation;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import
org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableSchema;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
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.tsfile.TsFileResource;
@@ -40,7 +47,9 @@ 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.rpc.TSStatusCode;
+import com.google.common.util.concurrent.ListenableFuture;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.utils.Pair;
@@ -62,6 +71,7 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import static
org.apache.iotdb.commons.schema.MemUsageUtil.computeStringMemUsage;
+import static
org.apache.iotdb.db.queryengine.plan.execution.config.TableConfigTaskVisitor.validateDatabaseName;
public class LoadTsFileTableSchemaCache {
private static final Logger LOGGER =
LoggerFactory.getLogger(LoadTsFileTableSchemaCache.class);
@@ -82,6 +92,8 @@ public class LoadTsFileTableSchemaCache {
private final LoadTsFileMemoryBlock block;
private String database;
+ private boolean needToCreateDatabase;
+ private Map<String, org.apache.tsfile.file.metadata.TableSchema>
tableSchemaMap;
private final Metadata metadata;
private final MPPQueryContext context;
@@ -100,7 +112,8 @@ public class LoadTsFileTableSchemaCache {
private int currentBatchDevicesCount = 0;
- public LoadTsFileTableSchemaCache(Metadata metadata, MPPQueryContext context)
+ public LoadTsFileTableSchemaCache(
+ final Metadata metadata, final MPPQueryContext context, final boolean
needToCreateDatabase)
throws LoadRuntimeOutOfMemoryException {
this.block =
LoadTsFileMemoryManager.getInstance()
@@ -109,24 +122,31 @@ public class LoadTsFileTableSchemaCache {
this.context = context;
this.currentBatchTable2Devices = new HashMap<>();
this.currentModifications = new ArrayList<>();
+ this.needToCreateDatabase = needToCreateDatabase;
}
- public void setDatabase(String database) {
+ public void setDatabase(final String database) {
this.database = database;
}
- public void autoCreateAndVerify(IDeviceID device) {
+ public void setTableSchemaMap(
+ final Map<String, org.apache.tsfile.file.metadata.TableSchema>
tableSchemaMap) {
+ this.tableSchemaMap = tableSchemaMap;
+ }
+
+ public void autoCreateAndVerify(final IDeviceID device) throws
LoadAnalyzeException {
try {
if (ModificationUtils.isDeviceDeletedByMods(currentModifications,
currentTimeIndex, device)) {
return;
}
- } catch (IllegalPathException e) {
+ } catch (final IllegalPathException e) {
LOGGER.warn(
"Failed to check if device {} is deleted by mods. Will see it as not
deleted.",
device,
e);
}
+ createTableAndDatabaseIfNecessary(device.getTableName());
// TODO: add permission check and record auth cost
addDevice(device);
if (shouldFlushDevices()) {
@@ -245,8 +265,26 @@ public class LoadTsFileTableSchemaCache {
return Arrays.copyOf(segments, lastNonNullIndex + 1);
}
- public void createTable(TableSchema fileSchema, MPPQueryContext context,
Metadata metadata)
+ public void createTableAndDatabaseIfNecessary(final String tableName)
throws LoadAnalyzeException {
+ final org.apache.tsfile.file.metadata.TableSchema schema =
tableSchemaMap.remove(tableName);
+ if (Objects.isNull(schema)) {
+ return;
+ }
+
+ // Check on creation, do not auto-create tables or database that cannot be
inserted
+ Coordinator.getInstance()
+ .getAccessControl()
+ .checkCanInsertIntoTable(
+ context.getSession().getUserName(), new
QualifiedObjectName(database, tableName));
+
+ if (needToCreateDatabase) {
+ autoCreateTableDatabaseIfAbsent(database);
+ needToCreateDatabase = false;
+ }
+ final org.apache.iotdb.db.queryengine.plan.relational.metadata.TableSchema
fileSchema =
+
org.apache.iotdb.db.queryengine.plan.relational.metadata.TableSchema.fromTsFileTableSchema(
+ tableName, schema);
final TableSchema realSchema =
metadata.validateTableHeaderSchema(database, fileSchema, context,
true, true).orElse(null);
if (Objects.isNull(realSchema)) {
@@ -258,6 +296,32 @@ public class LoadTsFileTableSchemaCache {
verifyTableDataTypeAndGenerateIdColumnMapper(fileSchema, realSchema);
}
+ private void autoCreateTableDatabaseIfAbsent(final String database) throws
LoadAnalyzeException {
+ validateDatabaseName(database);
+ if (DataNodeTableCache.getInstance().isDatabaseExist(database)) {
+ return;
+ }
+
+ Coordinator.getInstance()
+ .getAccessControl()
+ .checkCanCreateDatabase(context.getSession().getUserName(), database);
+ final CreateDBTask task =
+ new CreateDBTask(new TDatabaseSchema(database).setIsTableModel(true),
true);
+ try {
+ final ListenableFuture<ConfigTaskResult> future =
+ task.execute(ClusterConfigTaskExecutor.getInstance());
+ final ConfigTaskResult result = future.get();
+ if (result.getStatusCode().getStatusCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new LoadAnalyzeException(
+ String.format(
+ "Auto create database failed: %s, status code: %s",
+ database, result.getStatusCode()));
+ }
+ } catch (final Exception e) {
+ throw new LoadAnalyzeException("Auto create database failed because: " +
e.getMessage());
+ }
+ }
+
private void verifyTableDataTypeAndGenerateIdColumnMapper(
TableSchema fileSchema, TableSchema realSchema) throws
LoadAnalyzeException {
final int realIdColumnCount = realSchema.getIdColumns().size();
@@ -352,6 +416,10 @@ public class LoadTsFileTableSchemaCache {
}
}
+ public void setCurrentTimeIndex(final ITimeIndex timeIndex) {
+ this.currentTimeIndex = timeIndex;
+ }
+
public void close() {
clearDevices();
clearIdColumnMapper();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/LoadTsFile.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/LoadTsFile.java
index 180056c774e..5bbb8939f91 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/LoadTsFile.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/LoadTsFile.java
@@ -41,11 +41,11 @@ public class LoadTsFile extends Statement {
private int databaseLevel; // For loading to tree-model only
private String database; // For loading to table-model only
- private boolean deleteAfterLoad = false;
- private boolean convertOnTypeMismatch = true;
- private long tabletConversionThresholdBytes = -1;
- private boolean autoCreateDatabase = true;
- private boolean verify = true;
+ private boolean deleteAfterLoad;
+ private boolean convertOnTypeMismatch;
+ private long tabletConversionThresholdBytes;
+ private boolean autoCreateDatabase;
+ private boolean verify;
private boolean isAsyncLoad = false;
private boolean isGeneratedByPipe = false;
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 7bb7445abdd..d5b477e61e4 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
@@ -40,7 +40,6 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
-import java.util.Optional;
public class ModificationUtils {
@@ -404,16 +403,18 @@ public class ModificationUtils {
}
public static boolean isDeviceDeletedByMods(
- Collection<ModEntry> currentModifications, ITimeIndex currentTimeIndex,
IDeviceID device)
+ final Collection<ModEntry> currentModifications,
+ final ITimeIndex currentTimeIndex,
+ final IDeviceID device)
throws IllegalPathException {
- if (currentTimeIndex == null) {
- return false;
- }
- Optional<Long> startTime = currentTimeIndex.getStartTime(device);
- Optional<Long> endTime = currentTimeIndex.getEndTime(device);
- if (startTime.isPresent() && endTime.isPresent()) {
- return isAllDeletedByMods(currentModifications, device, startTime.get(),
endTime.get());
- }
- return false;
+ return isAllDeletedByMods(
+ currentModifications,
+ device,
+ Objects.isNull(currentTimeIndex)
+ ? Long.MIN_VALUE
+ : currentTimeIndex.getStartTime(device).orElse(Long.MIN_VALUE),
+ Objects.isNull(currentTimeIndex)
+ ? Long.MAX_VALUE
+ : currentTimeIndex.getEndTime(device).orElse(Long.MAX_VALUE));
}
}
