Skip to content

Commit dca8085

Browse files
CaideyipiCaideyipi
committed
Fix pipe history LoadTsFile schema retry
1 parent e0e79aa commit dca8085

5 files changed

Lines changed: 162 additions & 2 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeReceiverAutoCreateDisabledIT.java‎

Lines changed: 62 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import java.sql.ResultSetMetaData;
3737
import java.sql.SQLException;
3838
import java.sql.Statement;
39+
import java.util.Arrays;
3940
import java.util.HashSet;
4041
import java.util.Set;
4142

@@ -61,14 +62,16 @@ protected void setupConfig() {
6162
.getConfig()
6263
.getCommonConfig()
6364
.setDataReplicationFactor(1)
64-
.setSchemaReplicationFactor(1);
65+
.setSchemaReplicationFactor(1)
66+
.setPipeAutoSplitFullEnabled(true);
6567
receiverEnv
6668
.getConfig()
6769
.getCommonConfig()
6870
.setAutoCreateSchemaEnabled(false)
6971
.setDatanodeMemoryProportion("3:3:1:1:1:0")
7072
.setDataReplicationFactor(1)
71-
.setSchemaReplicationFactor(1);
73+
.setSchemaReplicationFactor(1)
74+
.setPipeAutoSplitFullEnabled(true);
7275
}
7376

7477
@Test
@@ -122,6 +125,63 @@ public void testReceiverAutoCreateSchemaDisabledWithSpecialTimeSeries() throws E
122125
}
123126
}
124127

128+
@Test
129+
public void testAutoSplitHistoryTsFileWithDeletionWhenReceiverAutoCreateSchemaDisabled()
130+
throws Exception {
131+
TestUtils.executeNonQueries(
132+
senderEnv,
133+
Arrays.asList(
134+
"create database root.sg",
135+
"create timeseries root.sg.non_aligned.s1 with datatype=INT32",
136+
"create timeseries root.sg.non_aligned.s2 with datatype=DOUBLE",
137+
"create aligned timeseries root.sg.aligned(s1 INT32, s2 DOUBLE)",
138+
"create timeseries root.sg.deleted_measurement.s1 with datatype=INT32",
139+
"create timeseries root.sg.deleted_measurement.s2 with datatype=DOUBLE",
140+
"insert into root.sg.non_aligned(time, s1, s2) values(1, 1, 1.0), (2, 2, 2.0), (3, 3, 3.0)",
141+
"insert into root.sg.aligned(time, s1, s2) values(1, 10, 10.0), (2, 20, 20.0), (3, 30, 30.0)",
142+
"insert into root.sg.deleted_measurement(time, s1, s2) values(1, 100, 100.0), (2, 200, 200.0)",
143+
"flush",
144+
"delete from root.sg.non_aligned.s1 where time > 2",
145+
"delete from root.sg.aligned.* where time > 2",
146+
"delete timeseries root.sg.deleted_measurement.s1",
147+
"flush"));
148+
149+
awaitUntilFlush(senderEnv);
150+
151+
TestUtils.executeNonQuery(
152+
senderEnv,
153+
String.format(
154+
"create pipe test with source ('inclusion'='all', 'source.history.enable'='true', 'source.realtime.mode'='batch') "
155+
+ "with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
156+
receiverEnv.getDataNodeWrapper(0).getIpAndPortString()));
157+
158+
TestUtils.assertDataEventuallyOnEnv(
159+
receiverEnv,
160+
"select * from root.sg.non_aligned",
161+
"Time,root.sg.non_aligned.s1,root.sg.non_aligned.s2,",
162+
new HashSet<>(Arrays.asList("1,1,1.0,", "2,2,2.0,", "3,null,3.0,")));
163+
TestUtils.assertDataEventuallyOnEnv(
164+
receiverEnv,
165+
"select * from root.sg.aligned",
166+
"Time,root.sg.aligned.s1,root.sg.aligned.s2,",
167+
new HashSet<>(Arrays.asList("1,10,10.0,", "2,20,20.0,")));
168+
TestUtils.assertDataEventuallyOnEnv(
169+
receiverEnv,
170+
"select * from root.sg.deleted_measurement",
171+
"Time,root.sg.deleted_measurement.s2,",
172+
new HashSet<>(Arrays.asList("1,100.0,", "2,200.0,")));
173+
TestUtils.assertDataEventuallyOnEnv(
174+
receiverEnv,
175+
"count timeseries root.sg.deleted_measurement.*",
176+
"count(timeseries),",
177+
new HashSet<>(Arrays.asList("1,")));
178+
TestUtils.assertDataEventuallyOnEnv(
179+
receiverEnv,
180+
"show devices root.sg.aligned",
181+
"Device,IsAligned,Template,TTL(ms),",
182+
new HashSet<>(Arrays.asList("root.sg.aligned,true,null,INF,")));
183+
}
184+
125185
private QueryResult queryForResult(final Statement statement, final String sql)
126186
throws SQLException {
127187
try (final ResultSet resultSet = statement.executeQuery(sql)) {

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +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_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode()
7778
|| status.getCode() == TSStatusCode.LOAD_FILE_ERROR.getStatusCode()
7879
&& status.getMessage() != null
7980
&& status.getMessage().contains("memory")) {

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

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,11 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
8383

8484
private static final Logger LOGGER = LoggerFactory.getLogger(LoadTsFileAnalyzer.class);
8585

86+
private static final String MISSING_SCHEMA_MESSAGE =
87+
"does not exist in IoTDB and can not be created";
88+
private static final String AUTO_CREATE_SCHEMA_HINT_MESSAGE =
89+
"Please check weather auto-create-schema is enabled";
90+
8691
private static final LoadTsFileCostMetricsSet LOAD_TSFILE_COST_METRICS_SET =
8792
LoadTsFileCostMetricsSet.getInstance();
8893

@@ -222,6 +227,10 @@ public IAnalysis analyzeFileByFile(IAnalysis analysis) {
222227
executeTabletConversionOnException(analysis, e);
223228
return analysis;
224229
} catch (Exception e) {
230+
if (isTemporaryUnavailableDueToPipeSchemaNotReady(e)) {
231+
setFailAnalysisForTemporaryUnavailablePipeSchema(analysis, e);
232+
return analysis;
233+
}
225234
final String exceptionMessage =
226235
String.format(
227236
"Auto create or verify schema error when executing statement %s. Detail: %s.",
@@ -346,6 +355,10 @@ private boolean doAnalyzeFileByFile(IAnalysis analysis) {
346355
"The file %s is not a valid tsfile. Please check the input file.",
347356
tsFile.getPath()));
348357
} catch (Exception e) {
358+
if (isTemporaryUnavailableDueToPipeSchemaNotReady(e)) {
359+
setFailAnalysisForTemporaryUnavailablePipeSchema(analysis, e);
360+
return false;
361+
}
349362
final String exceptionMessage =
350363
String.format(
351364
"Loading file %s failed. Detail: %s",
@@ -681,8 +694,27 @@ private void setFailAnalysisForAuthException(IAnalysis analysis, AuthException e
681694
analysis.setFailStatus(RpcUtils.getStatus(e.getCode(), e.getMessage()));
682695
}
683696

697+
private void setFailAnalysisForTemporaryUnavailablePipeSchema(
698+
final IAnalysis analysis, final Throwable throwable) {
699+
final String exceptionMessage =
700+
String.format(
701+
"Pipe generated LoadTsFile is waiting for schema metadata to be transferred. Detail: %s",
702+
throwable.getMessage() == null
703+
? throwable.getClass().getName()
704+
: throwable.getMessage());
705+
analysis.setFinishQueryAfterAnalyze(true);
706+
analysis.setFailStatus(
707+
RpcUtils.getStatus(TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION, exceptionMessage));
708+
setRealStatement(analysis);
709+
}
710+
684711
private void executeTabletConversionOnException(
685712
final IAnalysis analysis, final LoadAnalyzeException e) {
713+
if (isTemporaryUnavailableDueToPipeSchemaNotReady(e)) {
714+
setFailAnalysisForTemporaryUnavailablePipeSchema(analysis, e);
715+
return;
716+
}
717+
686718
if (shouldSkipConversion(e)) {
687719
analysis.setFailStatus(
688720
new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(e.getMessage()));
@@ -764,6 +796,26 @@ private void executeTabletConversionOnException(
764796
setRealStatement(analysis);
765797
}
766798

799+
boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable throwable) {
800+
if (!isGeneratedByPipe
801+
|| !isVerifySchema
802+
|| IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()) {
803+
return false;
804+
}
805+
806+
Throwable current = throwable;
807+
while (current != null) {
808+
final String message = current.getMessage();
809+
if (message != null
810+
&& message.contains(MISSING_SCHEMA_MESSAGE)
811+
&& message.contains(AUTO_CREATE_SCHEMA_HINT_MESSAGE)) {
812+
return true;
813+
}
814+
current = current.getCause();
815+
}
816+
return false;
817+
}
818+
767819
private boolean shouldSkipConversion(LoadAnalyzeException e) {
768820
return (e instanceof LoadAnalyzeTypeMismatchException) && !isConvertOnTypeMismatch;
769821
}

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
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

@@ -64,6 +65,18 @@ StatusUtils.OK, new TSStatus(TSStatusCode.OUT_OF_TTL.getStatusCode()))))
6465
.getCode());
6566
}
6667

68+
@Test
69+
public void testLoadTemporaryUnavailableClassification() throws Exception {
70+
Assert.assertEquals(
71+
TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode(),
72+
IoTDBDataNodeReceiver.STATEMENT_STATUS_VISITOR
73+
.process(
74+
LoadTsFileStatement.createUnchecked("temporary-unavailable.tsfile"),
75+
new TSStatus(TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode())
76+
.setMessage("schema is not ready"))
77+
.getCode());
78+
}
79+
6780
@Test
6881
public void testDatabaseNotExistRuntimeExceptionClassification() {
6982
Assert.assertEquals(

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

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,12 @@
1919

2020
package org.apache.iotdb.db.queryengine.plan.analyze.load;
2121

22+
import org.apache.iotdb.commons.exception.SemanticException;
2223
import org.apache.iotdb.commons.path.PartialPath;
2324
import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema;
2425
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
2526
import org.apache.iotdb.db.conf.IoTDBDescriptor;
27+
import org.apache.iotdb.db.exception.load.LoadAnalyzeException;
2628
import org.apache.iotdb.db.exception.load.LoadAnalyzeTypeMismatchException;
2729
import org.apache.iotdb.db.exception.load.LoadRuntimeOutOfMemoryException;
2830
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
@@ -198,6 +200,38 @@ public void testTreeSchemaVerifierShouldThrowMismatchWhenVerifyingDataType() thr
198200
}
199201
}
200202

203+
@Test
204+
public void testPipeGeneratedLoadMissingSchemaShouldBeTemporaryWhenAutoCreateDisabled()
205+
throws Exception {
206+
final boolean originalAutoCreateSchemaEnabled =
207+
IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled();
208+
IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(false);
209+
try (final LoadTsFileAnalyzer analyzer =
210+
new LoadTsFileAnalyzer(
211+
LoadTsFileStatement.createUnchecked("missing-schema.tsfile"),
212+
true,
213+
new MPPQueryContext(new QueryId("load_pipe_test")))) {
214+
Assert.assertTrue(
215+
analyzer.isTemporaryUnavailableDueToPipeSchemaNotReady(
216+
new LoadAnalyzeException(
217+
"Device root.sg.d1 does not exist in IoTDB and can not be created. "
218+
+ "Please check weather auto-create-schema is enabled.")));
219+
Assert.assertTrue(
220+
analyzer.isTemporaryUnavailableDueToPipeSchemaNotReady(
221+
new SemanticException(
222+
"Auto create or verify schema error. Detail: Measurement root.sg.d1.s1 "
223+
+ "does not exist in IoTDB and can not be created. "
224+
+ "Please check weather auto-create-schema is enabled.")));
225+
Assert.assertFalse(
226+
analyzer.isTemporaryUnavailableDueToPipeSchemaNotReady(
227+
new LoadAnalyzeException("Data type mismatch for measurement root.sg.d1.s1")));
228+
} finally {
229+
IoTDBDescriptor.getInstance()
230+
.getConfig()
231+
.setAutoCreateSchemaEnabled(originalAutoCreateSchemaEnabled);
232+
}
233+
}
234+
201235
private void writeTableTsFileWithMixedDevices(final File tsFile) throws Exception {
202236
if (tsFile.exists()) {
203237
Assert.assertTrue(tsFile.delete());

0 commit comments

Comments
 (0)