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
commit 6b20d82dea92b078b54c847c9824f19ca27379ad Author: Steve Yurong Su <[email protected]> AuthorDate: Thu Apr 24 12:34:18 2025 +0800 Revert "Load: Enhanced the table auto-creation logic (#15265)" This reverts commit 8eb73e0cf1f1514b83925a21c256019c21500970. --- .../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, 75 insertions(+), 232 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 38177fee0ff..36ff0d61811 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,7 +41,6 @@ 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; @@ -381,133 +380,4 @@ 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 fb8afc5ca58..0b69e91360d 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,6 +23,7 @@ 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; @@ -36,11 +37,16 @@ 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; @@ -50,6 +56,7 @@ 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; @@ -77,6 +84,7 @@ 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; @@ -514,15 +522,14 @@ public class LoadTsFileAnalyzer implements AutoCloseable { final Map<IDeviceID, List<TimeseriesMetadata>> device2TimeseriesMetadata = timeseriesMetadataIterator.next(); - if (!tsFileResource.resourceFileExists()) { - TsFileResourceUtils.updateTsFileResource(device2TimeseriesMetadata, tsFileResource); - getOrCreateTableSchemaCache().setCurrentTimeIndex(tsFileResource.getTimeIndex()); - } - if (isAutoCreateSchemaOrVerifySchemaEnabled) { getOrCreateTreeSchemaVerifier().autoCreateAndVerify(reader, device2TimeseriesMetadata); } + if (!tsFileResource.resourceFileExists()) { + TsFileResourceUtils.updateTsFileResource(device2TimeseriesMetadata, tsFileResource); + } + // TODO: how to get the correct write point count when // !isAutoCreateSchemaOrVerifySchemaEnabled writePointCount += getWritePointCount(device2TimeseriesMetadata); @@ -565,23 +572,34 @@ 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(); - if (!tsFileResource.resourceFileExists()) { - TsFileResourceUtils.updateTsFileResource(device2TimeseriesMetadata, tsFileResource); - getOrCreateTableSchemaCache().setCurrentTimeIndex(tsFileResource.getTimeIndex()); - } - for (IDeviceID deviceId : device2TimeseriesMetadata.keySet()) { getOrCreateTableSchemaCache().autoCreateAndVerify(deviceId); } + if (!tsFileResource.resourceFileExists()) { + TsFileResourceUtils.updateTsFileResource(device2TimeseriesMetadata, tsFileResource); + } + writePointCount += getWritePointCount(device2TimeseriesMetadata); } @@ -621,11 +639,35 @@ public class LoadTsFileAnalyzer implements AutoCloseable { private LoadTsFileTableSchemaCache getOrCreateTableSchemaCache() { if (tableSchemaCache == null) { - tableSchemaCache = new LoadTsFileTableSchemaCache(metadata, context, isAutoCreateDatabase); + tableSchemaCache = new LoadTsFileTableSchemaCache(metadata, context); } 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 1df3a90ae6f..cfca453a3ed 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,7 +21,6 @@ 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; @@ -29,16 +28,10 @@ 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; @@ -47,9 +40,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.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; @@ -71,7 +62,6 @@ 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); @@ -92,8 +82,6 @@ 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; @@ -112,8 +100,7 @@ public class LoadTsFileTableSchemaCache { private int currentBatchDevicesCount = 0; - public LoadTsFileTableSchemaCache( - final Metadata metadata, final MPPQueryContext context, final boolean needToCreateDatabase) + public LoadTsFileTableSchemaCache(Metadata metadata, MPPQueryContext context) throws LoadRuntimeOutOfMemoryException { this.block = LoadTsFileMemoryManager.getInstance() @@ -122,31 +109,24 @@ public class LoadTsFileTableSchemaCache { this.context = context; this.currentBatchTable2Devices = new HashMap<>(); this.currentModifications = new ArrayList<>(); - this.needToCreateDatabase = needToCreateDatabase; } - public void setDatabase(final String database) { + public void setDatabase(String database) { this.database = database; } - public void setTableSchemaMap( - final Map<String, org.apache.tsfile.file.metadata.TableSchema> tableSchemaMap) { - this.tableSchemaMap = tableSchemaMap; - } - - public void autoCreateAndVerify(final IDeviceID device) throws LoadAnalyzeException { + public void autoCreateAndVerify(IDeviceID device) { try { if (ModificationUtils.isDeviceDeletedByMods(currentModifications, currentTimeIndex, device)) { return; } - } catch (final IllegalPathException e) { + } catch (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()) { @@ -265,26 +245,8 @@ public class LoadTsFileTableSchemaCache { return Arrays.copyOf(segments, lastNonNullIndex + 1); } - public void createTableAndDatabaseIfNecessary(final String tableName) + public void createTable(TableSchema fileSchema, MPPQueryContext context, Metadata metadata) 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)) { @@ -296,32 +258,6 @@ 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(); @@ -416,10 +352,6 @@ 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 5bbb8939f91..180056c774e 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; - private boolean convertOnTypeMismatch; - private long tabletConversionThresholdBytes; - private boolean autoCreateDatabase; - private boolean verify; + private boolean deleteAfterLoad = false; + private boolean convertOnTypeMismatch = true; + private long tabletConversionThresholdBytes = -1; + private boolean autoCreateDatabase = true; + private boolean verify = true; 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 d5b477e61e4..7bb7445abdd 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,6 +40,7 @@ import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Objects; +import java.util.Optional; public class ModificationUtils { @@ -403,18 +404,16 @@ public class ModificationUtils { } public static boolean isDeviceDeletedByMods( - final Collection<ModEntry> currentModifications, - final ITimeIndex currentTimeIndex, - final IDeviceID device) + Collection<ModEntry> currentModifications, ITimeIndex currentTimeIndex, IDeviceID device) throws IllegalPathException { - 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)); + 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; } }
