Pipe: Double-living Semantic Correction to Forward Events from Cluster A to B by Default Unless They Originate from B by XNX02 · Pull Request #15112 · apache/iotdb

XNX02 · 2025-03-17T16:06:35Z

No description provided.

…living

SteveYurongSu · 2025-03-19T03:07:37Z

...de/src/main/java/org/apache/iotdb/db/pipe/event/common/deletion/PipeDeleteDataNodeEvent.java

  private AbstractDeleteDataNode deleteDataNode;
  private DeletionResource deletionResource;
  private boolean isGeneratedByPipe;
+  private String originClusterId;


SteveYurongSu · 2025-03-19T03:07:59Z

...de/src/main/java/org/apache/iotdb/db/pipe/event/common/deletion/PipeDeleteDataNodeEvent.java

+        ByteBuffer.allocate(
+            Byte.BYTES + planBuffer.limit() + computeOriginClusterIdBufferSize(originClusterId));
    ReadWriteIOUtils.write(isGeneratedByPipe, result);
+    ReadWriteIOUtils.write(originClusterId, result);


SteveYurongSu · 2025-03-19T03:11:12Z

...nfignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java

+    if (byteBuffer.hasRemaining()) {
+      boolean hasClusterId = byteBuffer.get() != 0;
+      if (hasClusterId) {
+        int strLength = byteBuffer.getShort();
+        byte[] bytes = new byte[strLength];
+        byteBuffer.get(bytes);
+        originClusterId = new String(bytes);
+      } else {
+        originClusterId = null;
+      }
+    }


Check and add some UTs

SteveYurongSu · 2025-03-19T03:16:36Z

...ode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java

+            : insertTabletNode.getOriginClusterId();
+    if (Objects.isNull(workMemTable.getCurrentOriginClusterId())) {
+      workMemTable.setCurrentOriginClusterId(originClusterId);
+    } else if (!Objects.equals(originClusterId, workMemTable.getCurrentOriginClusterId())) {


originClusterId != workMemTable.getCurrentOriginClusterId()

Make a static final cache map in the receiver and get the id string from the map

map <id, id> putIfAbsent

comment here if using the ref comparing

SteveYurongSu · 2025-03-19T03:21:03Z

...ode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java

    }
+    final String originClusterId =
+        insertTabletNode.getOriginClusterId() == null
+            ? config.getClusterId()


Generated or cached?

SteveYurongSu · 2025-03-19T03:21:56Z

...ode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java

+        insertTabletNode.getOriginClusterId() == null
+            ? config.getClusterId()
+            : insertTabletNode.getOriginClusterId();
+    if (Objects.isNull(workMemTable.getCurrentOriginClusterId())) {


if from the same cluster

…living

Caideyipi · 2025-04-16T09:00:30Z

...ava/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeEnrichedPlanV2.java

+
+  @Override
+  public String toString() {
+    return "PipeEnrichedPlanV2{" + "innerPlan='" + innerPlan + "'}";


Better add "originClusterId" here

Caideyipi · 2025-04-16T09:04:05Z

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/PermissionManager.java

   */
-  public TSStatus operatePermission(AuthorPlan authorPlan, boolean isGeneratedByPipe) {
+  public TSStatus operatePermission(
+      AuthorPlan authorPlan, boolean isGeneratedByPipe, String originClusterIds) {


originClusterId~~... Same in other files~~

Caideyipi · 2025-04-16T09:29:11Z

...de/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java

@@ -87,6 +87,19 @@ public AuthOperationProcedure(
    this.timeoutMS = commonConfig.getDatanodeTokenTimeoutMS();
  }


May remove the unused constructors... Same in other files

Caideyipi · 2025-04-16T09:37:23Z

...src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java

  public void pipeEnrichedPlanTest() throws IOException {
-    final PipeEnrichedPlan plan =
-        new PipeEnrichedPlan(
+    final PipeEnrichedPlanV1 plan =


Better use "V2".. Same in other tests

Caideyipi · 2025-04-16T12:15:14Z

...main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionWritePlanEvent.java

    isGeneratedByPipe = ReadWriteIOUtils.readBool(buffer);
    configPhysicalPlan = ConfigPhysicalPlan.Factory.create(buffer);
+
+    // There might be an ignoredChildrenSize 0


The size may not appear on configNodes...

Caideyipi · 2025-04-16T12:17:29Z

...de/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java

    }
  }

+  public Set<String> getSinkClusterIds() {


Is it used?

Caideyipi · 2025-04-16T12:19:20Z

...de/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java

+  }
+
+  public void setSinkClusterIds(Set<String> sinkClusterIds) {
+    realtimeExtractor.setSinkClusterIds(sinkClusterIds);


Personally I think it's better to put this in IoTDBExtractor...

Caideyipi · 2025-04-16T12:21:18Z

...de/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java

    @Override
    public RegionExecutionResult visitPipeEnrichedWritePlanNode(
        final PipeEnrichedWritePlanNode node, final WritePlanNodeExecutionContext context) {
+      node.setOriginClusterId(node.getOriginClusterId());


Seemingly some extra operations are needed to pass the "originClusterId" to the state machine....

Caideyipi · 2025-04-16T12:30:04Z

iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift

  1: required list<common.TConsensusGroupId> schemaRegionIdList
  2: required binary pathPatternTree
  3: optional bool isGeneratedByPipe
+  4: optional string originCluster


Better add "Id" here

Caideyipi · 2025-04-16T12:34:03Z

iotdb-protocol/thrift-datanode/src/main/thrift/client.thrift

 struct TPipeTransferResp {
  1:required common.TSStatus status
  2:optional binary body
+  3:optional string clusterId


Better use "body" instead of adding a new field?

Caideyipi · 2025-04-17T02:02:53Z

.../java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeConnectorSubtaskManager.java

+        if (pipeConnector instanceof IoTDBDataNodeSyncConnector) {
+          attributeSortedStringToSinkClusterIdsMap.put(
+              attributeSortedString,
+              ((IoTDBDataNodeSyncConnector) pipeConnector).getEndPointsClusterIds());


Better put this in "IoTDBConnector"....

Caideyipi · 2025-04-17T02:15:30Z

.../datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java

            insertRowStatement.getValues(),
            insertRowStatement.isNeedInferType());
    insertNode.setFailedMeasurementNumber(insertRowStatement.getFailedMeasurementNumber());
+    insertNode.setOriginClusterId(insertRowStatement.getOriginClusterId());


Do we really need this?

Caideyipi · 2025-04-17T02:19:38Z

...mmons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBNonDataRegionExtractor.java


  protected abstract void confineHistoricalEventTransferTypes(final PipeSnapshotEvent event);

+  public Set<String> getSinkClusterIds() {


May be useless

Caideyipi · 2025-04-17T02:23:49Z

...e/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java

  private long tabletConversionThresholdBytes = -1;
  private boolean autoCreateDatabase = true;
  private boolean isGeneratedByPipe = false;
+  private String originClusterId;


Is it used?

Caideyipi · 2025-04-17T02:24:09Z

...e/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java


  protected long ramBytesUsed = Long.MIN_VALUE;

+  protected String originClusterId;


Is it used?

Caideyipi · 2025-04-17T03:14:31Z

...ore/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java

  protected String password = CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE;

+  // Used to store the clusterId for location comparison
+  public static final Map<String, String> CLUSTER_ID_MAP = new HashMap<>();


Why not use a hashSet....

Caideyipi · 2025-04-17T03:18:49Z

...a/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeEnrichedWritePlanNode.java

  public static PipeEnrichedWritePlanNode deserialize(final ByteBuffer buffer) {
-    return new PipeEnrichedWritePlanNode((WritePlanNode) PlanNodeType.deserialize(buffer));
+    return new PipeEnrichedWritePlanNode(
+        (WritePlanNode) PlanNodeType.deserialize(buffer),


Will there be some ignored children sizes..

Caideyipi · 2025-04-17T03:20:32Z

...de/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java

    @Override
    public RegionExecutionResult visitPipeEnrichedWritePlanNode(
        final PipeEnrichedWritePlanNode node, final WritePlanNodeExecutionContext context) {
+      node.setOriginClusterId(node.getOriginClusterId());


Seemingly some extra operations are needed to pass the "originClusterId" to the state machine....

Caideyipi · 2025-04-17T03:23:18Z

...java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeEnrichedInsertNode.java

-    return new PipeEnrichedInsertNode((InsertNode) PlanNodeType.deserialize(buffer));
+    return new PipeEnrichedInsertNode(
+        (InsertNode) PlanNodeType.deserialize(buffer),
+        buffer.hasRemaining() ? ReadWriteIOUtils.readString(buffer) : null);


What about ignore children size?

Caideyipi · 2025-04-17T03:23:42Z

...rg/apache/iotdb/db/queryengine/plan/planner/plan/node/pipe/PipeEnrichedNonWritePlanNode.java

  public static PipeEnrichedNonWritePlanNode deserialize(ByteBuffer buffer) {
-    return new PipeEnrichedNonWritePlanNode(PlanNodeType.deserialize(buffer));
+    return new PipeEnrichedNonWritePlanNode(
+        PlanNodeType.deserialize(buffer),


What about ignored children size?

Caideyipi · 2025-04-17T03:32:01Z

...ode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java

+        endTime);
+  }
+
+  public PipeTsFileInsertionEvent(


Caideyipi · 2025-04-17T03:33:52Z

...e/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java

  private final LoadTsFileDataCacheMemoryBlock block;
+  private final String originClusterId;

  public LoadTsFileScheduler(


Better remove this?

Caideyipi · 2025-04-17T03:43:29Z

Will any similar mechanism be applied to "WriteBackSink" (i.e. write-back-sink won't transfer data written back once)?

XNX02 added 20 commits March 10, 2025 01:59

clusterId

cf53a0f

update

1b0ad1e

update

9af0445

Merge branch 'master' of https://github.com/apache/iotdb into double-…

03bbeb4

…living

schema

690caa0

config

23029c9

config procudure

8fdca08

update originclusteid

897d41a

update configregionxtratctor

1a1cb5e

fix clusterId

998f491

update

3d828a3

update

1efd28e

Merge branch 'master' of https://github.com/apache/iotdb into double-…

3584efd

…living

deletedata

37aaba1

table model

edee9e8

tsfile simple

1f0c9b5

update

02fbfe9

Merge branch 'master' of https://github.com/apache/iotdb into double-…

ec94f2f

…living

improve

4df0f9d

update

13eff8b

SteveYurongSu self-assigned this Mar 18, 2025

SteveYurongSu reviewed Mar 19, 2025

View reviewed changes

XNX02 added 3 commits March 19, 2025 22:48

add UT for PipeEnrichedProcedureTest

7b7abbb

Merge branch 'master' of https://github.com/apache/iotdb into double-…

4d00afa

…living

update

6a3284f

XNX02 added 4 commits March 20, 2025 23:08

serialize&deserialize

f735818

update

0c0d90e

Merge branch 'master' of https://github.com/apache/iotdb into double-…

3b4e37e

…living

fix

af22cab

XNX02 marked this pull request as ready for review March 23, 2025 16:00

XNX02 added 5 commits March 24, 2025 02:06

update

0c4b22c

update

62b00d4

pipeenrichedplanV2

4779f7c

update

2e3feb7

update

2b1df48

Caideyipi reviewed Apr 16, 2025

View reviewed changes

Caideyipi reviewed Apr 17, 2025

View reviewed changes

SteveYurongSu closed this May 23, 2025

		@@ -87,6 +87,19 @@ public AuthOperationProcedure(
		this.timeoutMS = commonConfig.getDatanodeTokenTimeoutMS();
		}


		protected abstract void confineHistoricalEventTransferTypes(final PipeSnapshotEvent event);

		public Set<String> getSinkClusterIds() {


		protected long ramBytesUsed = Long.MIN_VALUE;

		protected String originClusterId;

Conversation

XNX02 commented Mar 17, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Caideyipi Apr 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Caideyipi commented Apr 17, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Caideyipi Apr 16, 2025 •

edited

Loading