Skip to content

Commit 9ad7c19

Browse files
authored
Pipe: Retry history LoadTsFile while waiting for schema (#18031) (#18042)
* Fix pipe history LoadTsFile schema retry * Address load tsfile schema retry review * Fix load tsfile tests with real temp files (cherry picked from commit 1811e13)
1 parent 65d2582 commit 9ad7c19

4 files changed

Lines changed: 133 additions & 10 deletions

File tree

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
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.exception;
21+
22+
public class LoadAnalyzeMissingSchemaException extends LoadAnalyzeException {
23+
24+
public LoadAnalyzeMissingSchemaException(final String message) {
25+
super(message);
26+
}
27+
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementTSStatusVisitor.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -74,9 +74,7 @@ public TSStatus visitNode(final StatementNode node, final TSStatus status) {
7474
public TSStatus visitLoadFile(
7575
final LoadTsFileStatement loadTsFileStatement, final TSStatus status) {
7676
if (status.getCode() == TSStatusCode.SYSTEM_READ_ONLY.getStatusCode()
77-
|| status.getCode() == TSStatusCode.LOAD_FILE_ERROR.getStatusCode()
78-
&& status.getMessage() != null
79-
&& status.getMessage().contains("memory")) {
77+
|| status.getCode() == TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode()) {
8078
return new TSStatus(
8179
TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode())
8280
.setMessage(status.getMessage());

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

Lines changed: 73 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import org.apache.iotdb.db.conf.IoTDBConfig;
3838
import org.apache.iotdb.db.conf.IoTDBDescriptor;
3939
import org.apache.iotdb.db.exception.LoadAnalyzeException;
40+
import org.apache.iotdb.db.exception.LoadAnalyzeMissingSchemaException;
4041
import org.apache.iotdb.db.exception.LoadAnalyzeTypeMismatchException;
4142
import org.apache.iotdb.db.exception.load.LoadEmptyFileException;
4243
import org.apache.iotdb.db.exception.load.LoadFileException;
@@ -214,6 +215,9 @@ public Analysis analyzeFileByFile(Analysis analysis) {
214215
// the real result on the conversion will be set in the analysis.
215216
return analysis;
216217
} catch (Exception e) {
218+
if (setTemporaryUnavailableStatusIfNecessary(analysis, e)) {
219+
return analysis;
220+
}
217221
final String exceptionMessage =
218222
String.format(
219223
"Auto create or verify schema error when executing statement %s. Detail: %s.",
@@ -315,6 +319,9 @@ private boolean doAnalyzeFileByFile(Analysis analysis) {
315319
"The file %s is not a valid tsfile. Please check the input file.",
316320
tsFile.getPath()));
317321
} catch (Exception e) {
322+
if (setTemporaryUnavailableStatusIfNecessary(analysis, e)) {
323+
return false;
324+
}
318325
final String exceptionMessage =
319326
String.format(
320327
"Loading file %s failed. Detail: %s",
@@ -484,6 +491,10 @@ private Analysis setFailAnalysisForAuthException(Analysis analysis, AuthExceptio
484491

485492
private Analysis executeTabletConversionOnException(
486493
final Analysis analysis, final LoadAnalyzeException e) {
494+
if (setTemporaryUnavailableStatusIfNecessary(analysis, e)) {
495+
return analysis;
496+
}
497+
487498
if (shouldSkipConversion(e)) {
488499
analysis.setFailStatus(
489500
new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(e.getMessage()));
@@ -517,6 +528,52 @@ private Analysis executeTabletConversionOnException(
517528
return analysis;
518529
}
519530

531+
private boolean setTemporaryUnavailableStatusIfNecessary(
532+
final Analysis analysis, final Throwable throwable) {
533+
if (isTemporaryUnavailableDueToPipeSchemaNotReady(throwable)) {
534+
setFailAnalysisForTemporaryUnavailablePipeSchema(analysis, throwable);
535+
return true;
536+
}
537+
if (isGeneratedByPipe && LoadTsFileDataTypeConverter.isMemoryPressureException(throwable)) {
538+
analysis.setFinishQueryAfterAnalyze(true);
539+
analysis.setFailStatus(LoadTsFileDataTypeConverter.getMemoryPressureStatus(throwable));
540+
analysis.setStatement(loadTsFileStatement);
541+
return true;
542+
}
543+
return false;
544+
}
545+
546+
private void setFailAnalysisForTemporaryUnavailablePipeSchema(
547+
final Analysis analysis, final Throwable throwable) {
548+
final String exceptionMessage =
549+
String.format(
550+
"Pipe generated LoadTsFile is waiting for schema metadata to be transferred. Detail: %s",
551+
throwable.getMessage() == null
552+
? throwable.getClass().getName()
553+
: throwable.getMessage());
554+
analysis.setFinishQueryAfterAnalyze(true);
555+
analysis.setFailStatus(
556+
RpcUtils.getStatus(TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION, exceptionMessage));
557+
analysis.setStatement(loadTsFileStatement);
558+
}
559+
560+
boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable throwable) {
561+
if (!isGeneratedByPipe
562+
|| !isVerifySchema
563+
|| IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()) {
564+
return false;
565+
}
566+
567+
Throwable current = throwable;
568+
while (current != null) {
569+
if (current instanceof LoadAnalyzeMissingSchemaException) {
570+
return true;
571+
}
572+
current = current.getCause();
573+
}
574+
return false;
575+
}
576+
520577
private boolean shouldSkipConversion(LoadAnalyzeException e) {
521578
return (e instanceof LoadAnalyzeTypeMismatchException)
522579
&& !loadTsFileStatement.isConvertOnTypeMismatch();
@@ -545,7 +602,7 @@ public void setCurrentTimeIndex(final ITimeIndex timeIndex) {
545602
public void autoCreateAndVerify(
546603
TsFileSequenceReader reader,
547604
Map<IDeviceID, List<TimeseriesMetadata>> device2TimeSeriesMetadataList)
548-
throws IOException, AuthException, LoadAnalyzeTypeMismatchException {
605+
throws IOException, AuthException, LoadAnalyzeException {
549606
for (final Map.Entry<IDeviceID, List<TimeseriesMetadata>> entry :
550607
device2TimeSeriesMetadataList.entrySet()) {
551608
final IDeviceID device = entry.getKey();
@@ -647,14 +704,14 @@ public void flushAndClearDeviceIsAlignedCacheIfNecessary() throws SemanticExcept
647704
schemaCache.clearDeviceIsAlignedCacheIfNecessary();
648705
}
649706

650-
public void flush() throws AuthException, LoadAnalyzeTypeMismatchException {
707+
public void flush() throws AuthException, LoadAnalyzeException {
651708
doAutoCreateAndVerify();
652709

653710
schemaCache.clearTimeSeries();
654711
}
655712

656713
private void doAutoCreateAndVerify()
657-
throws SemanticException, AuthException, LoadAnalyzeTypeMismatchException {
714+
throws SemanticException, AuthException, LoadAnalyzeException {
658715
if (schemaCache.getDevice2TimeSeries().isEmpty()) {
659716
return;
660717
}
@@ -677,6 +734,15 @@ private void doAutoCreateAndVerify()
677734
}
678735
} catch (AuthException | LoadAnalyzeTypeMismatchException e) {
679736
throw e;
737+
} catch (LoadAnalyzeMissingSchemaException e) {
738+
if (isTemporaryUnavailableDueToPipeSchemaNotReady(e)) {
739+
throw e;
740+
}
741+
LOGGER.warn("Auto create or verify schema error.", e);
742+
throw new SemanticException(
743+
String.format(
744+
"Auto create or verify schema error when executing statement %s. Detail: %s.",
745+
loadTsFileStatement, e.getMessage()));
680746
} catch (Exception e) {
681747
if (e.getCause() instanceof LoadAnalyzeTypeMismatchException && isConvertOnTypeMismatch) {
682748
throw (LoadAnalyzeTypeMismatchException) e.getCause();
@@ -863,10 +929,10 @@ private void verifySchema(ISchemaTree schemaTree)
863929
.collect(Collectors.toList()));
864930

865931
if (iotdbDeviceSchemaInfo == null) {
866-
throw new LoadAnalyzeException(
932+
throw new LoadAnalyzeMissingSchemaException(
867933
String.format(
868934
"Device %s does not exist in IoTDB and can not be created. "
869-
+ "Please check weather auto-create-schema is enabled.",
935+
+ "Please check whether auto-create-schema is enabled.",
870936
device));
871937
}
872938

@@ -889,10 +955,10 @@ private void verifySchema(ISchemaTree schemaTree)
889955
final MeasurementSchema tsFileSchema = tsfileTimeseriesSchemas.get(i);
890956
final MeasurementSchema iotdbSchema = iotdbTimeseriesSchemas.get(i);
891957
if (iotdbSchema == null) {
892-
throw new LoadAnalyzeException(
958+
throw new LoadAnalyzeMissingSchemaException(
893959
String.format(
894960
"Measurement %s does not exist in IoTDB and can not be created. "
895-
+ "Please check weather auto-create-schema is enabled.",
961+
+ "Please check whether auto-create-schema is enabled.",
896962
device + TsFileConstant.PATH_SEPARATOR + tsfileTimeseriesSchemas.get(i)));
897963
}
898964

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/PipeStatementTsStatusVisitorTest.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,14 @@
2424
import org.apache.iotdb.commons.utils.StatusUtils;
2525
import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiver;
2626
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement;
27+
import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
2728
import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.BatchActivateTemplateStatement;
2829
import org.apache.iotdb.rpc.TSStatusCode;
2930

3031
import org.junit.Assert;
3132
import org.junit.Test;
3233

34+
import java.io.File;
3335
import java.util.Arrays;
3436
import java.util.Collections;
3537

@@ -64,6 +66,36 @@ StatusUtils.OK, new TSStatus(TSStatusCode.OUT_OF_TTL.getStatusCode()))))
6466
.getCode());
6567
}
6668

69+
@Test
70+
public void testLoadTemporaryUnavailableClassification() throws Exception {
71+
final File tsFile = File.createTempFile("temporary-unavailable", ".tsfile");
72+
tsFile.deleteOnExit();
73+
74+
Assert.assertEquals(
75+
TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode(),
76+
IoTDBDataNodeReceiver.STATEMENT_STATUS_VISITOR
77+
.process(
78+
LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
79+
new TSStatus(TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode())
80+
.setMessage("schema is not ready"))
81+
.getCode());
82+
}
83+
84+
@Test
85+
public void testLoadFileErrorWithMemoryMessageIsNotClassifiedByMessage() throws Exception {
86+
final File tsFile = File.createTempFile("memory-error", ".tsfile");
87+
tsFile.deleteOnExit();
88+
89+
Assert.assertEquals(
90+
TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
91+
IoTDBDataNodeReceiver.STATEMENT_STATUS_VISITOR
92+
.process(
93+
LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
94+
new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode())
95+
.setMessage("memory pressure"))
96+
.getCode());
97+
}
98+
6799
@Test
68100
public void testDatabaseNotExistRuntimeExceptionClassification() {
69101
Assert.assertEquals(

0 commit comments

Comments
 (0)