Skip to content

Commit 91c5132

Browse files
CaideyipiCaideyipi
authored
[To dev/1.3] Load: Optimized the exception error message of convertion to tablet (apache#16499)
1 parent 8ec5aac commit 91c5132

10 files changed

+133
-65
lines changed

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -563,7 +563,7 @@ private void doAnalyzeSingleTableFile(
563563
}
564564

565565
getOrCreateTableSchemaCache().flush();
566-
getOrCreateTableSchemaCache().clearIdColumnMapper();
566+
getOrCreateTableSchemaCache().clearTagColumnMapper();
567567

568568
TimestampPrecisionUtils.checkTimestampPrecision(tsFileResource.getFileEndTime());
569569
tsFileResource.setStatus(TsFileResourceStatus.NORMAL);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java‎

Lines changed: 36 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -100,13 +100,13 @@ public class LoadTsFileTableSchemaCache {
100100
private Map<String, Set<IDeviceID>> currentBatchTable2Devices;
101101

102102
// tableName -> Pair<device column count, device column mapping>
103-
private Map<String, Pair<Integer, Map<Integer, Integer>>> tableIdColumnMapper = new HashMap<>();
103+
private Map<String, Pair<Integer, Map<Integer, Integer>>> tableTagColumnMapper = new HashMap<>();
104104

105105
private PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer> currentModifications;
106106
private ITimeIndex currentTimeIndex;
107107

108108
private long batchTable2DevicesMemoryUsageSizeInBytes = 0;
109-
private long tableIdColumnMapperMemoryUsageSizeInBytes = 0;
109+
private long tableTagColumnMapperMemoryUsageSizeInBytes = 0;
110110
private long currentModificationsMemoryUsageSizeInBytes = 0;
111111
private long currentTimeIndexMemoryUsageSizeInBytes = 0;
112112

@@ -217,22 +217,22 @@ public String getTableName() {
217217
@Override
218218
public List<Object[]> getDeviceIdList() {
219219
final List<Object[]> devices = new ArrayList<>();
220-
final Pair<Integer, Map<Integer, Integer>> idColumnCountAndMapper =
221-
tableIdColumnMapper.get(tableName);
222-
if (Objects.isNull(idColumnCountAndMapper)) {
220+
final Pair<Integer, Map<Integer, Integer>> tagColumnCountAndMapper =
221+
tableTagColumnMapper.get(tableName);
222+
if (Objects.isNull(tagColumnCountAndMapper)) {
223223
// This should not happen
224-
LOGGER.warn("Failed to find id column mapping for table {}", tableName);
224+
LOGGER.warn("Failed to find tag column mapping for table {}", tableName);
225225
}
226226

227227
for (final IDeviceID device : currentBatchTable2Devices.get(tableName)) {
228-
if (Objects.isNull(idColumnCountAndMapper)) {
228+
if (Objects.isNull(tagColumnCountAndMapper)) {
229229
devices.add(Arrays.copyOfRange(device.getSegments(), 1, device.getSegments().length));
230230
continue;
231231
}
232232

233-
final Object[] deviceIdArray = new String[idColumnCountAndMapper.getLeft()];
233+
final Object[] deviceIdArray = new String[tagColumnCountAndMapper.getLeft()];
234234
for (final Map.Entry<Integer, Integer> fileColumn2RealColumn :
235-
idColumnCountAndMapper.getRight().entrySet()) {
235+
tagColumnCountAndMapper.getRight().entrySet()) {
236236
final int fileColumnIndex = fileColumn2RealColumn.getKey();
237237
final int realColumnIndex = fileColumn2RealColumn.getValue();
238238
deviceIdArray[realColumnIndex] =
@@ -294,7 +294,7 @@ public void createTableAndDatabaseIfNecessary(final String tableName)
294294
"Failed to validate schema for table {%s, %s}",
295295
fileSchema.getTableName(), fileSchema));
296296
}
297-
verifyTableDataTypeAndGenerateIdColumnMapper(fileSchema, realSchema);
297+
verifyTableDataTypeAndGenerateTagColumnMapper(fileSchema, realSchema);
298298
}
299299

300300
private void autoCreateTableDatabaseIfAbsent(final String database) throws LoadAnalyzeException {
@@ -329,18 +329,18 @@ private void autoCreateTableDatabaseIfAbsent(final String database) throws LoadA
329329
}
330330
}
331331

332-
private void verifyTableDataTypeAndGenerateIdColumnMapper(
332+
private void verifyTableDataTypeAndGenerateTagColumnMapper(
333333
TableSchema fileSchema, TableSchema realSchema) throws LoadAnalyzeException {
334-
final int realIdColumnCount = realSchema.getIdColumns().size();
335-
final Map<Integer, Integer> idColumnMapping =
336-
tableIdColumnMapper
334+
final int realTagColumnCount = realSchema.getTagColumns().size();
335+
final Map<Integer, Integer> tagColumnMapping =
336+
tableTagColumnMapper
337337
.computeIfAbsent(
338-
realSchema.getTableName(), k -> new Pair<>(realIdColumnCount, new HashMap<>()))
338+
realSchema.getTableName(), k -> new Pair<>(realTagColumnCount, new HashMap<>()))
339339
.getRight();
340340

341-
Map<String, Integer> idColumnNameToIndex = new HashMap<>();
342-
for (int i = 0; i < realSchema.getIdColumns().size(); i++) {
343-
idColumnNameToIndex.put(realSchema.getIdColumns().get(i).getName(), i);
341+
Map<String, Integer> tagColumnNameToIndex = new HashMap<>();
342+
for (int i = 0; i < realSchema.getTagColumns().size(); i++) {
343+
tagColumnNameToIndex.put(realSchema.getTagColumns().get(i).getName(), i);
344344
}
345345
Map<String, ColumnSchema> fieldColumnNameToSchema = new HashMap<>();
346346
for (ColumnSchema column : realSchema.getColumns()) {
@@ -349,16 +349,16 @@ private void verifyTableDataTypeAndGenerateIdColumnMapper(
349349
}
350350
}
351351

352-
int idColumnIndex = 0;
352+
int tagColumnIndex = 0;
353353
for (ColumnSchema fileColumn : fileSchema.getColumns()) {
354354
if (fileColumn.getColumnCategory() == TsTableColumnCategory.TAG) {
355-
Integer realIndex = idColumnNameToIndex.get(fileColumn.getName());
355+
Integer realIndex = tagColumnNameToIndex.get(fileColumn.getName());
356356
if (realIndex != null) {
357-
idColumnMapping.put(idColumnIndex++, realIndex);
357+
tagColumnMapping.put(tagColumnIndex++, realIndex);
358358
} else {
359359
throw new LoadAnalyzeException(
360360
String.format(
361-
"Id column %s in TsFile is not found in IoTDB table %s",
361+
"Tag column %s in TsFile is not found in IoTDB table %s",
362362
fileColumn.getName(), realSchema.getTableName()));
363363
}
364364
} else if (fileColumn.getColumnCategory() == TsTableColumnCategory.FIELD) {
@@ -374,19 +374,19 @@ private void verifyTableDataTypeAndGenerateIdColumnMapper(
374374
}
375375
}
376376
}
377-
updateTableIdColumnMapperMemoryUsageSizeInBytes();
377+
updateTableTagColumnMapperMemoryUsageSizeInBytes();
378378
}
379379

380-
private void updateTableIdColumnMapperMemoryUsageSizeInBytes() {
381-
block.reduceMemoryUsage(tableIdColumnMapperMemoryUsageSizeInBytes);
382-
tableIdColumnMapperMemoryUsageSizeInBytes = 0;
380+
private void updateTableTagColumnMapperMemoryUsageSizeInBytes() {
381+
block.reduceMemoryUsage(tableTagColumnMapperMemoryUsageSizeInBytes);
382+
tableTagColumnMapperMemoryUsageSizeInBytes = 0;
383383
for (final Map.Entry<String, Pair<Integer, Map<Integer, Integer>>> entry :
384-
tableIdColumnMapper.entrySet()) {
385-
tableIdColumnMapperMemoryUsageSizeInBytes += computeStringMemUsage(entry.getKey());
386-
tableIdColumnMapperMemoryUsageSizeInBytes +=
384+
tableTagColumnMapper.entrySet()) {
385+
tableTagColumnMapperMemoryUsageSizeInBytes += computeStringMemUsage(entry.getKey());
386+
tableTagColumnMapperMemoryUsageSizeInBytes +=
387387
(4L + 4L * 2 * entry.getValue().getRight().size());
388388
}
389-
block.addMemoryUsage(tableIdColumnMapperMemoryUsageSizeInBytes);
389+
block.addMemoryUsage(tableTagColumnMapperMemoryUsageSizeInBytes);
390390
}
391391

392392
public void setCurrentModificationsAndTimeIndex(
@@ -431,13 +431,13 @@ public void setCurrentTimeIndex(final ITimeIndex timeIndex) {
431431

432432
public void close() {
433433
clearDevices();
434-
clearIdColumnMapper();
434+
clearTagColumnMapper();
435435
clearModificationsAndTimeIndex();
436436

437437
block.close();
438438

439439
currentBatchTable2Devices = null;
440-
tableIdColumnMapper = null;
440+
tableTagColumnMapper = null;
441441
}
442442

443443
private void clearDevices() {
@@ -456,9 +456,9 @@ private void clearModificationsAndTimeIndex() {
456456
currentTimeIndexMemoryUsageSizeInBytes = 0;
457457
}
458458

459-
public void clearIdColumnMapper() {
460-
tableIdColumnMapper.clear();
461-
block.reduceMemoryUsage(tableIdColumnMapperMemoryUsageSizeInBytes);
462-
tableIdColumnMapperMemoryUsageSizeInBytes = 0;
459+
public void clearTagColumnMapper() {
460+
tableTagColumnMapper.clear();
461+
block.reduceMemoryUsage(tableTagColumnMapperMemoryUsageSizeInBytes);
462+
tableTagColumnMapperMemoryUsageSizeInBytes = 0;
463463
}
464464
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableSchema.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -97,13 +97,13 @@ public ColumnSchema getColumn(final String columnName) {
9797
}
9898

9999
/**
100-
* Given the name of an ID column, return the index of this column among all ID columns, return -1
101-
* if not found.
100+
* Given the name of an TAG column, return the index of this column among all TAG columns, return
101+
* -1 if not found.
102102
*/
103-
public int getIndexAmongIdColumns(final String idColumnName) {
103+
public int getIndexAmongTagColumns(final String tagColumnName) {
104104
int index = 0;
105-
for (final ColumnSchema column : getIdColumns()) {
106-
if (column.getName().equals(idColumnName)) {
105+
for (final ColumnSchema column : getTagColumns()) {
106+
if (column.getName().equals(tagColumnName)) {
107107
return index;
108108
}
109109
index++;
@@ -222,7 +222,7 @@ public String toString() {
222222
return "TableSchema{" + "tableName='" + tableName + '\'' + ", columns=" + columns + '}';
223223
}
224224

225-
public List<ColumnSchema> getIdColumns() {
225+
public List<ColumnSchema> getTagColumns() {
226226
return columns.stream()
227227
.filter(c -> c.getColumnCategory() == TsTableColumnCategory.TAG)
228228
.collect(Collectors.toList());

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -135,13 +135,13 @@ public Optional<TableSchema> validateTableHeaderSchema(
135135
// id columns are the prefix of the incoming id columns, or vice versa
136136
if (isStrictTagColumn) {
137137
final List<TsTableColumnSchema> realTagColumns = table.getTagColumnSchemaList();
138-
final List<ColumnSchema> incomingTagColumns = tableSchema.getIdColumns();
138+
final List<ColumnSchema> incomingTagColumns = tableSchema.getTagColumns();
139139
if (realTagColumns.size() <= incomingTagColumns.size()) {
140140
// When incoming table has more ID columns, the existing id columns
141141
// should be the prefix of the incoming id columns (or equal)
142142
for (int indexReal = 0; indexReal < realTagColumns.size(); indexReal++) {
143143
final String tagName = realTagColumns.get(indexReal).getColumnName();
144-
final int indexIncoming = tableSchema.getIndexAmongIdColumns(tagName);
144+
final int indexIncoming = tableSchema.getIndexAmongTagColumns(tagName);
145145
if (indexIncoming != indexReal) {
146146
throw new LoadAnalyzeTableColumnDisorderException(
147147
String.format(

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/WrappedInsertStatement.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -133,7 +133,7 @@ protected void validateTableSchema(
133133
validateTableSchema(incomingSchemaColumn, realSchemaColumn, i, innerTreeStatement);
134134
}
135135
// incoming schema should contain all id columns in real schema and have consistent order
136-
final List<ColumnSchema> realIdColumns = realSchema.getIdColumns();
136+
final List<ColumnSchema> realIdColumns = realSchema.getTagColumns();
137137
adjustIdColumns(realIdColumns, innerTreeStatement);
138138
}
139139

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTableConvertedInsertTabletStatementExceptionVisitor.java‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.db.storageengine.load.converter;
21+
22+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
23+
import org.apache.iotdb.commons.exception.auth.AccessDeniedException;
24+
import org.apache.iotdb.db.exception.load.LoadRuntimeOutOfMemoryException;
25+
import org.apache.iotdb.db.exception.sql.SemanticException;
26+
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor;
27+
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LoadTsFile;
28+
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Node;
29+
import org.apache.iotdb.rpc.TSStatusCode;
30+
31+
public class LoadTableConvertedInsertTabletStatementExceptionVisitor
32+
extends AstVisitor<TSStatus, Exception> {
33+
@Override
34+
public TSStatus visitNode(final Node node, final Exception context) {
35+
if (context instanceof AccessDeniedException) {
36+
return new TSStatus(TSStatusCode.NO_PERMISSION.getStatusCode())
37+
.setMessage(context.getMessage());
38+
}
39+
return new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode())
40+
.setMessage(context.getMessage());
41+
}
42+
43+
@Override
44+
public TSStatus visitLoadTsFile(final LoadTsFile loadTsFile, final Exception context) {
45+
if (context instanceof LoadRuntimeOutOfMemoryException) {
46+
return new TSStatus(TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode())
47+
.setMessage(context.getMessage());
48+
} else if (context instanceof SemanticException) {
49+
return new TSStatus(TSStatusCode.LOAD_USER_CONFLICT_EXCEPTION.getStatusCode())
50+
.setMessage(context.getMessage());
51+
}
52+
return visitStatement(loadTsFile, context);
53+
}
54+
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTableStatementDataTypeConvertExecutionVisitor.java‎

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -65,10 +65,13 @@ public LoadTableStatementDataTypeConvertExecutionVisitor(StatementExecutor state
6565
public Optional<TSStatus> visitLoadTsFile(
6666
final LoadTsFile loadTsFileStatement, final String databaseName) {
6767
if (Objects.isNull(databaseName)) {
68-
LOGGER.warn(
69-
"Database name is unexpectedly null for LoadTsFileStatement: {}. Skip data type conversion.",
70-
loadTsFileStatement);
71-
return Optional.empty();
68+
final String errorMsg =
69+
String.format(
70+
"Database name is unexpectedly null for LoadTsFileStatement: %s. Skip data type conversion.",
71+
loadTsFileStatement);
72+
LOGGER.warn(errorMsg);
73+
return Optional.of(
74+
new TSStatus(TSStatusCode.SEMANTIC_ERROR.getStatusCode()).setMessage(errorMsg));
7275
}
7376

7477
LOGGER.info("Start data type conversion for LoadTsFileStatement: {}.", loadTsFileStatement);
@@ -100,15 +103,17 @@ public Optional<TSStatus> visitLoadTsFile(
100103
.constructStatement(),
101104
loadTsFileStatement.isConvertOnTypeMismatch());
102105

103-
if (!handleTSStatus(
104-
executeInsertTabletWithRetry(statement, databaseName), loadTsFileStatement)) {
105-
return Optional.empty();
106+
final TSStatus status = executeInsertTabletWithRetry(statement, databaseName);
107+
if (!handleTSStatus(status, loadTsFileStatement)) {
108+
return Optional.of(status);
106109
}
107110
}
108111
} catch (final Exception e) {
109112
LOGGER.warn(
110113
"Failed to convert data type for LoadTsFileStatement: {}.", loadTsFileStatement, e);
111-
return Optional.empty();
114+
return Optional.of(
115+
LoadTsFileDataTypeConverter.TABLE_STATEMENT_EXCEPTION_VISITOR.process(
116+
loadTsFileStatement, e));
112117
}
113118
}
114119

@@ -156,7 +161,7 @@ private TSStatus executeInsertTabletWithRetry(
156161
if (e instanceof InterruptedException) {
157162
Thread.currentThread().interrupt();
158163
}
159-
result = statement.accept(LoadTsFileDataTypeConverter.STATEMENT_EXCEPTION_VISITOR, e);
164+
result = statement.accept(LoadTsFileDataTypeConverter.TREE_STATEMENT_EXCEPTION_VISITOR, e);
160165
}
161166
return result;
162167
}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadConvertedInsertTabletStatementExceptionVisitor.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeConvertedInsertTabletStatementExceptionVisitor.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828
import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
2929
import org.apache.iotdb.rpc.TSStatusCode;
3030

31-
public class LoadConvertedInsertTabletStatementExceptionVisitor
31+
public class LoadTreeConvertedInsertTabletStatementExceptionVisitor
3232
extends StatementVisitor<TSStatus, Exception> {
3333

3434
@Override

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ file, new IoTDBTreePattern(null), Long.MIN_VALUE, Long.MAX_VALUE, null, null)) {
117117
tabletRawReqSizes.clear();
118118

119119
if (!handleTSStatus(result, loadTsFileStatement)) {
120-
return Optional.empty();
120+
return Optional.of(result);
121121
}
122122

123123
tabletRawReqs.add(tabletRawReq);
@@ -127,7 +127,9 @@ file, new IoTDBTreePattern(null), Long.MIN_VALUE, Long.MAX_VALUE, null, null)) {
127127
} catch (final Exception e) {
128128
LOGGER.warn(
129129
"Failed to convert data type for LoadTsFileStatement: {}.", loadTsFileStatement, e);
130-
return Optional.empty();
130+
return Optional.of(
131+
loadTsFileStatement.accept(
132+
LoadTsFileDataTypeConverter.TREE_STATEMENT_EXCEPTION_VISITOR, e));
131133
}
132134
}
133135

@@ -144,12 +146,14 @@ file, new IoTDBTreePattern(null), Long.MIN_VALUE, Long.MAX_VALUE, null, null)) {
144146
tabletRawReqSizes.clear();
145147

146148
if (!handleTSStatus(result, loadTsFileStatement)) {
147-
return Optional.empty();
149+
return Optional.of(result);
148150
}
149151
} catch (final Exception e) {
150152
LOGGER.warn(
151153
"Failed to convert data type for LoadTsFileStatement: {}.", loadTsFileStatement, e);
152-
return Optional.empty();
154+
return Optional.of(
155+
loadTsFileStatement.accept(
156+
LoadTsFileDataTypeConverter.TREE_STATEMENT_EXCEPTION_VISITOR, e));
153157
}
154158
}
155159
} finally {
@@ -181,14 +185,14 @@ file, new IoTDBTreePattern(null), Long.MIN_VALUE, Long.MAX_VALUE, null, null)) {
181185
}
182186

183187
private TSStatus executeInsertMultiTabletsWithRetry(
184-
final List<PipeTransferTabletRawReq> tabletRawReqs, boolean isConvertOnTypeMismatch) {
188+
final List<PipeTransferTabletRawReq> tabletRawReqs, final boolean isConvertedOnTypeMismatch) {
185189
final InsertMultiTabletsStatement batchStatement = new InsertMultiTabletsStatement();
186190
batchStatement.setInsertTabletStatementList(
187191
tabletRawReqs.stream()
188192
.map(
189193
req ->
190194
new LoadConvertedInsertTabletStatement(
191-
req.constructStatement(), isConvertOnTypeMismatch))
195+
req.constructStatement(), isConvertedOnTypeMismatch))
192196
.collect(Collectors.toList()));
193197

194198
TSStatus result;
@@ -214,7 +218,8 @@ private TSStatus executeInsertMultiTabletsWithRetry(
214218
if (e instanceof InterruptedException) {
215219
Thread.currentThread().interrupt();
216220
}
217-
result = batchStatement.accept(LoadTsFileDataTypeConverter.STATEMENT_EXCEPTION_VISITOR, e);
221+
result =
222+
batchStatement.accept(LoadTsFileDataTypeConverter.TREE_STATEMENT_EXCEPTION_VISITOR, e);
218223
}
219224
return result;
220225
}

0 commit comments

Comments
 (0)