Skip to content

Commit b46c58b

Browse files
CaideyipiCaideyipi
authored
Fix pipe tsfile receiver database handling (#17815)
1 parent f44d9b7 commit b46c58b

5 files changed

Lines changed: 206 additions & 12 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java‎

Lines changed: 50 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@
5959
import org.apache.iotdb.db.pipe.receiver.visitor.PipeTreeStatementDataTypeConvertExecutionVisitor;
6060
import org.apache.iotdb.db.pipe.receiver.visitor.PipeTreeStatementToBatchVisitor;
6161
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
62+
import org.apache.iotdb.db.pipe.resource.log.PipePeriodicalLogReducer;
6263
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
6364
import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req;
6465
import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req;
@@ -568,12 +569,10 @@ protected TSStatus loadFileV2(
568569
private TSStatus loadTsFileAsync(final String dataBaseName, final List<String> absolutePaths)
569570
throws IOException {
570571
final Map<String, String> loadAttributes =
571-
ActiveLoadPathHelper.buildAttributes(
572+
buildLoadTsFileAttributesForAsync(
572573
dataBaseName,
573-
null,
574574
shouldConvertDataTypeOnTypeMismatch,
575575
validateTsFile.get(),
576-
null,
577576
shouldMarkAsPipeRequest.get());
578577

579578
if (!LoadUtil.loadFilesToActiveDir(loadAttributes, absolutePaths, true)) {
@@ -582,17 +581,38 @@ private TSStatus loadTsFileAsync(final String dataBaseName, final List<String> a
582581
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
583582
}
584583

584+
static Map<String, String> buildLoadTsFileAttributesForAsync(
585+
final String dataBaseName,
586+
final boolean shouldConvertDataTypeOnTypeMismatch,
587+
final boolean validateTsFile,
588+
final boolean shouldMarkAsPipeRequest) {
589+
return ActiveLoadPathHelper.buildAttributes(
590+
dataBaseName,
591+
LoadTsFileStatement.getDatabaseLevelByTreeDatabase(dataBaseName),
592+
shouldConvertDataTypeOnTypeMismatch,
593+
validateTsFile,
594+
null,
595+
shouldMarkAsPipeRequest);
596+
}
597+
585598
private TSStatus loadTsFileSync(final String dataBaseName, final String fileAbsolutePath)
586599
throws FileNotFoundException {
600+
return executeStatementAndClassifyExceptions(
601+
buildLoadTsFileStatementForSync(dataBaseName, fileAbsolutePath, validateTsFile.get()));
602+
}
603+
604+
static LoadTsFileStatement buildLoadTsFileStatementForSync(
605+
final String dataBaseName, final String fileAbsolutePath, final boolean validateTsFile)
606+
throws FileNotFoundException {
587607
final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(fileAbsolutePath);
588608
statement.setDeleteAfterLoad(true);
589609
statement.setConvertOnTypeMismatch(true);
590-
statement.setVerifySchema(validateTsFile.get());
610+
statement.setVerifySchema(validateTsFile);
591611
statement.setAutoCreateDatabase(
592612
IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled());
593613
statement.setDatabase(dataBaseName);
594-
595-
return executeStatementAndClassifyExceptions(statement);
614+
statement.updateDatabaseLevelByTreeDatabase();
615+
return statement;
596616
}
597617

598618
private TSStatus loadSchemaSnapShot(
@@ -845,12 +865,7 @@ private TSStatus executeStatementAndClassifyExceptions(
845865
return STATEMENT_STATUS_VISITOR.process(statement, result);
846866
}
847867
} catch (final Exception e) {
848-
PipeLogger.log(
849-
LOGGER::warn,
850-
e,
851-
"Receiver id = %s: Exception encountered while executing statement %s: ",
852-
receiverId.get(),
853-
statement.getPipeLoggingString());
868+
logStatementExceptionIfNecessary(statement, e);
854869
return STATEMENT_EXCEPTION_VISITOR.process(statement, e);
855870
} finally {
856871
if (Objects.nonNull(allocatedMemoryBlock)) {
@@ -860,6 +875,29 @@ private TSStatus executeStatementAndClassifyExceptions(
860875
}
861876
}
862877

878+
private void logStatementExceptionIfNecessary(final Statement statement, final Exception e) {
879+
if (shouldLogStatementException(receiverId.get(), statement, e)) {
880+
PipeLogger.log(
881+
LOGGER::warn,
882+
e,
883+
"Receiver id = %s: Exception encountered while executing statement %s: ",
884+
receiverId.get(),
885+
Objects.isNull(statement) ? null : statement.getPipeLoggingString());
886+
}
887+
}
888+
889+
static boolean shouldLogStatementException(
890+
final long receiverId, final Statement statement, final Exception e) {
891+
// Use the reducer cache as a gate. The actual stack trace is logged only when it passes.
892+
return PipePeriodicalLogReducer.log(
893+
message -> {},
894+
"Receiver id = %s, statement = %s, exception = %s, message = %s",
895+
receiverId,
896+
Objects.isNull(statement) ? null : statement.getPipeLoggingString(),
897+
e.getClass().getName(),
898+
e.getMessage());
899+
}
900+
863901
private TSStatus executeStatementWithPermissionCheckAndRetryOnDataTypeMismatch(
864902
final Statement statement) {
865903
if (statement == null) {

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2323
import org.apache.iotdb.commons.exception.IoTDBException;
24+
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
2425
import org.apache.iotdb.commons.exception.MetadataException;
2526
import org.apache.iotdb.commons.exception.SemanticException;
2627
import org.apache.iotdb.commons.exception.auth.AccessDeniedException;
@@ -52,6 +53,12 @@ public TSStatus visitNode(final StatementNode node, final Exception context) {
5253
if (context instanceof AccessDeniedException) {
5354
return new TSStatus(TSStatusCode.NO_PERMISSION.getStatusCode())
5455
.setMessage(context.getMessage());
56+
} else if (context instanceof IoTDBRuntimeException
57+
&& ((IoTDBRuntimeException) context).getErrorCode()
58+
== TSStatusCode.DATABASE_NOT_EXIST.getStatusCode()) {
59+
return new TSStatus(
60+
TSStatusCode.PIPE_RECEIVER_PARALLEL_OR_USER_CONFLICT_EXCEPTION.getStatusCode())
61+
.setMessage(context.getMessage());
5562
}
5663
return new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode())
5764
.setMessage(context.getMessage());

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,9 @@
1919

2020
package org.apache.iotdb.db.queryengine.plan.statement.crud;
2121

22+
import org.apache.iotdb.commons.exception.IllegalPathException;
2223
import org.apache.iotdb.commons.path.PartialPath;
24+
import org.apache.iotdb.commons.utils.PathUtils;
2325
import org.apache.iotdb.db.conf.IoTDBConfig;
2426
import org.apache.iotdb.db.conf.IoTDBDescriptor;
2527
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
@@ -44,6 +46,7 @@
4446
import java.util.List;
4547
import java.util.Map;
4648

49+
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_ROOT;
4750
import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.ASYNC_LOAD_KEY;
4851
import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.CONVERT_ON_TYPE_MISMATCH_KEY;
4952
import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.DATABASE_LEVEL_KEY;
@@ -337,6 +340,28 @@ private void initAttributes(final Map<String, String> loadAttributes) {
337340
}
338341
}
339342

343+
public void updateDatabaseLevelByTreeDatabase() {
344+
final Integer databaseLevel = getDatabaseLevelByTreeDatabase(database);
345+
if (databaseLevel != null) {
346+
this.databaseLevel = databaseLevel;
347+
}
348+
}
349+
350+
public static Integer getDatabaseLevelByTreeDatabase(final String database) {
351+
if (database == null) {
352+
return null;
353+
}
354+
try {
355+
final String[] nodes = PathUtils.splitPathToDetachedNodes(database);
356+
if (nodes.length > 1 && PATH_ROOT.equals(nodes[0])) {
357+
return nodes.length - 1;
358+
}
359+
} catch (final IllegalPathException ignored) {
360+
// Keep the configured database level when database is not a legal tree path.
361+
}
362+
return null;
363+
}
364+
340365
public boolean reconstructStatementIfMiniFileConverted(final List<Boolean> isMiniTsFile) {
341366
int lastNonMiniTsFileIndex = -1;
342367

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

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
package org.apache.iotdb.db.pipe.receiver;
2121

2222
import org.apache.iotdb.common.rpc.thrift.TSStatus;
23+
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
2324
import org.apache.iotdb.commons.utils.StatusUtils;
2425
import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiver;
2526
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement;
@@ -62,4 +63,17 @@ public void testTTLIdempotency() {
6263
StatusUtils.OK, new TSStatus(TSStatusCode.OUT_OF_TTL.getStatusCode()))))
6364
.getCode());
6465
}
66+
67+
@Test
68+
public void testDatabaseNotExistRuntimeExceptionClassification() {
69+
Assert.assertEquals(
70+
TSStatusCode.PIPE_RECEIVER_PARALLEL_OR_USER_CONFLICT_EXCEPTION.getStatusCode(),
71+
IoTDBDataNodeReceiver.STATEMENT_EXCEPTION_VISITOR
72+
.process(
73+
new InsertRowsStatement(),
74+
new IoTDBRuntimeException(
75+
"Create DataPartition failed because the database: root.test.sg_0 is not exists",
76+
TSStatusCode.DATABASE_NOT_EXIST.getStatusCode()))
77+
.getCode());
78+
}
6579
}

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
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.pipe.receiver.protocol.thrift;
21+
22+
import org.apache.iotdb.db.conf.IoTDBDescriptor;
23+
import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
24+
import org.apache.iotdb.db.storageengine.load.active.ActiveLoadPathHelper;
25+
import org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator;
26+
27+
import org.junit.Assert;
28+
import org.junit.Test;
29+
30+
import java.nio.file.Files;
31+
import java.nio.file.Path;
32+
import java.util.Map;
33+
34+
public class IoTDBDataNodeReceiverTest {
35+
36+
@Test
37+
public void testLoadTsFileSyncStatementUsesTreeDatabaseLevelFromDatabaseName() throws Exception {
38+
final Path tsFile = Files.createTempFile("pipe-load-tree-database-level", ".tsfile");
39+
try {
40+
final LoadTsFileStatement statement =
41+
IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(
42+
"root.test.sg_0", tsFile.toString(), true);
43+
44+
Assert.assertEquals("root.test.sg_0", statement.getDatabase());
45+
Assert.assertEquals(2, statement.getDatabaseLevel());
46+
} finally {
47+
Files.deleteIfExists(tsFile);
48+
}
49+
}
50+
51+
@Test
52+
public void testLoadTsFileAsyncAttributesUseTreeDatabaseLevelFromDatabaseName() throws Exception {
53+
final Path tsFile = Files.createTempFile("pipe-async-load-tree-database-level", ".tsfile");
54+
try {
55+
final Map<String, String> attributes =
56+
IoTDBDataNodeReceiver.buildLoadTsFileAttributesForAsync(
57+
"root.test.sg_0", true, true, true);
58+
59+
Assert.assertEquals(
60+
"root.test.sg_0", attributes.get(LoadTsFileConfigurator.DATABASE_NAME_KEY));
61+
Assert.assertEquals("2", attributes.get(LoadTsFileConfigurator.DATABASE_LEVEL_KEY));
62+
63+
final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(tsFile.toString());
64+
ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement, true);
65+
Assert.assertEquals("root.test.sg_0", statement.getDatabase());
66+
Assert.assertEquals(2, statement.getDatabaseLevel());
67+
} finally {
68+
Files.deleteIfExists(tsFile);
69+
}
70+
}
71+
72+
@Test
73+
public void testLoadTsFileSyncStatementKeepsDefaultDatabaseLevelWhenDatabaseNameIsNull()
74+
throws Exception {
75+
final Path tsFile = Files.createTempFile("pipe-load-default-database-level", ".tsfile");
76+
try {
77+
final LoadTsFileStatement statement =
78+
IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(null, tsFile.toString(), true);
79+
80+
Assert.assertNull(statement.getDatabase());
81+
Assert.assertEquals(
82+
IoTDBDescriptor.getInstance().getConfig().getDefaultDatabaseLevel(),
83+
statement.getDatabaseLevel());
84+
} finally {
85+
Files.deleteIfExists(tsFile);
86+
}
87+
}
88+
89+
@Test
90+
public void testRepeatedStatementExceptionLogIsReduced() throws Exception {
91+
final Path tsFile = Files.createTempFile("pipe-load-log-reducer", ".tsfile");
92+
try {
93+
final LoadTsFileStatement statement =
94+
IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(
95+
"root.test.sg_0", tsFile.toString(), true);
96+
final long receiverId = System.nanoTime();
97+
final Exception exception = new RuntimeException("repeated receiver exception " + receiverId);
98+
99+
Assert.assertTrue(
100+
IoTDBDataNodeReceiver.shouldLogStatementException(receiverId, statement, exception));
101+
Assert.assertFalse(
102+
IoTDBDataNodeReceiver.shouldLogStatementException(receiverId, statement, exception));
103+
Assert.assertTrue(
104+
IoTDBDataNodeReceiver.shouldLogStatementException(
105+
receiverId, statement, new RuntimeException("another receiver exception")));
106+
} finally {
107+
Files.deleteIfExists(tsFile);
108+
}
109+
}
110+
}

0 commit comments

Comments
 (0)