Optimize relational tablet batch inserts

integration-test/src/test/java/org/apache/iotdb/relational/it/session/pool/IoTDBInsertTableSessionPoolIT.java

@@ -250,6 +250,187 @@ public void testInsertKeyword() throws IoTDBConnectionException, StatementExecut
         assertEquals("20240815", rec.getFields().get(12).getStringValue());
       }
       assertFalse(rs1.hasNext());
+
+      Tablet tablet2 =
+          new Tablet(
+              "table20",
+              IMeasurementSchema.getMeasurementNameList(schemas),
+              schemas.stream().map(IMeasurementSchema::getType).collect(Collectors.toList()),
+              columnTypes,
+              10);
+
+      for (long row = 10; row < 20; row++) {
+        int rowIndex = tablet2.getRowSize();
+        tablet2.addTimestamp(rowIndex, timestamp + row);
+        tablet2.addValue("device_id", rowIndex, "2");
+        tablet2.addValue("attribute", rowIndex, "2");
+        tablet2.addValue("boolean", rowIndex, true);
+        tablet2.addValue("int32", rowIndex, Integer.valueOf("2"));
+        tablet2.addValue("int64", rowIndex, Long.valueOf("2"));
+        tablet2.addValue("float", rowIndex, Float.valueOf("2.0"));
+        tablet2.addValue("double", rowIndex, Double.valueOf("2.0"));
+        tablet2.addValue("text", rowIndex, "false");
+        tablet2.addValue("string", rowIndex, "false");
+        tablet2.addValue("blob", rowIndex, new Binary("iotdb", Charset.defaultCharset()));
+        tablet2.addValue("timestamp", rowIndex, 2L);
+        tablet2.addValue("date", rowIndex, LocalDate.parse("2024-08-16"));
+      }
+
+      Tablet tablet3 =
+          new Tablet(
+              "table20",
+              IMeasurementSchema.getMeasurementNameList(schemas),
+              schemas.stream().map(IMeasurementSchema::getType).collect(Collectors.toList()),
+              columnTypes,
+              10);
+
+      for (long row = 20; row < 30; row++) {
+        int rowIndex = tablet3.getRowSize();
+        tablet3.addTimestamp(rowIndex, timestamp + row);
+        tablet3.addValue("device_id", rowIndex, "2");
+        tablet3.addValue("attribute", rowIndex, "2");
+        tablet3.addValue("boolean", rowIndex, true);
+        tablet3.addValue("int32", rowIndex, Integer.valueOf("2"));
+        tablet3.addValue("int64", rowIndex, Long.valueOf("2"));
+        tablet3.addValue("float", rowIndex, Float.valueOf("2.0"));
+        tablet3.addValue("double", rowIndex, Double.valueOf("2.0"));
+        tablet3.addValue("text", rowIndex, "false");
+        tablet3.addValue("string", rowIndex, "false");
+        tablet3.addValue("blob", rowIndex, new Binary("iotdb", Charset.defaultCharset()));
+        tablet3.addValue("timestamp", rowIndex, 2L);
+        tablet3.addValue("date", rowIndex, LocalDate.parse("2024-08-16"));
+      }
+      session.insertTablets(Arrays.asList(tablet2, tablet3));
+
+      try (SessionDataSet rs2 =
+          session.executeQueryStatement("select count(*) from table20 where device_id='2'")) {
+        RowRecord rec = rs2.next();
+        assertEquals(20, rec.getFields().get(0).getLongV());
+        assertFalse(rs2.hasNext());
+      }
+    }
+  }
+
+  @Test
+  public void testInsertTabletsWithDifferentTables()
+      throws IoTDBConnectionException, StatementExecutionException {
+    final ITableSessionPool sessionPool = EnvFactory.getEnv().getTableSessionPool(1);
+    try (final ITableSession session = sessionPool.getSession()) {
+      session.executeNonQueryStatement("USE \"test\"");
+      session.executeNonQueryStatement("DROP TABLE IF EXISTS multi_tablet_table1");
+      session.executeNonQueryStatement("DROP TABLE IF EXISTS multi_tablet_table2");
+      session.executeNonQueryStatement("DROP TABLE IF EXISTS multi_tablet_table3");
+      session.executeNonQueryStatement(
+          "CREATE TABLE multi_tablet_table1 (tag1 STRING TAG, attr1 STRING ATTRIBUTE, s1 INT64 FIELD)");
+      session.executeNonQueryStatement(
+          "CREATE TABLE multi_tablet_table2 (tag2 STRING TAG, s2 DOUBLE FIELD)");
+      session.executeNonQueryStatement(
+          "CREATE TABLE multi_tablet_table3 (region STRING TAG, plant STRING TAG, status STRING ATTRIBUTE, s3 BOOLEAN FIELD)");
+
+      final Tablet tablet1 = createTabletForTable1(0, 3);
+      final Tablet tablet2 = createTabletForTable2(10, 3);
+      final Tablet tablet3 = createTabletForTable1(3, 2);
+      final Tablet tablet4 = createTabletForTable3(20, 4);
+
+      session.insertTablets(Arrays.asList(tablet1, tablet2, tablet3, tablet4));
+
+      assertCount(session, "SELECT COUNT(*) FROM multi_tablet_table1 WHERE tag1 = 'tag1'", 5);
+      assertCount(session, "SELECT COUNT(*) FROM multi_tablet_table2 WHERE tag2 = 'tag2'", 3);
+      assertCount(
+          session,
+          "SELECT COUNT(*) FROM multi_tablet_table3 WHERE region = 'r1' AND plant = 'p1'",
+          4);
+      assertCount(
+          session,
+          "SELECT COUNT(*) FROM multi_tablet_table1 WHERE tag1 = 'tag1' AND attr1 = 'attr1'",
+          5);
+      assertCount(
+          session,
+          "SELECT COUNT(*) FROM multi_tablet_table3 WHERE region = 'r1' AND plant = 'p1' AND status = 'running'",
+          4);
+    }
+  }
+
+  private static Tablet createTabletForTable1(final long startTime, final int rowCount) {
+    final List<IMeasurementSchema> schemas = new ArrayList<>();
+    schemas.add(new MeasurementSchema("tag1", TSDataType.STRING));
+    schemas.add(new MeasurementSchema("attr1", TSDataType.STRING));
+    schemas.add(new MeasurementSchema("s1", TSDataType.INT64));
+    final List<ColumnCategory> columnTypes =
+        Arrays.asList(ColumnCategory.TAG, ColumnCategory.ATTRIBUTE, ColumnCategory.FIELD);
+    final Tablet tablet =
+        new Tablet(
+            "multi_tablet_table1",
+            IMeasurementSchema.getMeasurementNameList(schemas),
+            IMeasurementSchema.getDataTypeList(schemas),
+            columnTypes,
+            rowCount);
+    for (int i = 0; i < rowCount; i++) {
+      final int rowIndex = tablet.getRowSize();
+      tablet.addTimestamp(rowIndex, startTime + i);
+      tablet.addValue("tag1", rowIndex, "tag1");
+      tablet.addValue("attr1", rowIndex, "attr1");
+      tablet.addValue("s1", rowIndex, startTime + i);
+    }
+    return tablet;
+  }
+
+  private static Tablet createTabletForTable2(final long startTime, final int rowCount) {
+    final List<IMeasurementSchema> schemas = new ArrayList<>();
+    schemas.add(new MeasurementSchema("tag2", TSDataType.STRING));
+    schemas.add(new MeasurementSchema("s2", TSDataType.DOUBLE));
+    final List<ColumnCategory> columnTypes =
+        Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD);
+    final Tablet tablet =
+        new Tablet(
+            "multi_tablet_table2",
+            IMeasurementSchema.getMeasurementNameList(schemas),
+            IMeasurementSchema.getDataTypeList(schemas),
+            columnTypes,
+            rowCount);
+    for (int i = 0; i < rowCount; i++) {
+      final int rowIndex = tablet.getRowSize();
+      tablet.addTimestamp(rowIndex, startTime + i);
+      tablet.addValue("tag2", rowIndex, "tag2");
+      tablet.addValue("s2", rowIndex, (startTime + i) * 1.0);
+    }
+    return tablet;
+  }
+
+  private static Tablet createTabletForTable3(final long startTime, final int rowCount) {
+    final List<IMeasurementSchema> schemas = new ArrayList<>();
+    schemas.add(new MeasurementSchema("region", TSDataType.STRING));
+    schemas.add(new MeasurementSchema("plant", TSDataType.STRING));
+    schemas.add(new MeasurementSchema("status", TSDataType.STRING));
+    schemas.add(new MeasurementSchema("s3", TSDataType.BOOLEAN));
+    final List<ColumnCategory> columnTypes =
+        Arrays.asList(
+            ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.ATTRIBUTE, ColumnCategory.FIELD);
+    final Tablet tablet =
+        new Tablet(
+            "multi_tablet_table3",
+            IMeasurementSchema.getMeasurementNameList(schemas),
+            IMeasurementSchema.getDataTypeList(schemas),
+            columnTypes,
+            rowCount);
+    for (int i = 0; i < rowCount; i++) {
+      final int rowIndex = tablet.getRowSize();
+      tablet.addTimestamp(rowIndex, startTime + i);
+      tablet.addValue("region", rowIndex, "r1");
+      tablet.addValue("plant", rowIndex, "p1");
+      tablet.addValue("status", rowIndex, "running");
+      tablet.addValue("s3", rowIndex, true);
+    }
+    return tablet;
+  }
+
+  private static void assertCount(
+      final ITableSession session, final String query, final long expectedCount)
+      throws IoTDBConnectionException, StatementExecutionException {
+    try (final SessionDataSet dataSet = session.executeQueryStatement(query)) {
+      final RowRecord record = dataSet.next();
+      assertEquals(expectedCount, record.getFields().get(0).getLongV());
+      assertFalse(dataSet.hasNext());
     }
   }
 }

iotdb-client/isession/src/main/java/org/apache/iotdb/isession/ITableSession.java

@@ -24,6 +24,8 @@
 
 import org.apache.tsfile.write.record.Tablet;
 
+import java.util.List;
+
 /**
  * This interface defines a session for interacting with IoTDB tables. It supports operations such
  * as data insertion, executing queries, and closing the session. Implementations of this interface
@@ -46,6 +48,16 @@ public interface ITableSession extends AutoCloseable {
    */
   void insert(Tablet tablet) throws StatementExecutionException, IoTDBConnectionException;
 
+  /**
+   * Inserts multiple {@link Tablet}s into the database.
+   *
+   * @param tablets the tablets containing time-series data to be inserted.
+   * @throws StatementExecutionException if an error occurs while executing the statement.
+   * @throws IoTDBConnectionException if there is an issue with the IoTDB connection.
+   */
+  void insertTablets(List<Tablet> tablets)
+      throws StatementExecutionException, IoTDBConnectionException;
+
   /**
    * Executes a non-query SQL statement, such as a DDL or DML command.
    *

iotdb-client/isession/src/main/java/org/apache/iotdb/isession/SessionConfig.java

@@ -31,6 +31,7 @@ public class SessionConfig {
   public static final int DEFAULT_CONNECTION_TIMEOUT_MS = 0;
   public static final boolean DEFAULT_REDIRECTION_MODE = true;
   public static final boolean DEFAULT_RECORDS_AUTO_CONVERT_TABLET = true;
+  public static final boolean DEFAULT_ENABLE_MERGE_TABLETS = true;
   public static final int CPU_CORES = Runtime.getRuntime().availableProcessors();
   public static final int DEFAULT_SESSION_EXECUTOR_THREAD_NUM = 2 * CPU_CORES;
   public static final int DEFAULT_SESSION_EXECUTOR_TASK_NUM = 1_000;

iotdb-client/session/src/main/java/org/apache/iotdb/session/AbstractSessionBuilder.java

@@ -53,6 +53,7 @@ public abstract class AbstractSessionBuilder {
   // datanode while encountering retriable errors in current DataNode
   public boolean enableRedirection = SessionConfig.DEFAULT_REDIRECTION_MODE;
   public boolean enableRecordsAutoConvertTablet = SessionConfig.DEFAULT_RECORDS_AUTO_CONVERT_TABLET;
+  public boolean enableMergeTablets = SessionConfig.DEFAULT_ENABLE_MERGE_TABLETS;
   public Version version = SessionConfig.DEFAULT_VERSION;
   public long timeOut = SessionConfig.DEFAULT_QUERY_TIME_OUT;