Skip to content

Commit 251125a

Browse files
CaideyipiCaideyipi
authored
Fix pipe tsfile receiver database handling (#17815) (#17836)
1 parent 4052320 commit 251125a

13 files changed

Lines changed: 385 additions & 37 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -265,6 +265,10 @@ public File getTsFile() {
265265
return tsFile;
266266
}
267267

268+
public String getDatabaseName() {
269+
return Objects.isNull(resource) ? null : resource.getDatabaseName();
270+
}
271+
268272
public File getModFile() {
269273
return modFile;
270274
}

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

Lines changed: 66 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@
5252
import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementTSStatusVisitor;
5353
import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementToBatchVisitor;
5454
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
55+
import org.apache.iotdb.db.pipe.resource.log.PipePeriodicalLogReducer;
5556
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
5657
import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req;
5758
import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req;
@@ -459,45 +460,70 @@ protected String getSenderPort() {
459460
protected TSStatus loadFileV1(final PipeTransferFileSealReqV1 req, final String fileAbsolutePath)
460461
throws IOException {
461462
return isUsingAsyncLoadTsFileStrategy.get()
462-
? loadTsFileAsync(Collections.singletonList(fileAbsolutePath))
463-
: loadTsFileSync(fileAbsolutePath);
463+
? loadTsFileAsync(null, Collections.singletonList(fileAbsolutePath))
464+
: loadTsFileSync(null, fileAbsolutePath);
464465
}
465466

466467
@Override
467468
protected TSStatus loadFileV2(
468469
final PipeTransferFileSealReqV2 req, final List<String> fileAbsolutePaths)
469470
throws IOException, IllegalPathException {
470-
return req instanceof PipeTransferTsFileSealWithModReq
471-
// TsFile's absolute path will be the second element
472-
? (isUsingAsyncLoadTsFileStrategy.get()
473-
? loadTsFileAsync(fileAbsolutePaths)
474-
: loadTsFileSync(fileAbsolutePaths.get(1)))
475-
: loadSchemaSnapShot(req.getParameters(), fileAbsolutePaths);
471+
if (req instanceof PipeTransferTsFileSealWithModReq) {
472+
final String dataBaseName =
473+
((PipeTransferTsFileSealWithModReq) req).getDatabaseNameByTsFileName();
474+
return isUsingAsyncLoadTsFileStrategy.get()
475+
? loadTsFileAsync(dataBaseName, fileAbsolutePaths)
476+
: loadTsFileSync(dataBaseName, fileAbsolutePaths.get(req.getFileNames().size() - 1));
477+
}
478+
return loadSchemaSnapShot(req.getParameters(), fileAbsolutePaths);
476479
}
477480

478-
private TSStatus loadTsFileAsync(final List<String> absolutePaths) throws IOException {
481+
private TSStatus loadTsFileAsync(final String dataBaseName, final List<String> absolutePaths)
482+
throws IOException {
479483
final Map<String, String> loadAttributes =
480-
ActiveLoadPathHelper.buildAttributes(
481-
null,
484+
buildLoadTsFileAttributesForAsync(
485+
dataBaseName,
482486
shouldConvertDataTypeOnTypeMismatch,
483487
validateTsFile.get(),
484-
null,
485488
shouldMarkAsPipeRequest.get());
486489
if (!ActiveLoadUtil.loadFilesToActiveDir(loadAttributes, absolutePaths, true)) {
487490
throw new PipeException("Load active listening pipe dir is not set.");
488491
}
489492
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
490493
}
491494

492-
private TSStatus loadTsFileSync(final String fileAbsolutePath) throws FileNotFoundException {
495+
static Map<String, String> buildLoadTsFileAttributesForAsync(
496+
final String dataBaseName,
497+
final boolean shouldConvertDataTypeOnTypeMismatch,
498+
final boolean validateTsFile,
499+
final boolean shouldMarkAsPipeRequest) {
500+
return ActiveLoadPathHelper.buildAttributes(
501+
dataBaseName,
502+
LoadTsFileStatement.getDatabaseLevelByTreeDatabase(dataBaseName),
503+
shouldConvertDataTypeOnTypeMismatch,
504+
validateTsFile,
505+
null,
506+
shouldMarkAsPipeRequest);
507+
}
508+
509+
private TSStatus loadTsFileSync(final String dataBaseName, final String fileAbsolutePath)
510+
throws FileNotFoundException {
511+
return executeStatementAndClassifyExceptions(
512+
buildLoadTsFileStatementForSync(dataBaseName, fileAbsolutePath, validateTsFile.get()));
513+
}
514+
515+
static LoadTsFileStatement buildLoadTsFileStatementForSync(
516+
final String dataBaseName, final String fileAbsolutePath, final boolean validateTsFile)
517+
throws FileNotFoundException {
493518
final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(fileAbsolutePath);
494519
statement.setDeleteAfterLoad(true);
495520
statement.setConvertOnTypeMismatch(true);
496-
statement.setVerifySchema(validateTsFile.get());
521+
statement.setVerifySchema(validateTsFile);
497522
statement.setAutoCreateDatabase(
498523
IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled());
499-
500-
return executeStatementAndClassifyExceptions(statement);
524+
statement.setDatabase(dataBaseName);
525+
statement.updateDatabaseLevelByTreeDatabase();
526+
return statement;
501527
}
502528

503529
private TSStatus loadSchemaSnapShot(
@@ -704,12 +730,7 @@ private TSStatus executeStatementAndClassifyExceptions(
704730
return STATEMENT_STATUS_VISITOR.process(statement, result);
705731
}
706732
} catch (final Exception e) {
707-
PipeLogger.log(
708-
LOGGER::warn,
709-
e,
710-
"Receiver id = %s: Exception encountered while executing statement %s: ",
711-
receiverId.get(),
712-
statement.getPipeLoggingString());
733+
logStatementExceptionIfNecessary(statement, e);
713734
return STATEMENT_EXCEPTION_VISITOR.process(statement, e);
714735
} finally {
715736
if (Objects.nonNull(allocatedMemoryBlock)) {
@@ -719,6 +740,29 @@ private TSStatus executeStatementAndClassifyExceptions(
719740
}
720741
}
721742

743+
private void logStatementExceptionIfNecessary(final Statement statement, final Exception e) {
744+
if (shouldLogStatementException(receiverId.get(), statement, e)) {
745+
PipeLogger.log(
746+
LOGGER::warn,
747+
e,
748+
"Receiver id = %s: Exception encountered while executing statement %s: ",
749+
receiverId.get(),
750+
Objects.isNull(statement) ? null : statement.getPipeLoggingString());
751+
}
752+
}
753+
754+
static boolean shouldLogStatementException(
755+
final long receiverId, final Statement statement, final Exception e) {
756+
// Use the reducer cache as a gate. The actual stack trace is logged only when it passes.
757+
return PipePeriodicalLogReducer.log(
758+
message -> {},
759+
"Receiver id = %s, statement = %s, exception = %s, message = %s",
760+
receiverId,
761+
Objects.isNull(statement) ? null : statement.getPipeLoggingString(),
762+
e.getClass().getName(),
763+
e.getMessage());
764+
}
765+
722766
private TSStatus executeStatementWithRetryOnDataTypeMismatch(final Statement statement) {
723767
if (statement == null) {
724768
return RpcUtils.getStatus(

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

Lines changed: 8 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.db.exception.load.LoadRuntimeOutOfMemoryException;
2627
import org.apache.iotdb.db.exception.sql.SemanticException;
@@ -48,6 +49,13 @@
4849
public class PipeStatementExceptionVisitor extends StatementVisitor<TSStatus, Exception> {
4950
@Override
5051
public TSStatus visitNode(final StatementNode node, final Exception context) {
52+
if (context instanceof IoTDBRuntimeException
53+
&& ((IoTDBRuntimeException) context).getErrorCode()
54+
== TSStatusCode.DATABASE_NOT_EXIST.getStatusCode()) {
55+
return new TSStatus(
56+
TSStatusCode.PIPE_RECEIVER_PARALLEL_OR_USER_CONFLICT_EXCEPTION.getStatusCode())
57+
.setMessage(context.getMessage());
58+
}
5159
return new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode())
5260
.setMessage(context.getMessage());
5361
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFileSealWithModReq.java‎

Lines changed: 66 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,9 @@
2525

2626
import java.io.IOException;
2727
import java.util.Arrays;
28+
import java.util.Collections;
2829
import java.util.HashMap;
30+
import java.util.Map;
2931

3032
public class PipeTransferTsFileSealWithModReq extends PipeTransferFileSealReqV2 {
3133

@@ -38,17 +40,59 @@ protected PipeRequestType getPlanType() {
3840
return PipeRequestType.TRANSFER_TS_FILE_SEAL_WITH_MOD;
3941
}
4042

43+
private static final String DATABASE_NAME_KEY_PREFIX = "DATABASE_NAME_";
44+
45+
public String getDatabaseNameByTsFileName() {
46+
return getParameters() == null
47+
? null
48+
: getParameters()
49+
.get(
50+
generateDatabaseNameWithFileNameKey(getFileNames().get(getFileNames().size() - 1)));
51+
}
52+
53+
private static String generateDatabaseNameWithFileNameKey(final String fileName) {
54+
return DATABASE_NAME_KEY_PREFIX + fileName;
55+
}
56+
57+
private static Map<String, String> generateDatabaseNameParameter(
58+
final String tsFileName, final String dataBaseName) {
59+
return dataBaseName == null
60+
? new HashMap<>()
61+
: Collections.singletonMap(generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName);
62+
}
63+
4164
/////////////////////////////// Thrift ///////////////////////////////
4265

4366
public static PipeTransferTsFileSealWithModReq toTPipeTransferReq(
4467
String modFileName, long modFileLength, String tsFileName, long tsFileLength)
4568
throws IOException {
69+
return toTPipeTransferReq(modFileName, modFileLength, tsFileName, tsFileLength, null);
70+
}
71+
72+
public static PipeTransferTsFileSealWithModReq toTPipeTransferReq(
73+
final String modFileName,
74+
final long modFileLength,
75+
final String tsFileName,
76+
final long tsFileLength,
77+
final String dataBaseName)
78+
throws IOException {
4679
return (PipeTransferTsFileSealWithModReq)
4780
new PipeTransferTsFileSealWithModReq()
4881
.convertToTPipeTransferReq(
4982
Arrays.asList(modFileName, tsFileName),
5083
Arrays.asList(modFileLength, tsFileLength),
51-
new HashMap<>());
84+
generateDatabaseNameParameter(tsFileName, dataBaseName));
85+
}
86+
87+
public static PipeTransferTsFileSealWithModReq toTPipeTransferReq(
88+
final String tsFileName, final long tsFileLength, final String dataBaseName)
89+
throws IOException {
90+
return (PipeTransferTsFileSealWithModReq)
91+
new PipeTransferTsFileSealWithModReq()
92+
.convertToTPipeTransferReq(
93+
Collections.singletonList(tsFileName),
94+
Collections.singletonList(tsFileLength),
95+
generateDatabaseNameParameter(tsFileName, dataBaseName));
5296
}
5397

5498
public static PipeTransferTsFileSealWithModReq fromTPipeTransferReq(TPipeTransferReq req) {
@@ -61,11 +105,31 @@ public static PipeTransferTsFileSealWithModReq fromTPipeTransferReq(TPipeTransfe
61105
public static byte[] toTPipeTransferBytes(
62106
String modFileName, long modFileLength, String tsFileName, long tsFileLength)
63107
throws IOException {
108+
return toTPipeTransferBytes(modFileName, modFileLength, tsFileName, tsFileLength, null);
109+
}
110+
111+
public static byte[] toTPipeTransferBytes(
112+
final String modFileName,
113+
final long modFileLength,
114+
final String tsFileName,
115+
final long tsFileLength,
116+
final String dataBaseName)
117+
throws IOException {
64118
return new PipeTransferTsFileSealWithModReq()
65119
.convertToTPipeTransferSnapshotSealBytes(
66120
Arrays.asList(modFileName, tsFileName),
67121
Arrays.asList(modFileLength, tsFileLength),
68-
new HashMap<>());
122+
generateDatabaseNameParameter(tsFileName, dataBaseName));
123+
}
124+
125+
public static byte[] toTPipeTransferBytes(
126+
final String tsFileName, final long tsFileLength, final String dataBaseName)
127+
throws IOException {
128+
return new PipeTransferTsFileSealWithModReq()
129+
.convertToTPipeTransferSnapshotSealBytes(
130+
Collections.singletonList(tsFileName),
131+
Collections.singletonList(tsFileLength),
132+
generateDatabaseNameParameter(tsFileName, dataBaseName));
69133
}
70134

71135
/////////////////////////////// Object ///////////////////////////////

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -255,7 +255,7 @@ private void doTransfer(
255255
final Map<Pair<String, Long>, Double> pipe2WeightMap = batchToTransfer.deepCopyPipe2WeightMap();
256256

257257
for (final File tsFile : sealedFiles) {
258-
doTransfer(pipe2WeightMap, socket, tsFile, null, tsFile.getName());
258+
doTransfer(pipe2WeightMap, socket, tsFile, null, null, tsFile.getName());
259259
try {
260260
RetryUtils.retryOnException(
261261
() -> {
@@ -379,6 +379,7 @@ private void doTransfer(
379379
pipeTsFileInsertionEvent.isWithMod() && supportModsIfIsDataNodeReceiver
380380
? pipeTsFileInsertionEvent.getModFile()
381381
: null,
382+
pipeTsFileInsertionEvent.getDatabaseName(),
382383
pipeTsFileInsertionEvent.toString());
383384
}
384385

@@ -387,6 +388,7 @@ private void doTransfer(
387388
final AirGapSocket socket,
388389
final File tsFile,
389390
final File modFile,
391+
final String dataBaseName,
390392
final String receiverStatusContext)
391393
throws PipeException, IOException {
392394
final String errorMessage = String.format("Seal file %s error. Socket %s.", tsFile, socket);
@@ -397,7 +399,7 @@ private void doTransfer(
397399
if (!sendWeighted(
398400
socket,
399401
PipeTransferTsFileSealWithModReq.toTPipeTransferBytes(
400-
modFile.getName(), modFile.length(), tsFile.getName(), tsFile.length()),
402+
modFile.getName(), modFile.length(), tsFile.getName(), tsFile.length(), dataBaseName),
401403
pipe2WeightMap)) {
402404
receiverStatusHandler.handle(
403405
new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
@@ -411,7 +413,10 @@ private void doTransfer(
411413
transferFilePieces(pipe2WeightMap, tsFile, socket, false);
412414
if (!sendWeighted(
413415
socket,
414-
PipeTransferTsFileSealReq.toTPipeTransferBytes(tsFile.getName(), tsFile.length()),
416+
dataBaseName == null
417+
? PipeTransferTsFileSealReq.toTPipeTransferBytes(tsFile.getName(), tsFile.length())
418+
: PipeTransferTsFileSealWithModReq.toTPipeTransferBytes(
419+
tsFile.getName(), tsFile.length(), dataBaseName),
415420
pipe2WeightMap)) {
416421
receiverStatusHandler.handle(
417422
new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -255,7 +255,8 @@ private void transferInBatchWithoutCheck(
255255
eventsHadBeenAddedToRetryQueue,
256256
sealedFile,
257257
null,
258-
false));
258+
false,
259+
null));
259260
}
260261
} catch (final Exception e) {
261262
PipeLogger.log(LOGGER::warn, e, "Failed to transfer tsfile batch (%s).", sealedFiles);
@@ -400,7 +401,8 @@ private boolean transferWithoutCheck(final TsFileInsertionEvent tsFileInsertionE
400401
pipeTsFileInsertionEvent.getTsFile(),
401402
pipeTsFileInsertionEvent.getModFile(),
402403
pipeTsFileInsertionEvent.isWithMod()
403-
&& clientManager.supportModsIfIsDataNodeReceiver());
404+
&& clientManager.supportModsIfIsDataNodeReceiver(),
405+
pipeTsFileInsertionEvent.getDatabaseName());
404406

405407
transfer(pipeTransferTsFileHandler);
406408
return true;

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,7 @@ public class PipeTransferTsFileHandler extends PipeTransferTrackableHandler {
7878
private File currentFile;
7979

8080
private final boolean transferMod;
81+
private final String dataBaseName;
8182

8283
private final int readFileBufferSize;
8384
private PipeTsFileMemoryBlock memoryBlock;
@@ -98,7 +99,8 @@ public PipeTransferTsFileHandler(
9899
final AtomicBoolean eventsHadBeenAddedToRetryQueue,
99100
final File tsFile,
100101
final File modFile,
101-
final boolean transferMod)
102+
final boolean transferMod,
103+
final String dataBaseName)
102104
throws InterruptedException {
103105
super(connector);
104106

@@ -111,6 +113,7 @@ public PipeTransferTsFileHandler(
111113
this.tsFile = tsFile;
112114
this.modFile = modFile;
113115
this.transferMod = transferMod;
116+
this.dataBaseName = dataBaseName;
114117
currentFile = transferMod ? modFile : tsFile;
115118

116119
// NOTE: Waiting for resource enough for slicing here may cause deadlock!
@@ -191,8 +194,16 @@ public void transfer(
191194
final TPipeTransferReq uncompressedReq =
192195
transferMod
193196
? PipeTransferTsFileSealWithModReq.toTPipeTransferReq(
194-
modFile.getName(), modFile.length(), tsFile.getName(), tsFile.length())
195-
: PipeTransferTsFileSealReq.toTPipeTransferReq(tsFile.getName(), tsFile.length());
197+
modFile.getName(),
198+
modFile.length(),
199+
tsFile.getName(),
200+
tsFile.length(),
201+
dataBaseName)
202+
: dataBaseName == null
203+
? PipeTransferTsFileSealReq.toTPipeTransferReq(
204+
tsFile.getName(), tsFile.length())
205+
: PipeTransferTsFileSealWithModReq.toTPipeTransferReq(
206+
tsFile.getName(), tsFile.length(), dataBaseName);
196207
final TPipeTransferReq req = sink.compressIfNeeded(uncompressedReq);
197208

198209
pipeName2WeightMap.forEach(

0 commit comments

Comments
 (0)