This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch table_insert_tablets
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/table_insert_tablets by this
push:
new eb27e3f212b Fix relational tablet batch redirection and empty tablets
eb27e3f212b is described below
commit eb27e3f212b3e79c9f8ac60fc4fd4503cb9d40b3
Author: Tian Jiang <[email protected]>
AuthorDate: Mon Jul 20 10:26:32 2026 +0800
Fix relational tablet batch redirection and empty tablets
---
.../java/org/apache/iotdb/session/Session.java | 21 ++++++---
.../apache/iotdb/session/SessionConnection.java | 5 +++
.../iotdb/session/SessionConnectionTest.java | 16 +++++++
.../java/org/apache/iotdb/session/SessionTest.java | 51 ++++++++++++++++++++++
4 files changed, 86 insertions(+), 7 deletions(-)
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
index 63e11b9ba51..edb6690d411 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -2874,6 +2874,19 @@ public class Session implements ISession {
*/
public void insertRelationalTablets(List<Tablet> tablets)
throws IoTDBConnectionException, StatementExecutionException {
+ if (tablets.isEmpty()) {
+ throw new BatchExecutionException(SessionMessages.NO_TABLET_INSERTING);
+ }
+ final List<Tablet> nonEmptyTablets = new ArrayList<>(tablets.size());
+ for (final Tablet tablet : tablets) {
+ if (tablet.getRowSize() > 0) {
+ nonEmptyTablets.add(tablet);
+ }
+ }
+ if (nonEmptyTablets.isEmpty()) {
+ return;
+ }
+ tablets = nonEmptyTablets;
final boolean recordMergeTabletsCost = enableMergeTablets;
long mergeTabletsCost = 0;
if (enableMergeTablets) {
@@ -2881,9 +2894,6 @@ public class Session implements ISession {
tablets = mergeRelationalTablets(tablets);
mergeTabletsCost = System.nanoTime() - mergeTabletsStartTime;
}
- if (tablets.isEmpty()) {
- throw new BatchExecutionException(SessionMessages.NO_TABLET_INSERTING);
- }
final long insertTabletsStartTime = System.nanoTime();
if (enableRedirection) {
insertRelationalTabletsWithLeaderCache(tablets);
@@ -2893,10 +2903,7 @@ public class Session implements ISession {
for (Tablet tablet : tablets) {
request.addToColumnCategoriesList(toEnumOrdinalsAsBytes(tablet.getColumnTypes()));
}
- try {
- getDefaultSessionConnection().insertTablets(request);
- } catch (RedirectException ignored) {
- }
+ getDefaultSessionConnection().insertTabletsWithoutRedirect(request);
}
if (recordMergeTabletsCost) {
recordMergeTabletsCost(mergeTabletsCost, System.nanoTime() -
insertTabletsStartTime);
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
index 4c30ba468f8..42ded5a7e90 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
@@ -876,6 +876,11 @@ public class SessionConnection {
() -> insertTabletsInternal(request), () -> devices);
}
+ protected void insertTabletsWithoutRedirect(TSInsertTabletsReq request)
+ throws IoTDBConnectionException, StatementExecutionException {
+ callWithRetryAndVerify(() -> insertTabletsInternal(request));
+ }
+
private TSStatus insertTabletsInternal(TSInsertTabletsReq request) throws
TException {
request.setSessionId(sessionId);
return client.insertTablets(request);
diff --git
a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionConnectionTest.java
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionConnectionTest.java
index fea78466a78..5c3adb81c06 100644
---
a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionConnectionTest.java
+++
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionConnectionTest.java
@@ -351,6 +351,22 @@ public class SessionConnectionTest {
sessionConnection.deleteTimeseries(Arrays.asList("root.sg1.d1.s1"));
}
+ @Test
+ public void testInsertTabletsWithoutRedirectIgnoresRowRedirectStatus()
+ throws IoTDBConnectionException, StatementExecutionException, TException
{
+ final TSStatus first = new
TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode());
+ first.setRedirectNode(new TEndPoint("127.0.0.2", 6667));
+ final TSStatus second = new
TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode());
+ second.setRedirectNode(new TEndPoint("127.0.0.3", 6667));
+ final TSStatus status = new
TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode());
+ status.setSubStatus(Arrays.asList(first, second));
+ Mockito.when(client.insertTablets(any())).thenReturn(status);
+
+ sessionConnection.insertTabletsWithoutRedirect(new TSInsertTabletsReq());
+
+ Mockito.verify(client).insertTablets(any(TSInsertTabletsReq.class));
+ }
+
@Test
public void testDeleteEmptyTimeseries()
throws IoTDBConnectionException, StatementExecutionException, TException
{
diff --git
a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionTest.java
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionTest.java
index ac5959bd3e9..05d5f0a4643 100644
---
a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionTest.java
+++
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionTest.java
@@ -1368,6 +1368,57 @@ public class SessionTest {
Mockito.verify(sessionConnection).insertTablets(any(TSInsertTabletsReq.class),
anyList());
}
+ @Test
+ public void
testInsertRelationalTabletsWithRedirectionDisabledIgnoresRedirect() throws
Exception {
+ final List<String> measurements = Arrays.asList("tag1", "s1");
+ final Tablet tablet =
+ new Tablet(
+ "table1",
+ measurements,
+ Arrays.asList(TSDataType.STRING, TSDataType.INT64),
+ Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD),
+ 2);
+ tablet.addTimestamp(0, 1L);
+ tablet.addValue("tag1", 0, "d1");
+ tablet.addValue("s1", 0, 11L);
+ tablet.addTimestamp(1, 2L);
+ tablet.addValue("tag1", 1, "d2");
+ tablet.addValue("s1", 1, 22L);
+ Whitebox.setInternalState(session, "enableRedirection", false);
+
+ ((Session)
session).insertRelationalTablets(Collections.singletonList(tablet));
+
+
Mockito.verify(sessionConnection).insertTabletsWithoutRedirect(any(TSInsertTabletsReq.class));
+ Mockito.verify(sessionConnection,
Mockito.never()).insertTablets(any(TSInsertTabletsReq.class));
+ Mockito.verify(sessionConnection, Mockito.never())
+ .insertTablets(any(TSInsertTabletsReq.class), anyList());
+ }
+
+ @Test
+ public void testInsertRelationalTabletsIgnoresAllEmptyTablets() throws
Exception {
+ final Tablet first =
+ new Tablet(
+ "table1",
+ Arrays.asList("tag1", "s1"),
+ Arrays.asList(TSDataType.STRING, TSDataType.INT64),
+ Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD),
+ 1);
+ final Tablet second =
+ new Tablet(
+ "table1",
+ Arrays.asList("tag1", "s1"),
+ Arrays.asList(TSDataType.STRING, TSDataType.INT64),
+ Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD),
+ 1);
+ Whitebox.setInternalState(session, "enableRedirection", true);
+
+ ((Session) session).insertRelationalTablets(Arrays.asList(first, second));
+
+ Mockito.verify(sessionConnection, Mockito.never())
+ .insertTablets(any(TSInsertTabletsReq.class), anyList());
+ Mockito.verify(sessionConnection,
Mockito.never()).insertTablets(any(TSInsertTabletsReq.class));
+ }
+
@Test
public void testSplitRelationalTabletAllocatesOncePerEndpoint() throws
Exception {
final int rowCount = 1000;
