You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/08/11 03:58:11 UTC
[iotdb] branch iotdb-6105-1.2 created (now bb00e8effaa)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch iotdb-6105-1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at bb00e8effaa [IOTDB-6105] Load: NPE when analyzing tsfile (#10821)
This branch includes the following new commits:
new bb00e8effaa [IOTDB-6105] Load: NPE when analyzing tsfile (#10821)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: [IOTDB-6105] Load: NPE when analyzing tsfile (#10821)
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch iotdb-6105-1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit bb00e8effaa79a9be2d9e0d6dcb8608366f55771
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Fri Aug 11 00:28:43 2023 +0800
[IOTDB-6105] Load: NPE when analyzing tsfile (#10821)
(cherry picked from commit 2e6d9527c047bd5487af8e868d156dc511ac4578)
---
.../pipe/connector/legacy/loader/TsFileLoader.java | 2 +-
.../queryengine/plan/analyze/AnalyzeVisitor.java | 397 +--------------
.../plan/analyze/LoadTsfileAnalyzer.java | 543 +++++++++++++++++++++
.../db/queryengine/plan/parser/ASTVisitor.java | 2 +-
.../plan/statement/crud/LoadTsFileStatement.java | 29 +-
5 files changed, 564 insertions(+), 409 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/legacy/loader/TsFileLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/legacy/loader/TsFileLoader.java
index 9966f05e48e..05dc5a817ae 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/legacy/loader/TsFileLoader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/legacy/loader/TsFileLoader.java
@@ -53,7 +53,7 @@ public class TsFileLoader implements ILoader {
try {
LoadTsFileStatement statement = new LoadTsFileStatement(tsFile.getAbsolutePath());
statement.setDeleteAfterLoad(true);
- statement.setSgLevel(parseSgLevel());
+ statement.setDatabaseLevel(parseSgLevel());
statement.setVerifySchema(true);
statement.setAutoCreateDatabase(false);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
index 2e345832fee..50fa6b50988 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
@@ -39,8 +39,6 @@ import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
import org.apache.iotdb.confignode.rpc.thrift.TGetDataNodeLocationsResp;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.exception.LoadFileException;
-import org.apache.iotdb.db.exception.VerifyMetadataException;
import org.apache.iotdb.db.exception.metadata.template.TemplateIncompatibleException;
import org.apache.iotdb.db.exception.metadata.view.UnsupportedViewException;
import org.apache.iotdb.db.exception.sql.MeasurementNotExistException;
@@ -49,7 +47,6 @@ import org.apache.iotdb.db.exception.sql.StatementAnalyzeException;
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
-import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
@@ -59,10 +56,8 @@ import org.apache.iotdb.db.queryengine.common.schematree.DeviceSchemaInfo;
import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree;
import org.apache.iotdb.db.queryengine.execution.operator.window.WindowType;
import org.apache.iotdb.db.queryengine.metric.QueryPlanCostMetricSet;
-import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator;
-import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
import org.apache.iotdb.db.queryengine.plan.expression.ExpressionType;
import org.apache.iotdb.db.queryengine.plan.expression.binary.CompareBinaryExpression;
@@ -117,7 +112,6 @@ import org.apache.iotdb.db.queryengine.plan.statement.metadata.CountTimeSeriesSt
import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateAlignedTimeSeriesStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateMultiTimeSeriesStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateTimeSeriesStatement;
-import org.apache.iotdb.db.queryengine.plan.statement.metadata.DatabaseSchemaStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowChildNodesStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowChildPathsStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowClusterStatement;
@@ -140,18 +134,10 @@ import org.apache.iotdb.db.queryengine.plan.statement.sys.ShowQueriesStatement;
import org.apache.iotdb.db.queryengine.plan.statement.sys.ShowVersionStatement;
import org.apache.iotdb.db.schemaengine.SchemaConstant;
import org.apache.iotdb.db.schemaengine.template.Template;
-import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
-import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
-import org.apache.iotdb.db.utils.FileLoaderUtils;
import org.apache.iotdb.db.utils.TimePartitionUtils;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
-import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
-import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
-import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.TimeRange;
import org.apache.iotdb.tsfile.read.filter.GroupByFilter;
import org.apache.iotdb.tsfile.read.filter.GroupByMonthFilter;
@@ -160,14 +146,11 @@ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
-import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.File;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -2490,92 +2473,8 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
@Override
public Analysis visitLoadFile(LoadTsFileStatement loadTsFileStatement, MPPQueryContext context) {
- context.setQueryType(QueryType.WRITE);
-
- int tsfileNum = loadTsFileStatement.getTsFiles().size();
-
- // analyze tsfile metadata
- for (int i = 0; i < tsfileNum; i++) {
- File tsFile = loadTsFileStatement.getTsFiles().get(i);
- if (tsFile.length() == 0) {
- if (logger.isWarnEnabled()) {
- logger.warn(String.format("TsFile %s is empty.", tsFile.getPath()));
- }
- throw new SemanticException(
- String.format(
- "TsFile %s is empty, please check it be flushed to disk correctly.",
- tsFile.getPath()));
- }
- try {
- analyzeTsFile(loadTsFileStatement, tsFile, context);
- } catch (IllegalArgumentException e) {
- logger.warn(
- String.format(
- "Parse file %s to resource error, this TsFile maybe empty.", tsFile.getPath()),
- e);
- throw new SemanticException(
- String.format("TsFile %s is empty or incomplete.", tsFile.getPath()));
- } catch (Exception e) {
- logger.warn(String.format("Parse file %s to resource error.", tsFile.getPath()), e);
- throw new SemanticException(
- String.format("Parse file %s to resource error", tsFile.getPath()));
- }
-
- if (i + 1 == tsfileNum) {
- break;
- }
-
- double progressPercentage = (i + 1) * 100.00 / tsfileNum;
-
- logger.info(
- "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
- i + 1, tsfileNum, String.format("%.2f", progressPercentage));
- }
-
- logger.info(
- "Load - Analysis Stage:{}/{} tsfiles have been analyzed, progress: {}%",
- tsfileNum, tsfileNum, "100.00");
- // load function will query data partition in scheduler
- Analysis analysis = new Analysis();
- analysis.setStatement(loadTsFileStatement);
- return analysis;
- }
-
- private void autoCreateAndVerifySchema(
- LoadTsFileStatement loadTsFileStatement,
- Map<String, Map<MeasurementSchema, File>> device2Schemas,
- Map<String, Pair<Boolean, File>> device2IsAligned,
- MPPQueryContext context)
- throws SemanticException {
- if (device2Schemas.isEmpty()) {
- return;
- }
- try {
- if (loadTsFileStatement.isVerifySchema()) {
- verifyLoadingMeasurements(device2Schemas);
- }
- if (loadTsFileStatement.isAutoCreateDatabase()) {
- autoCreateSg(loadTsFileStatement.getSgLevel(), device2Schemas);
- }
- ISchemaTree schemaTree =
- autoCreateSchema(
- device2Schemas,
- device2IsAligned,
- context); // schema fetcher will not auto create if config set
- // isAutoCreateSchemaEnabled is false.
- if (loadTsFileStatement.isVerifySchema()) {
- verifySchema(schemaTree, device2Schemas, device2IsAligned);
- }
- } catch (Exception e) {
- logger.warn("Auto create or verify schema error.", e);
- throw new SemanticException(
- String.format(
- "Auto create or verify schema error when executing statement %s.",
- loadTsFileStatement));
- } finally {
- device2Schemas.clear();
- device2IsAligned.clear();
- }
+ return new LoadTsfileAnalyzer(loadTsFileStatement, context, partitionFetcher, schemaFetcher)
+ .analyzeFileByFile();
}
/** get analysis according to statement and params */
@@ -2596,298 +2495,6 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
return analysis;
}
- private void analyzeTsFile(LoadTsFileStatement statement, File tsFile, MPPQueryContext context)
- throws IOException, VerifyMetadataException {
- try (TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) {
- Map<String, List<TimeseriesMetadata>> device2Metadata = null;
-
- if (IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()
- || statement.isVerifySchema()) {
- device2Metadata = reader.getAllTimeseriesMetadata(true);
- Map<String, Map<MeasurementSchema, File>> device2Schemas = new HashMap<>();
- Map<String, Pair<Boolean, File>> device2IsAligned = new HashMap<>();
- // construct schema
- int deviceSize = device2Metadata.size();
- int deviceCount = 0;
- int timeseriesCount = 0;
-
- for (Map.Entry<String, List<TimeseriesMetadata>> entry : device2Metadata.entrySet()) {
- String device = entry.getKey();
- deviceCount++;
- List<TimeseriesMetadata> timeseriesMetadataList = entry.getValue();
- boolean isAligned = false;
-
- int timeseriesMetadataListSize = timeseriesMetadataList.size();
-
- for (int timeseriesIndex = 0;
- timeseriesIndex < timeseriesMetadataListSize;
- timeseriesIndex++) {
- TimeseriesMetadata timeseriesMetadata = timeseriesMetadataList.get(timeseriesIndex);
- TSDataType dataType = timeseriesMetadata.getTsDataType();
- if (!dataType.equals(TSDataType.VECTOR)) {
- Pair<CompressionType, TSEncoding> pair =
- reader.readTimeseriesCompressionTypeAndEncoding(timeseriesMetadata);
- MeasurementSchema measurementSchema =
- new MeasurementSchema(
- timeseriesMetadata.getMeasurementId(),
- dataType,
- pair.getRight(),
- pair.getLeft());
- device2Schemas
- .computeIfAbsent(device, o -> new HashMap<>())
- .put(measurementSchema, tsFile);
- timeseriesCount++;
- } else {
- isAligned = true;
- }
-
- // if the number of timeseries exceeds the threshold or loop to the last timeseries of
- // the last device, we should create and verify schema , and clean the device2Schemas
- // and device2IsAligned map.
- if (timeseriesCount > CONFIG.getMaxLoadingTimeseriesNumber()
- || (deviceCount == deviceSize
- && timeseriesIndex == timeseriesMetadataListSize - 1)) {
-
- // check if the device has the same aligned definition in all tsfiles
- if (hasDeviceSameDefinition(device2IsAligned, device, tsFile, isAligned)) {
- autoCreateAndVerifySchema(statement, device2Schemas, device2IsAligned, context);
- timeseriesCount = 0;
- } else {
- throw new VerifyMetadataException(
- String.format(
- "Device %s has different aligned definition in tsFile %s and other TsFile.",
- device, tsFile.getParentFile()));
- }
-
- logger.info(
- "Load - Create and Verify Schemas Stage: the device {} in tsfile {} have been created and verified.",
- entry.getKey(),
- tsFile.getName());
- }
- }
-
- // when the number of devices does not exceed the threshold and it's not the last
- // timeseries of the last device, we also need to check if the device has the same aligned
- // definition in all tsfiles before going to the next device loop.
- if (!hasDeviceSameDefinition(device2IsAligned, device, tsFile, isAligned)) {
- throw new VerifyMetadataException(
- String.format(
- "Device %s has different aligned definition in tsFile %s and other TsFile.",
- device, tsFile.getParentFile()));
- }
- }
- }
-
- // construct TsFileResource
- TsFileResource tsFileResource = new TsFileResource(tsFile);
-
- if (!tsFileResource.resourceFileExists()) {
- if (device2Metadata == null) {
- device2Metadata = reader.getAllTimeseriesMetadata(true);
- }
- FileLoaderUtils.updateTsFileResource(
- device2Metadata, tsFileResource); // serialize it in LoadSingleTsFileNode
- tsFileResource.updatePlanIndexes(reader.getMinPlanIndex());
- tsFileResource.updatePlanIndexes(reader.getMaxPlanIndex());
- } else {
- tsFileResource.deserialize();
- }
-
- tsFileResource.setStatus(TsFileResourceStatus.NORMAL);
- statement.addTsFileResource(tsFileResource);
- }
- }
-
- private boolean hasDeviceSameDefinition(
- Map<String, Pair<Boolean, File>> device2IsAligned,
- String device,
- File tsFile,
- boolean isAligned) {
- return device2IsAligned
- .computeIfAbsent(device, o -> new Pair<>(isAligned, tsFile))
- .left
- .equals(isAligned);
- }
-
- private void autoCreateSg(int sgLevel, Map<String, Map<MeasurementSchema, File>> device2Schemas)
- throws VerifyMetadataException, LoadFileException, IllegalPathException {
- sgLevel += 1; // e.g. "root.sg" means sgLevel = 1, "root.sg.test" means sgLevel=2
- Set<PartialPath> sgSet = new HashSet<>();
- for (String device : device2Schemas.keySet()) {
- PartialPath devicePath = new PartialPath(device);
-
- String[] nodes = devicePath.getNodes();
- String[] sgNodes = new String[sgLevel];
- if (nodes.length < sgLevel) {
- throw new VerifyMetadataException(
- String.format("Sg level %d is longer than device %s.", sgLevel, device));
- }
- System.arraycopy(nodes, 0, sgNodes, 0, sgLevel);
- PartialPath sgPath = new PartialPath(sgNodes);
- sgSet.add(sgPath);
- }
-
- for (PartialPath sgPath : sgSet) {
- DatabaseSchemaStatement statement =
- new DatabaseSchemaStatement(DatabaseSchemaStatement.DatabaseSchemaStatementType.CREATE);
- statement.setDatabasePath(sgPath);
- // load doesn't print exception log
- statement.setEnablePrintExceptionLog(false);
- executeSetDatabaseStatement(statement);
- }
- }
-
- private void executeSetDatabaseStatement(Statement statement) throws LoadFileException {
- long queryId = SessionManager.getInstance().requestQueryId();
- ExecutionResult result =
- Coordinator.getInstance()
- .execute(
- statement,
- queryId,
- null,
- "",
- partitionFetcher,
- schemaFetcher,
- IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold());
- if (result.status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()
- && result.status.code != TSStatusCode.DATABASE_ALREADY_EXISTS.getStatusCode()) {
- logger.warn(
- "Create Database error, statement: {}, result status is: {}", statement, result.status);
- throw new LoadFileException(
- String.format("Can not execute create database statement: %s", statement));
- }
- }
-
- private ISchemaTree autoCreateSchema(
- Map<String, Map<MeasurementSchema, File>> device2Schemas,
- Map<String, Pair<Boolean, File>> device2IsAligned,
- MPPQueryContext context)
- throws IllegalPathException {
- List<PartialPath> deviceList = new ArrayList<>();
- List<String[]> measurementList = new ArrayList<>();
- List<TSDataType[]> dataTypeList = new ArrayList<>();
- List<TSEncoding[]> encodingsList = new ArrayList<>();
- List<CompressionType[]> compressionTypesList = new ArrayList<>();
- List<Boolean> isAlignedList = new ArrayList<>();
-
- for (Map.Entry<String, Map<MeasurementSchema, File>> entry : device2Schemas.entrySet()) {
- int measurementSize = entry.getValue().size();
- String[] measurements = new String[measurementSize];
- TSDataType[] tsDataTypes = new TSDataType[measurementSize];
- TSEncoding[] encodings = new TSEncoding[measurementSize];
- CompressionType[] compressionTypes = new CompressionType[measurementSize];
-
- int index = 0;
- for (MeasurementSchema measurementSchema : entry.getValue().keySet()) {
- measurements[index] = measurementSchema.getMeasurementId();
- tsDataTypes[index] = measurementSchema.getType();
- encodings[index] = measurementSchema.getEncodingType();
- compressionTypes[index++] = measurementSchema.getCompressor();
- }
-
- deviceList.add(new PartialPath(entry.getKey()));
- measurementList.add(measurements);
- dataTypeList.add(tsDataTypes);
- encodingsList.add(encodings);
- compressionTypesList.add(compressionTypes);
- isAlignedList.add(device2IsAligned.get(entry.getKey()).left);
- }
-
- return SchemaValidator.validate(
- schemaFetcher,
- deviceList,
- measurementList,
- dataTypeList,
- encodingsList,
- compressionTypesList,
- isAlignedList,
- context);
- }
-
- private void verifyLoadingMeasurements(Map<String, Map<MeasurementSchema, File>> device2Schemas)
- throws VerifyMetadataException {
- for (Map.Entry<String, Map<MeasurementSchema, File>> deviceEntry : device2Schemas.entrySet()) {
- Map<String, MeasurementSchema> id2Schema = new HashMap<>();
- Map<MeasurementSchema, File> schema2TsFile = deviceEntry.getValue();
- for (Map.Entry<MeasurementSchema, File> entry : schema2TsFile.entrySet()) {
- String measurementId = entry.getKey().getMeasurementId();
- if (!id2Schema.containsKey(measurementId)) {
- id2Schema.put(measurementId, entry.getKey());
- } else {
- MeasurementSchema conflictSchema = id2Schema.get(measurementId);
- String msg =
- String.format(
- "Measurement %s Conflict, TsFile %s has measurement: %s, TsFile %s has measurement %s.",
- deviceEntry.getKey() + measurementId,
- entry.getValue().getPath(),
- entry.getKey(),
- schema2TsFile.get(conflictSchema).getPath(),
- conflictSchema);
- logger.warn(msg);
- throw new VerifyMetadataException(msg);
- }
- }
- }
- }
-
- private void verifySchema(
- ISchemaTree schemaTree,
- Map<String, Map<MeasurementSchema, File>> device2Schemas,
- Map<String, Pair<Boolean, File>> device2IsAligned)
- throws VerifyMetadataException, IllegalPathException {
- for (Map.Entry<String, Map<MeasurementSchema, File>> entry : device2Schemas.entrySet()) {
- String device = entry.getKey();
- MeasurementSchema[] tsFileSchemas =
- entry.getValue().keySet().toArray(new MeasurementSchema[0]);
- DeviceSchemaInfo schemaInfo =
- schemaTree.searchDeviceSchemaInfo(
- new PartialPath(device),
- Arrays.stream(tsFileSchemas)
- .map(MeasurementSchema::getMeasurementId)
- .collect(Collectors.toList()));
- if (schemaInfo.isAligned() != Boolean.TRUE.equals(device2IsAligned.get(device).left)) {
- throw new VerifyMetadataException(
- device,
- "Is aligned",
- device2IsAligned.get(device).left.toString(),
- device2IsAligned.get(device).right.getPath(),
- String.valueOf(schemaInfo.isAligned()));
- }
- List<MeasurementSchema> originSchemaList = schemaInfo.getMeasurementSchemaList();
- int measurementSize = originSchemaList.size();
- for (int j = 0; j < measurementSize; j++) {
- MeasurementSchema originSchema = originSchemaList.get(j);
- MeasurementSchema tsFileSchema = tsFileSchemas[j];
- String measurementPath =
- device + TsFileConstant.PATH_SEPARATOR + originSchema.getMeasurementId();
- if (!tsFileSchema.getType().equals(originSchema.getType())) {
- throw new VerifyMetadataException(
- measurementPath,
- "Datatype",
- tsFileSchema.getType().name(),
- entry.getValue().get(tsFileSchema).getPath(),
- originSchema.getType().name());
- }
- if (!tsFileSchema.getEncodingType().equals(originSchema.getEncodingType())) {
- logger.warn(
- "Encoding type not match, measurement: {}, TsFile: {}, TsFile encoding: {}, origin encoding: {}",
- measurementPath,
- entry.getValue().get(tsFileSchema).getPath(),
- tsFileSchema.getEncodingType().name(),
- originSchema.getEncodingType().name());
- }
- if (!tsFileSchema.getCompressor().equals(originSchema.getCompressor())) {
- logger.warn(
- "Compression type not match, measurement: {}, TsFile: {}, TsFile compression: {}, origin compression: {}",
- measurementPath,
- entry.getValue().get(tsFileSchema).getPath(),
- tsFileSchema.getCompressor().name(),
- originSchema.getCompressor().name());
- }
- }
- }
- }
-
@Override
public Analysis visitShowTimeSeries(
ShowTimeSeriesStatement showTimeSeriesStatement, MPPQueryContext context) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/LoadTsfileAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/LoadTsfileAnalyzer.java
new file mode 100644
index 00000000000..13bd528e56b
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/LoadTsfileAnalyzer.java
@@ -0,0 +1,543 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.analyze;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.LoadFileException;
+import org.apache.iotdb.db.exception.VerifyMetadataException;
+import org.apache.iotdb.db.exception.sql.SemanticException;
+import org.apache.iotdb.db.protocol.session.SessionManager;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.schematree.DeviceSchemaInfo;
+import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree;
+import org.apache.iotdb.db.queryengine.plan.Coordinator;
+import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
+import org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator;
+import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
+import org.apache.iotdb.db.queryengine.plan.statement.Statement;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.DatabaseSchemaStatement;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class LoadTsfileAnalyzer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(LoadTsfileAnalyzer.class);
+
+ private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
+
+ private final LoadTsFileStatement loadTsFileStatement;
+ private final MPPQueryContext context;
+
+ private final IPartitionFetcher partitionFetcher;
+ private final ISchemaFetcher schemaFetcher;
+
+ private final SchemaAutoCreatorAndVerifier schemaAutoCreatorAndVerifier =
+ new SchemaAutoCreatorAndVerifier();
+
+ LoadTsfileAnalyzer(
+ LoadTsFileStatement loadTsFileStatement,
+ MPPQueryContext context,
+ IPartitionFetcher partitionFetcher,
+ ISchemaFetcher schemaFetcher) {
+ this.loadTsFileStatement = loadTsFileStatement;
+ this.context = context;
+
+ this.partitionFetcher = partitionFetcher;
+ this.schemaFetcher = schemaFetcher;
+ }
+
+ public Analysis analyzeFileByFile() {
+ context.setQueryType(QueryType.WRITE);
+
+ // analyze tsfile metadata file by file
+ for (int i = 0, tsfileNum = loadTsFileStatement.getTsFiles().size(); i < tsfileNum; i++) {
+ final File tsFile = loadTsFileStatement.getTsFiles().get(i);
+
+ if (tsFile.length() == 0) {
+ if (LOGGER.isWarnEnabled()) {
+ LOGGER.warn(String.format("TsFile %s is empty.", tsFile.getPath()));
+ }
+ throw new SemanticException(
+ String.format(
+ "TsFile %s is empty, please check it be flushed to disk correctly.",
+ tsFile.getPath()));
+ }
+
+ try {
+ analyzeSingleTsFile(tsFile);
+ if (LOGGER.isInfoEnabled()) {
+ LOGGER.info(
+ "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
+ i + 1, tsfileNum, String.format("%.3f", (i + 1) * 100.00 / tsfileNum));
+ }
+ } catch (IllegalArgumentException e) {
+ LOGGER.warn(
+ String.format(
+ "Parse file %s to resource error, this TsFile maybe empty.", tsFile.getPath()),
+ e);
+ throw new SemanticException(
+ String.format("TsFile %s is empty or incomplete.", tsFile.getPath()));
+ } catch (Exception e) {
+ LOGGER.warn(String.format("Parse file %s to resource error.", tsFile.getPath()), e);
+ throw new SemanticException(
+ String.format("Parse file %s to resource error", tsFile.getPath()));
+ }
+ }
+
+ schemaAutoCreatorAndVerifier.flush();
+ LOGGER.info("Load - Analysis Stage: all tsfiles have been analyzed.");
+
+ // data partition will be queried in the scheduler
+ final Analysis analysis = new Analysis();
+ analysis.setStatement(loadTsFileStatement);
+ return analysis;
+ }
+
+ private void analyzeSingleTsFile(File tsFile) throws IOException {
+ try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) {
+ // can be reused when constructing tsfile resource
+ Map<String, List<TimeseriesMetadata>> device2TimeseriesMetadata = null;
+
+ // auto create or verify schema
+ if (IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()
+ || loadTsFileStatement.isVerifySchema()) {
+ // cache timeseries metadata for the next step
+ device2TimeseriesMetadata = reader.getAllTimeseriesMetadata(true);
+
+ final TimeSeriesIterator timeSeriesIterator =
+ new TimeSeriesIterator(tsFile, device2TimeseriesMetadata);
+ while (timeSeriesIterator.hasNext()) {
+ schemaAutoCreatorAndVerifier.autoCreateAndVerify(reader, timeSeriesIterator);
+ }
+
+ schemaAutoCreatorAndVerifier.flushAndClearDeviceIsAlignedCacheIfNecessary();
+ }
+
+ // construct tsfile resource
+ final TsFileResource tsFileResource = new TsFileResource(tsFile);
+ if (!tsFileResource.resourceFileExists()) {
+ if (device2TimeseriesMetadata == null) {
+ device2TimeseriesMetadata = reader.getAllTimeseriesMetadata(true);
+ }
+ // it will be serialized in LoadSingleTsFileNode
+ FileLoaderUtils.updateTsFileResource(device2TimeseriesMetadata, tsFileResource);
+ tsFileResource.updatePlanIndexes(reader.getMinPlanIndex());
+ tsFileResource.updatePlanIndexes(reader.getMaxPlanIndex());
+ } else {
+ tsFileResource.deserialize();
+ }
+ tsFileResource.setStatus(TsFileResourceStatus.NORMAL);
+ loadTsFileStatement.addTsFileResource(tsFileResource);
+ }
+ }
+
+ private static final class TimeSeriesIterator
+ implements Iterator<Pair<String, TimeseriesMetadata>> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TimeSeriesIterator.class);
+
+ private static final long LOG_PRINT_INTERVAL = 10000;
+ private long returnedTimeseriesCount = 0;
+ private boolean lastLogHasPrinted = false;
+
+ private final File tsFile;
+ private final Iterator<Map.Entry<String, List<TimeseriesMetadata>>>
+ device2TimeseriesMetadataIterator;
+
+ private String currentDevice;
+ private Iterator<TimeseriesMetadata> timeseriesMetadataIterator;
+
+ public TimeSeriesIterator(
+ File tsFile, Map<String, List<TimeseriesMetadata>> device2TimeseriesMetadata) {
+ this.tsFile = tsFile;
+ this.device2TimeseriesMetadataIterator = device2TimeseriesMetadata.entrySet().iterator();
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (timeseriesMetadataIterator == null || !timeseriesMetadataIterator.hasNext()) {
+ if (device2TimeseriesMetadataIterator.hasNext()) {
+ final Map.Entry<String, List<TimeseriesMetadata>> next =
+ device2TimeseriesMetadataIterator.next();
+ currentDevice = next.getKey();
+ timeseriesMetadataIterator = next.getValue().iterator();
+ } else {
+ if (!lastLogHasPrinted) {
+ LOGGER.info(
+ "Analyzing TsFile {}, all {} timeseries has been returned to analyzer.",
+ tsFile.getAbsolutePath(),
+ returnedTimeseriesCount);
+ lastLogHasPrinted = true;
+ }
+ return false;
+ }
+ }
+ return timeseriesMetadataIterator.hasNext();
+ }
+
+ @Override
+ public Pair<String, TimeseriesMetadata> next() {
+ if (returnedTimeseriesCount == 0) {
+ LOGGER.info(
+ "Analyzing TsFile {}, start to return timeseries to analyzer.",
+ tsFile.getAbsolutePath());
+ } else if (returnedTimeseriesCount % LOG_PRINT_INTERVAL == 0) {
+ LOGGER.info(
+ "Analyzing TsFile {}, until now {} timeseries has been returned to analyzer.",
+ tsFile.getAbsolutePath(),
+ returnedTimeseriesCount);
+ }
+ returnedTimeseriesCount++;
+
+ return new Pair<>(currentDevice, timeseriesMetadataIterator.next());
+ }
+ }
+
+ private final class SchemaAutoCreatorAndVerifier {
+
+ private final Map<String, Boolean> tsfileDevice2IsAligned = new HashMap<>();
+ private final Set<PartialPath> alreadySetDatabases = new HashSet<>();
+
+ private final int maxTimeseriesNumberPerBatch = CONFIG.getMaxLoadingTimeseriesNumber();
+ private final Map<String, Set<MeasurementSchema>> currentBatchDevice2TimeseriesSchemas =
+ new HashMap<>();
+ private int currentBatchTimeseriesCount = 0;
+
+ private SchemaAutoCreatorAndVerifier() {}
+
+ public void autoCreateAndVerify(
+ TsFileSequenceReader reader, TimeSeriesIterator timeSeriesIterator) throws IOException {
+ while (currentBatchTimeseriesCount < maxTimeseriesNumberPerBatch
+ && timeSeriesIterator.hasNext()) {
+ final Pair<String, TimeseriesMetadata> pair = timeSeriesIterator.next();
+ final String device = pair.left;
+ final TimeseriesMetadata timeseriesMetadata = pair.right;
+
+ final TSDataType dataType = timeseriesMetadata.getTsDataType();
+ if (dataType.equals(TSDataType.VECTOR)) {
+ tsfileDevice2IsAligned.put(device, true);
+
+ // not a timeseries, skip ++currentBatchTimeseriesCount
+ } else {
+ final Pair<CompressionType, TSEncoding> compressionEncodingPair =
+ reader.readTimeseriesCompressionTypeAndEncoding(timeseriesMetadata);
+ currentBatchDevice2TimeseriesSchemas
+ .computeIfAbsent(device, o -> new HashSet<>())
+ .add(
+ new MeasurementSchema(
+ timeseriesMetadata.getMeasurementId(),
+ dataType,
+ compressionEncodingPair.getRight(),
+ compressionEncodingPair.getLeft()));
+
+ tsfileDevice2IsAligned.putIfAbsent(device, false);
+
+ ++currentBatchTimeseriesCount;
+ }
+ }
+
+ if (currentBatchTimeseriesCount == maxTimeseriesNumberPerBatch) {
+ flush();
+ }
+ }
+
+ /**
+ * This can only be invoked after all timeseries in the current tsfile have been processed.
+ * Otherwise, the isAligned status may be wrong.
+ */
+ public void flushAndClearDeviceIsAlignedCacheIfNecessary() throws SemanticException {
+ // avoid OOM when loading a tsfile with too many timeseries
+ // or loading too many tsfiles at the same time
+ if (tsfileDevice2IsAligned.size() > 10000) {
+ flush();
+ tsfileDevice2IsAligned.clear();
+ }
+ }
+
+ public void flush() {
+ doAutoCreateAndVerify();
+
+ currentBatchDevice2TimeseriesSchemas.clear();
+ currentBatchTimeseriesCount = 0;
+ }
+
+ private void doAutoCreateAndVerify() throws SemanticException {
+ if (currentBatchDevice2TimeseriesSchemas.isEmpty()) {
+ return;
+ }
+
+ try {
+ if (loadTsFileStatement.isVerifySchema()) {
+ makeSureNoDuplicatedMeasurementsInDevices();
+ }
+
+ if (loadTsFileStatement.isAutoCreateDatabase()) {
+ autoCreateDatabase();
+ }
+
+ // schema fetcher will not auto create if config set
+ // isAutoCreateSchemaEnabled is false.
+ final ISchemaTree schemaTree = autoCreateSchema();
+
+ if (loadTsFileStatement.isVerifySchema()) {
+ verifySchema(schemaTree);
+ }
+ } catch (Exception e) {
+ LOGGER.warn("Auto create or verify schema error.", e);
+ throw new SemanticException(
+ String.format(
+ "Auto create or verify schema error when executing statement %s.",
+ loadTsFileStatement));
+ }
+ }
+
+ private void makeSureNoDuplicatedMeasurementsInDevices() throws VerifyMetadataException {
+ for (final Map.Entry<String, Set<MeasurementSchema>> entry :
+ currentBatchDevice2TimeseriesSchemas.entrySet()) {
+ final String device = entry.getKey();
+ final Map<String, MeasurementSchema> measurement2Schema = new HashMap<>();
+ for (final MeasurementSchema timeseriesSchema : entry.getValue()) {
+ final String measurement = timeseriesSchema.getMeasurementId();
+ if (measurement2Schema.containsKey(measurement)) {
+ throw new VerifyMetadataException(
+ String.format("Duplicated measurements %s in device %s.", measurement, device));
+ }
+ measurement2Schema.put(measurement, timeseriesSchema);
+ }
+ }
+ }
+
+ private void autoCreateDatabase()
+ throws VerifyMetadataException, LoadFileException, IllegalPathException {
+ final int databasePrefixNodesLength = loadTsFileStatement.getDatabaseLevel() + 1;
+ final Set<PartialPath> databaseSet = new HashSet<>();
+
+ for (final String device : currentBatchDevice2TimeseriesSchemas.keySet()) {
+ final PartialPath devicePath = new PartialPath(device);
+
+ final String[] devicePrefixNodes = devicePath.getNodes();
+ if (devicePrefixNodes.length < databasePrefixNodesLength) {
+ throw new VerifyMetadataException(
+ String.format(
+ "Database level %d is longer than device %s.",
+ databasePrefixNodesLength, device));
+ }
+
+ final String[] databasePrefixNodes = new String[databasePrefixNodesLength];
+ System.arraycopy(devicePrefixNodes, 0, databasePrefixNodes, 0, databasePrefixNodesLength);
+
+ databaseSet.add(new PartialPath(databasePrefixNodes));
+ }
+
+ databaseSet.removeAll(alreadySetDatabases);
+ for (final PartialPath databasePath : databaseSet) {
+ final DatabaseSchemaStatement statement =
+ new DatabaseSchemaStatement(DatabaseSchemaStatement.DatabaseSchemaStatementType.CREATE);
+ statement.setDatabasePath(databasePath);
+ // do not print exception log because it is not an error
+ statement.setEnablePrintExceptionLog(false);
+ executeSetDatabaseStatement(statement);
+ }
+ alreadySetDatabases.addAll(databaseSet);
+ }
+
+ private void executeSetDatabaseStatement(Statement statement) throws LoadFileException {
+ final long queryId = SessionManager.getInstance().requestQueryId();
+ final ExecutionResult result =
+ Coordinator.getInstance()
+ .execute(
+ statement,
+ queryId,
+ null,
+ "",
+ partitionFetcher,
+ schemaFetcher,
+ IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold());
+ if (result.status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ && result.status.code != TSStatusCode.DATABASE_ALREADY_EXISTS.getStatusCode()) {
+ LOGGER.warn(
+ "Create database error, statement: {}, result status is: {}", statement, result.status);
+ throw new LoadFileException(
+ String.format(
+ "Create database error, statement: %s, result status is: %s",
+ statement, result.status));
+ }
+ }
+
+ private ISchemaTree autoCreateSchema() throws IllegalPathException {
+ final List<PartialPath> deviceList = new ArrayList<>();
+ final List<String[]> measurementList = new ArrayList<>();
+ final List<TSDataType[]> dataTypeList = new ArrayList<>();
+ final List<TSEncoding[]> encodingsList = new ArrayList<>();
+ final List<CompressionType[]> compressionTypesList = new ArrayList<>();
+ final List<Boolean> isAlignedList = new ArrayList<>();
+
+ for (final Map.Entry<String, Set<MeasurementSchema>> entry :
+ currentBatchDevice2TimeseriesSchemas.entrySet()) {
+ final int measurementSize = entry.getValue().size();
+ final String[] measurements = new String[measurementSize];
+ final TSDataType[] tsDataTypes = new TSDataType[measurementSize];
+ final TSEncoding[] encodings = new TSEncoding[measurementSize];
+ final CompressionType[] compressionTypes = new CompressionType[measurementSize];
+
+ int index = 0;
+ for (final MeasurementSchema measurementSchema : entry.getValue()) {
+ measurements[index] = measurementSchema.getMeasurementId();
+ tsDataTypes[index] = measurementSchema.getType();
+ encodings[index] = measurementSchema.getEncodingType();
+ compressionTypes[index++] = measurementSchema.getCompressor();
+ }
+
+ deviceList.add(new PartialPath(entry.getKey()));
+ measurementList.add(measurements);
+ dataTypeList.add(tsDataTypes);
+ encodingsList.add(encodings);
+ compressionTypesList.add(compressionTypes);
+ isAlignedList.add(tsfileDevice2IsAligned.get(entry.getKey()));
+ }
+
+ return SchemaValidator.validate(
+ schemaFetcher,
+ deviceList,
+ measurementList,
+ dataTypeList,
+ encodingsList,
+ compressionTypesList,
+ isAlignedList,
+ context);
+ }
+
+ private void verifySchema(ISchemaTree schemaTree)
+ throws VerifyMetadataException, IllegalPathException {
+ for (final Map.Entry<String, Set<MeasurementSchema>> entry :
+ currentBatchDevice2TimeseriesSchemas.entrySet()) {
+ final String device = entry.getKey();
+ final List<MeasurementSchema> tsfileTimeseriesSchemas = new ArrayList<>(entry.getValue());
+ final DeviceSchemaInfo iotdbDeviceSchemaInfo =
+ schemaTree.searchDeviceSchemaInfo(
+ new PartialPath(device),
+ tsfileTimeseriesSchemas.stream()
+ .map(MeasurementSchema::getMeasurementId)
+ .collect(Collectors.toList()));
+
+ if (iotdbDeviceSchemaInfo == null) {
+ throw new VerifyMetadataException(
+ String.format(
+ "Device %s does not exist in IoTDB and can not be created. "
+ + "Please check weather auto-create-schema is enabled.",
+ device));
+ }
+
+ // check device schema: is aligned or not
+ final boolean isAlignedInTsFile = tsfileDevice2IsAligned.get(device);
+ final boolean isAlignedInIoTDB = iotdbDeviceSchemaInfo.isAligned();
+ if (isAlignedInTsFile != isAlignedInIoTDB) {
+ throw new VerifyMetadataException(
+ String.format(
+ "Device %s in TsFile is %s, but in IoTDB is %s.",
+ device,
+ isAlignedInTsFile ? "aligned" : "not aligned",
+ isAlignedInIoTDB ? "aligned" : "not aligned"));
+ }
+
+ // check timeseries schema
+ final List<MeasurementSchema> iotdbTimeseriesSchemas =
+ iotdbDeviceSchemaInfo.getMeasurementSchemaList();
+ for (int i = 0, n = iotdbTimeseriesSchemas.size(); i < n; i++) {
+ final MeasurementSchema tsFileSchema = tsfileTimeseriesSchemas.get(i);
+ final MeasurementSchema iotdbSchema = iotdbTimeseriesSchemas.get(i);
+ if (iotdbSchema == null) {
+ throw new VerifyMetadataException(
+ String.format(
+ "Measurement %s does not exist in IoTDB and can not be created. "
+ + "Please check weather auto-create-schema is enabled.",
+ device + TsFileConstant.PATH_SEPARATOR + tsfileTimeseriesSchemas.get(i)));
+ }
+
+ // check datatype
+ if (!tsFileSchema.getType().equals(iotdbSchema.getType())) {
+ throw new VerifyMetadataException(
+ String.format(
+ "Measurement %s%s%s datatype not match, TsFile: %s, IoTDB: %s",
+ device,
+ TsFileConstant.PATH_SEPARATOR,
+ iotdbSchema.getMeasurementId(),
+ tsFileSchema.getType(),
+ iotdbSchema.getType()));
+ }
+
+ // check encoding
+ if (!tsFileSchema.getEncodingType().equals(iotdbSchema.getEncodingType())) {
+ // we allow a measurement to have different encodings in different chunks
+ LOGGER.warn(
+ "Encoding type not match, measurement: {}{}{}, "
+ + "TsFile encoding: {}, IoTDB encoding: {}",
+ device,
+ TsFileConstant.PATH_SEPARATOR,
+ iotdbSchema.getMeasurementId(),
+ tsFileSchema.getEncodingType().name(),
+ iotdbSchema.getEncodingType().name());
+ }
+
+ // check compressor
+ if (!tsFileSchema.getCompressor().equals(iotdbSchema.getCompressor())) {
+ // we allow a measurement to have different compressors in different chunks
+ LOGGER.warn(
+ "Compressor not match, measurement: {}{}{}, "
+ + "TsFile compressor: {}, IoTDB compressor: {}",
+ device,
+ TsFileConstant.PATH_SEPARATOR,
+ iotdbSchema.getMeasurementId(),
+ tsFileSchema.getCompressor().name(),
+ iotdbSchema.getCompressor().name());
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
index 1f49e916428..b57af1905f3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
@@ -1887,7 +1887,7 @@ public class ASTVisitor extends IoTDBSqlParserBaseVisitor<Statement> {
if (ctx.ONSUCCESS() != null) {
loadTsFileStatement.setDeleteAfterLoad(ctx.DELETE() != null);
} else if (ctx.SGLEVEL() != null) {
- loadTsFileStatement.setSgLevel(Integer.parseInt(ctx.INTEGER_LITERAL().getText()));
+ loadTsFileStatement.setDatabaseLevel(Integer.parseInt(ctx.INTEGER_LITERAL().getText()));
} else if (ctx.VERIFY() != null) {
loadTsFileStatement.setVerifySchema(Boolean.parseBoolean(ctx.boolean_literal().getText()));
} else {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java
index b4e4f83250a..381d367afe9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java
@@ -35,18 +35,19 @@ import java.util.Collections;
import java.util.List;
public class LoadTsFileStatement extends Statement {
- private File file;
- private int sgLevel;
+
+ private final File file;
+ private int databaseLevel;
private boolean verifySchema;
private boolean deleteAfterLoad;
private boolean autoCreateDatabase;
- private List<File> tsFiles;
- private List<TsFileResource> resources;
+ private final List<File> tsFiles;
+ private final List<TsFileResource> resources;
public LoadTsFileStatement(String filePath) throws FileNotFoundException {
this.file = new File(filePath);
- this.sgLevel = IoTDBDescriptor.getInstance().getConfig().getDefaultStorageGroupLevel();
+ this.databaseLevel = IoTDBDescriptor.getInstance().getConfig().getDefaultStorageGroupLevel();
this.verifySchema = true;
this.deleteAfterLoad = true;
this.autoCreateDatabase = IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled();
@@ -69,7 +70,11 @@ public class LoadTsFileStatement extends Statement {
}
private void findAllTsFile(File file) {
- for (File nowFile : file.listFiles()) {
+ final File[] files = file.listFiles();
+ if (files == null) {
+ return;
+ }
+ for (File nowFile : files) {
if (nowFile.getName().endsWith(TsFileConstant.TSFILE_SUFFIX)) {
tsFiles.add(nowFile);
} else if (nowFile.isDirectory()) {
@@ -95,8 +100,8 @@ public class LoadTsFileStatement extends Statement {
this.deleteAfterLoad = deleteAfterLoad;
}
- public void setSgLevel(int sgLevel) {
- this.sgLevel = sgLevel;
+ public void setDatabaseLevel(int databaseLevel) {
+ this.databaseLevel = databaseLevel;
}
public void setVerifySchema(boolean verifySchema) {
@@ -119,8 +124,8 @@ public class LoadTsFileStatement extends Statement {
return autoCreateDatabase;
}
- public int getSgLevel() {
- return sgLevel;
+ public int getDatabaseLevel() {
+ return databaseLevel;
}
public List<File> getTsFiles() {
@@ -152,8 +157,8 @@ public class LoadTsFileStatement extends Statement {
+ file
+ ", deleteAfterLoad="
+ deleteAfterLoad
- + ", sgLevel="
- + sgLevel
+ + ", databaseLevel="
+ + databaseLevel
+ ", verifySchema="
+ verifySchema
+ ", tsFiles Size="