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/09 09:13:25 UTC
[iotdb] branch load-tsfile-npe created (now 12cfe355d43)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch load-tsfile-npe
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 12cfe355d43 [IOTDB-6105] Load: NPE when analyzing tsfile
This branch includes the following new commits:
new 12cfe355d43 [IOTDB-6105] Load: NPE when analyzing tsfile
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
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch load-tsfile-npe
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 12cfe355d435d771da4b500e8995f2632340f9f7
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Aug 9 17:13:07 2023 +0800
[IOTDB-6105] Load: NPE when analyzing tsfile
---
.../pipe/connector/legacy/loader/TsFileLoader.java | 2 +-
.../queryengine/plan/analyze/AnalyzeVisitor.java | 397 +-----------------
.../plan/analyze/LoadTsfileAnalyzer.java | 448 +++++++++++++++++++++
.../db/queryengine/plan/parser/ASTVisitor.java | 2 +-
.../plan/statement/crud/LoadTsFileStatement.java | 29 +-
5 files changed, 469 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..9c3d22012ac
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/LoadTsfileAnalyzer.java
@@ -0,0 +1,448 @@
+/*
+ * 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 Set<PartialPath> alreadySetDatabases = new HashSet<>();
+
+ 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);
+ 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()));
+ }
+ }
+
+ // load function will query data partition in scheduler
+ 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()) {
+ final Map<String, Boolean> device2IsAligned = new HashMap<>();
+
+ device2TimeseriesMetadata = reader.getAllTimeseriesMetadata(true);
+ final TimeSeriesIterator timeSeriesIterator =
+ new TimeSeriesIterator(tsFile, device2TimeseriesMetadata);
+ while (timeSeriesIterator.hasNext()) {
+ final Map<String, Set<MeasurementSchema>> device2TimeseriesSchemas = new HashMap<>();
+
+ // load timeseries metadata in batch with limited number
+ for (int timeseriesCount = 0, timeseriesLimit = CONFIG.getMaxLoadingTimeseriesNumber();
+ timeseriesCount < timeseriesLimit && timeSeriesIterator.hasNext();
+ ++timeseriesCount) {
+ 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)) {
+ device2IsAligned.put(device, true);
+ } else {
+ final Pair<CompressionType, TSEncoding> compressionEncodingPair =
+ reader.readTimeseriesCompressionTypeAndEncoding(timeseriesMetadata);
+ device2TimeseriesSchemas
+ .computeIfAbsent(device, o -> new HashSet<>())
+ .add(
+ new MeasurementSchema(
+ timeseriesMetadata.getMeasurementId(),
+ dataType,
+ compressionEncodingPair.getRight(),
+ compressionEncodingPair.getLeft()));
+ device2IsAligned.putIfAbsent(device, false);
+ }
+ }
+
+ autoCreateAndVerifySchema(device2TimeseriesSchemas, device2IsAligned);
+ }
+ }
+
+ // 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 void autoCreateAndVerifySchema(
+ Map<String, Set<MeasurementSchema>> device2TimeseriesSchemas,
+ Map<String, Boolean> device2IsAligned)
+ throws SemanticException {
+ if (device2TimeseriesSchemas.isEmpty()) {
+ return;
+ }
+
+ try {
+ if (loadTsFileStatement.isAutoCreateDatabase()) {
+ autoCreateDatabase(loadTsFileStatement.getDatabaseLevel(), device2TimeseriesSchemas);
+ }
+
+ // schema fetcher will not auto create if config set
+ // isAutoCreateSchemaEnabled is false.
+ final ISchemaTree schemaTree =
+ autoCreateSchema(device2TimeseriesSchemas, device2IsAligned, context);
+
+ if (loadTsFileStatement.isVerifySchema()) {
+ verifySchema(schemaTree, device2TimeseriesSchemas, 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));
+ }
+ }
+
+ private void autoCreateDatabase(
+ int databaseLevel, Map<String, Set<MeasurementSchema>> device2TimeseriesSchemas)
+ throws VerifyMetadataException, LoadFileException, IllegalPathException {
+ final Set<PartialPath> databaseSet = new HashSet<>();
+
+ // e.g. "root.sg" means sgLevel = 1, "root.sg.test" means sgLevel=2
+ databaseLevel += 1;
+ for (final String device : device2TimeseriesSchemas.keySet()) {
+ final PartialPath devicePath = new PartialPath(device);
+
+ final String[] devicePrefixNodes = devicePath.getNodes();
+ if (devicePrefixNodes.length < databaseLevel) {
+ throw new VerifyMetadataException(
+ String.format("Database level %d is longer than device %s.", databaseLevel, device));
+ }
+
+ final String[] databasePrefixNodes = new String[databaseLevel];
+ System.arraycopy(devicePrefixNodes, 0, databasePrefixNodes, 0, databaseLevel);
+
+ 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(
+ Map<String, Set<MeasurementSchema>> device2Schemas,
+ Map<String, Boolean> device2IsAligned,
+ MPPQueryContext context)
+ 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 : device2Schemas.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(device2IsAligned.get(entry.getKey()));
+ }
+
+ return SchemaValidator.validate(
+ schemaFetcher,
+ deviceList,
+ measurementList,
+ dataTypeList,
+ encodingsList,
+ compressionTypesList,
+ isAlignedList,
+ context);
+ }
+
+ private void verifySchema(
+ ISchemaTree schemaTree,
+ Map<String, Set<MeasurementSchema>> tsfileDevice2Schemas,
+ Map<String, Boolean> tsfileDevice2IsAligned)
+ throws VerifyMetadataException, IllegalPathException {
+ for (final Map.Entry<String, Set<MeasurementSchema>> entry : tsfileDevice2Schemas.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()));
+
+ // check device schema: is aligned or not
+ if (iotdbDeviceSchemaInfo.isAligned()
+ != Boolean.TRUE.equals(tsfileDevice2IsAligned.get(device))) {
+ throw new VerifyMetadataException(
+ String.format(
+ "Device %s in TsFile is %s, but in IoTDB is %s.",
+ device,
+ tsfileDevice2IsAligned.get(device) ? "aligned" : "not aligned",
+ iotdbDeviceSchemaInfo.isAligned() ? "aligned" : "not aligned"));
+ }
+
+ // check timeseries schema
+ final List<MeasurementSchema> iotdbTimeseriesSchemas =
+ iotdbDeviceSchemaInfo.getMeasurementSchemaList();
+ for (int i = 0, n = iotdbTimeseriesSchemas.size(); i < n; i++) {
+ final MeasurementSchema iotdbSchema = iotdbTimeseriesSchemas.get(i);
+ final MeasurementSchema tsFileSchema = tsfileTimeseriesSchemas.get(i);
+ final String measurementPath =
+ device + TsFileConstant.PATH_SEPARATOR + iotdbSchema.getMeasurementId();
+
+ // check datatype
+ if (!tsFileSchema.getType().equals(iotdbSchema.getType())) {
+ throw new VerifyMetadataException(
+ String.format(
+ "Measurement %s datatype not match, TsFile: %s, IoTDB: %s",
+ measurementPath, 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: {}",
+ measurementPath,
+ 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: {}",
+ measurementPath,
+ tsFileSchema.getCompressor().name(),
+ iotdbSchema.getCompressor().name());
+ }
+ }
+ }
+ }
+
+ private static 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 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 {
+ LOGGER.info(
+ "Analyzing TsFile {}, all {} timeseries has been returned to analyzer.",
+ tsFile.getAbsolutePath(),
+ returnedTimeseriesCount);
+ return false;
+ }
+ }
+ return timeseriesMetadataIterator.hasNext();
+ }
+
+ @Override
+ public Pair<String, TimeseriesMetadata> next() {
+ if (returnedTimeseriesCount++ % LOG_PRINT_INTERVAL == 0) {
+ LOGGER.info(
+ "Analyzing TsFile {}, {} timeseries has been returned to analyzer.",
+ tsFile.getAbsolutePath(),
+ returnedTimeseriesCount);
+ }
+ return new Pair<>(currentDevice, timeseriesMetadataIterator.next());
+ }
+ }
+}
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 e3b27b0402e..ce9bd6b11f4 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="