Pipe: Implement sink.data-distribution-strategy feature where a pipe can send data to all sink nodes#14922
Pipe: Implement sink.data-distribution-strategy feature where a pipe can send data to all sink nodes#14922XNX02 wants to merge 16 commits intoapache:masterfrom
Conversation
| PipeTransferConfigPlanReq.toTPipeTransferBytes( | ||
| pipeConfigRegionWritePlanEvent.getConfigPhysicalPlan()))) { |
There was a problem hiding this comment.
I think this byte[] only needs to be constructed once, which can reduce a lot of unnecessary overhead
There was a problem hiding this comment.
Appreciate it. I've made some changes about this problem. Please take a look when you have time and let me know if you think it's working as expected.
| PipeTransferConfigSnapshotSealReq.toTPipeTransferBytes( | ||
| // The pattern is surely Non-null | ||
| pipeConfigRegionSnapshotEvent.getTreePatternString(), | ||
| pipeConfigRegionSnapshotEvent.getTablePattern().getDatabasePattern(), | ||
| pipeConfigRegionSnapshotEvent.getTablePattern().getTablePattern(), | ||
| pipeConfigRegionSnapshotEvent.getTreePattern().isTreeModelDataAllowedToBeCaptured(), | ||
| pipeConfigRegionSnapshotEvent | ||
| .getTablePattern() | ||
| .isTableModelDataAllowedToBeCaptured(), | ||
| snapshot.getName(), | ||
| snapshot.length(), | ||
| Objects.nonNull(templateFile) ? templateFile.getName() : null, | ||
| Objects.nonNull(templateFile) ? templateFile.length() : 0, | ||
| pipeConfigRegionSnapshotEvent.getFileType(), | ||
| pipeConfigRegionSnapshotEvent.toSealTypeString()))) { |
| final TPipeTransferReq req = | ||
| compressIfNeeded( | ||
| PipeTransferConfigPlanReq.toTPipeTransferReq( | ||
| pipeConfigRegionWritePlanEvent.getConfigPhysicalPlan())); |
There was a problem hiding this comment.
If the byte[] here can also be reused, the sending performance can be greatly improved. It should also be noted here that the internal ByteBuffer needs to copy the internal array
There was a problem hiding this comment.
Could you clarify what you meant by 'internal ByteBuffer needs to copy the internal array'? It sounds that simply place the req outside the loop is not enough in this situation.
There was a problem hiding this comment.
Because ByteBuffer has capacity, limit, position and mark to record the corresponding progress. But I looked at the code and found that a new ByteBuffer will be created when writing. You can ignore my words.
| .trim() | ||
| .toLowerCase(); | ||
| validator.validate( | ||
| arg -> arg.equals("any") || arg.equals("all"), |
There was a problem hiding this comment.
Wouldn't it be better to define two constants for Any and All?
3 participants
No description provided.