You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/09/21 09:42:13 UTC
(iotdb) branch fix/pipe-tablet-batch-redirect created (now aa9df00c6a8)
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a change to branch fix/pipe-tablet-batch-redirect
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at aa9df00c6a8 Fix table-model pipe batch leader redirects
This branch includes the following new commits:
new aa9df00c6a8 Fix table-model pipe batch leader redirects
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: Fix table-model pipe batch leader redirects
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch fix/pipe-tablet-batch-redirect
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit aa9df00c6a8307e232460d7bb359eb22b071f439
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Mon Sep 21 17:41:43 2026 +0800
Fix table-model pipe batch leader redirects
---
.../protocol/thrift/IoTDBDataNodeReceiver.java | 25 +++++---
.../db/pipe/sink/util/cacher/LeaderCacheUtils.java | 15 +++--
.../protocol/thrift/IoTDBDataNodeReceiverTest.java | 70 ++++++++++++++++++++++
.../sink/util/cacher/LeaderCacheUtilsTest.java | 42 +++++++++++++
4 files changed, 139 insertions(+), 13 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
index 895993704e2..f92d798cc41 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -1279,18 +1279,27 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver {
}
/**
- * For {@link InsertRowsStatement} and {@link InsertMultiTabletsStatement}, the returned {@link
- * TSStatus} will use sub-status to record the endpoint for redirection. Each sub-status records
- * the redirection endpoint for one device path, and the order is the same as the order of the
- * device paths in the statement. However, this order is not guaranteed to be the same as in the
- * request. So for each sub-status which needs to redirect, we record the device path using the
- * message field.
+ * For tree-model {@link InsertRowsStatement} and {@link InsertMultiTabletsStatement}, the
+ * returned {@link TSStatus} uses sub-statuses to record redirection endpoints. Their order is the
+ * same as the device paths in the statement, but not necessarily the request, so attach the
+ * device path to each redirected sub-status.
*/
private TSStatus executeBatchStatementAndAddRedirectInfo(final InsertBaseStatement statement) {
final TSStatus result = executeStatementAndClassifyExceptions(statement, 5);
+ return addRedirectInfoForBatch(statement, result, receiverId.get());
+ }
+ static TSStatus addRedirectInfoForBatch(
+ final InsertBaseStatement statement, final TSStatus result, final long receiverId) {
if (result.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()
&& result.getSubStatusSize() > 0) {
+ // A table-model batch may contain rows for multiple devices. The pipe sink currently routes
+ // the entire event by its device ID, so caching a row's endpoint for the whole tablet/table
+ // could misroute later writes. Keep the successful write status without cache hints.
+ if (statement.isWriteToTable()) {
+ return result;
+ }
+
final List<PartialPath> devicePaths;
if (statement instanceof InsertRowsStatement) {
devicePaths = ((InsertRowsStatement) statement).getDevicePaths();
@@ -1299,7 +1308,7 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver {
} else {
LOGGER.warn(
DataNodePipeMessages.RECEIVER_ID_UNSUPPORTED_STATEMENT_TYPE_FOR_REDIRECTION,
- receiverId.get(),
+ receiverId,
statement);
return result;
}
@@ -1313,7 +1322,7 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver {
} else {
LOGGER.warn(
DataNodePipeMessages.RECEIVER_ID_THE_NUMBER_OF_DEVICE_PATHS,
- receiverId.get(),
+ receiverId,
statement,
result);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/cacher/LeaderCacheUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/cacher/LeaderCacheUtils.java
index 0f6beade80d..bb0357d3b90 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/cacher/LeaderCacheUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/cacher/LeaderCacheUtils.java
@@ -45,19 +45,24 @@ public class LeaderCacheUtils {
// requests may contain any number of statements because rows are grouped by database and table.
final List<Pair<String, TEndPoint>> redirectList = new ArrayList<>();
- if (!status.isSetSubStatus()) {
+ if (status == null || !status.isSetSubStatus()) {
return redirectList;
}
for (final TSStatus subStatus : status.getSubStatus()) {
- if (subStatus.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
+ if (subStatus == null
+ || subStatus.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()
+ || !subStatus.isSetSubStatus()) {
continue;
}
for (final TSStatus innerSubStatus : subStatus.getSubStatus()) {
- if (innerSubStatus.isSetRedirectNode()) {
- // We assume that innerSubStatus.getMessage() is a device path.
- // The message field should be a device path.
+ if (innerSubStatus != null
+ && innerSubStatus.isSetRedirectNode()
+ && innerSubStatus.isSetMessage()
+ && !innerSubStatus.getMessage().isEmpty()) {
+ // The receiver sets the message to a device path only when it can safely associate the
+ // redirection with a single tree-model device.
redirectList.add(
new Pair<>(innerSubStatus.getMessage(), innerSubStatus.getRedirectNode()));
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
index f6e57e15054..b71542a0cc0 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
@@ -19,7 +19,9 @@
package org.apache.iotdb.db.pipe.receiver.protocol.thrift;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.pipe.receiver.runtime.PipeReceiverRuntimeRegistry;
import org.apache.iotdb.commons.pipe.receiver.runtime.PipeReceiverRuntimeSnapshot;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -62,6 +64,74 @@ public class IoTDBDataNodeReceiverTest {
registry.clear();
}
+ @Test
+ public void testTableModelTabletRedirectDoesNotCachePerRowLeader() {
+ final InsertTabletStatement statement = new InsertTabletStatement();
+ statement.setWriteToTable(true);
+ statement.setRowCount(2);
+
+ final TSStatus firstRow =
+ new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())
+ .setRedirectNode(new TEndPoint("127.0.0.2", 6667));
+ final TSStatus secondRow =
+ new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())
+ .setRedirectNode(new TEndPoint("127.0.0.3", 6667));
+ final TSStatus result =
+ new TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode())
+ .setSubStatus(Arrays.asList(firstRow, secondRow));
+
+ Assert.assertSame(result, IoTDBDataNodeReceiver.addRedirectInfoForBatch(statement, result, 1));
+ Assert.assertFalse(firstRow.isSetMessage());
+ Assert.assertFalse(secondRow.isSetMessage());
+ }
+
+ @Test
+ public void testTableModelRowsRedirectDoesNotCachePerRowLeader() throws Exception {
+ final InsertRowStatement firstRow = new InsertRowStatement();
+ firstRow.setDevicePath(new PartialPath("table1"));
+ final InsertRowStatement secondRow = new InsertRowStatement();
+ secondRow.setDevicePath(new PartialPath("table1"));
+ final InsertRowsStatement statement = new InsertRowsStatement();
+ statement.setWriteToTable(true);
+ statement.setInsertRowStatementList(Arrays.asList(firstRow, secondRow));
+
+ final TSStatus firstRowStatus =
+ new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())
+ .setRedirectNode(new TEndPoint("127.0.0.2", 6667));
+ final TSStatus secondRowStatus =
+ new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())
+ .setRedirectNode(new TEndPoint("127.0.0.3", 6667));
+ final TSStatus result =
+ new TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode())
+ .setSubStatus(Arrays.asList(firstRowStatus, secondRowStatus));
+
+ Assert.assertSame(result, IoTDBDataNodeReceiver.addRedirectInfoForBatch(statement, result, 1));
+ Assert.assertFalse(firstRowStatus.isSetMessage());
+ Assert.assertFalse(secondRowStatus.isSetMessage());
+ }
+
+ @Test
+ public void testTreeModelBatchRedirectHasDevicePath() throws Exception {
+ final InsertRowStatement firstRow = new InsertRowStatement();
+ firstRow.setDevicePath(new PartialPath("root.sg.d1"));
+ final InsertRowStatement secondRow = new InsertRowStatement();
+ secondRow.setDevicePath(new PartialPath("root.sg.d2"));
+ final InsertRowsStatement statement = new InsertRowsStatement();
+ statement.setInsertRowStatementList(Arrays.asList(firstRow, secondRow));
+
+ final TSStatus local = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ final TSStatus redirected =
+ new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())
+ .setRedirectNode(new TEndPoint("127.0.0.2", 6667));
+ final TSStatus result =
+ new TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode())
+ .setSubStatus(Arrays.asList(local, redirected));
+
+ Assert.assertSame(result, IoTDBDataNodeReceiver.addRedirectInfoForBatch(statement, result, 1));
+ Assert.assertFalse(local.isSetMessage());
+ Assert.assertEquals("root.sg.d2", redirected.getMessage());
+ }
+
@Test
public void testLoadTsFileSyncStatementUsesTreeDatabaseLevelFromDatabaseName() throws Exception {
final Path tsFile = Files.createTempFile("pipe-load-tree-database-level", ".tsfile");
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/util/cacher/LeaderCacheUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/util/cacher/LeaderCacheUtilsTest.java
index 76c6bef2467..62d5b57cedf 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/util/cacher/LeaderCacheUtilsTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/util/cacher/LeaderCacheUtilsTest.java
@@ -59,4 +59,46 @@ public class LeaderCacheUtilsTest {
Assert.assertEquals("table1.device1", redirects.get(0).getLeft());
Assert.assertEquals(redirectEndPoint, redirects.get(0).getRight());
}
+
+ @Test
+ public void testIgnoreRedirectsWithoutDevicePath() {
+ final TEndPoint redirectEndPoint = new TEndPoint("127.0.0.2", 6667);
+ final TSStatus tableRowWithoutPath =
+ RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS).setRedirectNode(redirectEndPoint);
+ final TSStatus rowWithEmptyPath =
+ RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)
+ .setMessage("")
+ .setRedirectNode(redirectEndPoint);
+ final TSStatus treeRowWithPath =
+ RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)
+ .setMessage("root.sg.d1")
+ .setRedirectNode(redirectEndPoint);
+ final TSStatus batchStatus =
+ RpcUtils.getStatus(TSStatusCode.REDIRECTION_RECOMMEND)
+ .setSubStatus(
+ Arrays.asList(
+ RpcUtils.getStatus(TSStatusCode.REDIRECTION_RECOMMEND)
+ .setSubStatus(Arrays.asList(tableRowWithoutPath, rowWithEmptyPath)),
+ RpcUtils.getStatus(TSStatusCode.REDIRECTION_RECOMMEND)
+ .setSubStatus(Collections.singletonList(treeRowWithPath))));
+
+ final List<Pair<String, TEndPoint>> redirects =
+ LeaderCacheUtils.parseRecommendedRedirections(batchStatus);
+
+ Assert.assertEquals(1, redirects.size());
+ Assert.assertEquals("root.sg.d1", redirects.get(0).getLeft());
+ Assert.assertEquals(redirectEndPoint, redirects.get(0).getRight());
+ }
+
+ @Test
+ public void testIgnoreMalformedRedirectStatus() {
+ final TSStatus redirectWithoutSubStatus =
+ RpcUtils.getStatus(TSStatusCode.REDIRECTION_RECOMMEND);
+ final TSStatus batchStatus =
+ RpcUtils.getStatus(TSStatusCode.REDIRECTION_RECOMMEND)
+ .setSubStatus(Arrays.asList(null, redirectWithoutSubStatus));
+
+ Assert.assertTrue(LeaderCacheUtils.parseRecommendedRedirections(batchStatus).isEmpty());
+ Assert.assertTrue(LeaderCacheUtils.parseRecommendedRedirections(null).isEmpty());
+ }
}