Add insertTablets(List<Tablet>) and insertAlignedTablets(List<Tablet>) to Session interface

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

@@ -367,6 +367,22 @@ void insertAlignedTablets(Map<String, Tablet> tablets)
   void insertAlignedTablets(Map<String, Tablet> tablets, boolean sorted)
       throws IoTDBConnectionException, StatementExecutionException;
 
+  /**
+   * Inserts several tablets in one request. Unlike {@link #insertTablets(Map)}, several tablets may
+   * belong to the same device, e.g. one tablet per measurement for measurements with different
+   * timestamps. With redirection enabled, each tablet is routed by its own device id.
+   */
+  void insertTablets(List<Tablet> tablets)
+      throws IoTDBConnectionException, StatementExecutionException;
+
+  /**
+   * Inserts several tablets of aligned devices in one request. Unlike {@link
+   * #insertAlignedTablets(Map)}, several tablets may belong to the same device. With redirection
+   * enabled, each tablet is routed by its own device id.
+   */
+  void insertAlignedTablets(List<Tablet> tablets)
+      throws IoTDBConnectionException, StatementExecutionException;
+
   void testInsertTablet(Tablet tablet) throws IoTDBConnectionException, StatementExecutionException;
 
   void testInsertTablet(Tablet tablet, boolean sorted)

iotdb-client/isession/src/main/java/org/apache/iotdb/isession/pool/ISessionPool.java

@@ -73,6 +73,22 @@ void insertTablets(Map<String, Tablet> tablets, boolean sorted)
   void insertAlignedTablets(Map<String, Tablet> tablets, boolean sorted)
       throws IoTDBConnectionException, StatementExecutionException;
 
+  /**
+   * Inserts several tablets in one request. Unlike {@link #insertTablets(Map)}, several tablets may
+   * belong to the same device, e.g. one tablet per measurement for measurements with different
+   * timestamps. With redirection enabled, each tablet is routed by its own device id.
+   */
+  void insertTablets(List<Tablet> tablets)
+      throws IoTDBConnectionException, StatementExecutionException;
+
+  /**
+   * Inserts several tablets of aligned devices in one request. Unlike {@link
+   * #insertAlignedTablets(Map)}, several tablets may belong to the same device. With redirection
+   * enabled, each tablet is routed by its own device id.
+   */
+  void insertAlignedTablets(List<Tablet> tablets)
+      throws IoTDBConnectionException, StatementExecutionException;
+
   void insertRecords(
       List<String> deviceIds,
       List<Long> times,

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

@@ -3207,17 +3207,9 @@ public void insertTablets(Map<String, Tablet> tablets)
   public void insertTablets(Map<String, Tablet> tablets, boolean sorted)
       throws IoTDBConnectionException, StatementExecutionException {
     if (enableRedirection) {
-      insertTabletsWithLeaderCache(tablets, sorted, false);
+      insertTabletsWithLeaderCache(tabletMapToList(tablets), sorted, false);
     } else {
-      TSInsertTabletsReq request =
-          genTSInsertTabletsReq(new ArrayList<>(tablets.values()), sorted, false);
-      if (request == null) {
-        return;
-      }
-      try {
-        getDefaultSessionConnection().insertTablets(request);
-      } catch (RedirectException ignored) {
-      }
+      insertTabletsWithDefaultConnection(new ArrayList<>(tablets.values()), sorted, false);
     }
   }
 
@@ -3246,29 +3238,86 @@ public void insertAlignedTablets(Map<String, Tablet> tablets)
   public void insertAlignedTablets(Map<String, Tablet> tablets, boolean sorted)
       throws IoTDBConnectionException, StatementExecutionException {
     if (enableRedirection) {
-      insertTabletsWithLeaderCache(tablets, sorted, true);
+      insertTabletsWithLeaderCache(tabletMapToList(tablets), sorted, true);
     } else {
-      TSInsertTabletsReq request =
-          genTSInsertTabletsReq(new ArrayList<>(tablets.values()), sorted, true);
-      if (request == null) {
-        return;
-      }
-      try {
-        getDefaultSessionConnection().insertTablets(request);
-      } catch (RedirectException ignored) {
+      insertTabletsWithDefaultConnection(new ArrayList<>(tablets.values()), sorted, true);
+    }
+  }
+
+  /**
+   * Inserts several tablets of possibly the same device in one request. With redirection enabled,
+   * each tablet is routed by its own device id, so tablets whose devices have the same leader are
+   * sent to that leader together.
+   *
+   * @param tablets data batch, possibly several tablets per device
+   */
+  @Override
+  public void insertTablets(List<Tablet> tablets)
+      throws IoTDBConnectionException, StatementExecutionException {
+    insertTabletList(tablets, false);
+  }
+
+  /**
+   * Inserts several tablets of possibly the same aligned device in one request. With redirection
+   * enabled, each tablet is routed by its own device id, so tablets whose devices have the same
+   * leader are sent to that leader together.
+   *
+   * @param tablets data batch, possibly several tablets per device
+   */
+  @Override
+  public void insertAlignedTablets(List<Tablet> tablets)
+      throws IoTDBConnectionException, StatementExecutionException {
+    insertTabletList(tablets, true);
+  }
+
+  private void insertTabletList(List<Tablet> tablets, boolean isAligned)
+      throws IoTDBConnectionException, StatementExecutionException {
+    if (enableRedirection) {
+      List<KeyedTablet> keyedTablets = new ArrayList<>(tablets.size());
+      for (Tablet tablet : tablets) {
+        keyedTablets.add(new KeyedTablet(tablet.getDeviceId(), tablet));
       }
+      insertTabletsWithLeaderCache(keyedTablets, false, isAligned);
+    } else {
+      insertTabletsWithDefaultConnection(tablets, false, isAligned);
     }
   }
 
+  private void insertTabletsWithDefaultConnection(
+      List<Tablet> tablets, boolean sorted, boolean isAligned)
+      throws IoTDBConnectionException, StatementExecutionException {
+    TSInsertTabletsReq request = genTSInsertTabletsReq(tablets, sorted, isAligned);
+    if (request == null) {
+      return;
+    }
+    try {
+      getDefaultSessionConnection().insertTablets(request);
+    } catch (RedirectException ignored) {
+    }
+  }
+
+  /** A tablet and the key under which its leader is looked up in the leader cache. */
+  private record KeyedTablet(String leaderCacheKey, Tablet tablet) {}
+
+  /** Keys each tablet by its map key, which callers of the map methods may choose freely. */
+  private static List<KeyedTablet> tabletMapToList(Map<String, Tablet> tablets) {
+    return tablets.entrySet().stream()
+        .map(entry -> new KeyedTablet(entry.getKey(), entry.getValue()))
+        .toList();
+  }
+
+  /**
+   * @param tablets the tablets to insert; the same leader cache key may occur several times
+   */
   private void insertTabletsWithLeaderCache(
-      Map<String, Tablet> tablets, boolean sorted, boolean isAligned)
+      List<KeyedTablet> tablets, boolean sorted, boolean isAligned)
       throws IoTDBConnectionException, StatementExecutionException {
     Map<SessionConnection, TSInsertTabletsReq> tabletGroup = new HashMap<>();
-    for (Entry<String, Tablet> entry : tablets.entrySet()) {
-      final SessionConnection connection = getSessionConnection(entry.getKey());
+    for (KeyedTablet keyedTablet : tablets) {
+      final SessionConnection connection = getSessionConnection(keyedTablet.leaderCacheKey());
       TSInsertTabletsReq request =
           tabletGroup.computeIfAbsent(connection, k -> new TSInsertTabletsReq());
-      updateTSInsertTabletsReq(request, entry.getValue(), sorted, isAligned);
+      updateTSInsertTabletsReq(request, keyedTablet.tablet(), sorted, isAligned);
     }
 
     tabletGroup.entrySet().removeIf(e -> e.getValue().getPrefixPathsSize() == 0);

iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java

@@ -1113,6 +1113,63 @@ public void insertAlignedTablets(Map<String, Tablet> tablets, boolean sorted)
     }
   }
 
+  /**
+   * Inserts several tablets in one request; several tablets may belong to the same device.
+   *
+   * @param tablets data batch, possibly several tablets per device
+   * @see Session#insertTablets(List)
+   */
+  @SuppressWarnings({"squid:S112"}) // ignore Generic exceptions should never be thrown
+  @Override
+  public void insertTablets(List<Tablet> tablets)
+      throws IoTDBConnectionException, StatementExecutionException {
+    ISession session = getSession();
+    try {
+      session.insertTablets(tablets);
+      putBack(session);
+    } catch (IoTDBConnectionException e) {
+      // TException means the connection is broken, remove it and get a new one.
+      LOGGER.warn(SessionMessages.INSERT_TABLETS_FAILED, e);
+      cleanSessionAndMayThrowConnectionException(session, FINAL_RETRY, e);
+    } catch (StatementExecutionException | RuntimeException e) {
+      putBack(session);
+      throw e;
+    } catch (Throwable e) {
+      LOGGER.error(SessionMessages.UNEXPECTED_ERROR_IN_INSERT_TABLETS, e);
+      putBack(session);
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Inserts several tablets of aligned devices in one request; several tablets may belong to the
+   * same device.
+   *
+   * @param tablets data batch, possibly several tablets per device
+   * @see Session#insertAlignedTablets(List)
+   */
+  @SuppressWarnings({"squid:S112"}) // ignore Generic exceptions should never be thrown
+  @Override
+  public void insertAlignedTablets(List<Tablet> tablets)
+      throws IoTDBConnectionException, StatementExecutionException {
+    ISession session = getSession();
+    try {
+      session.insertAlignedTablets(tablets);
+      putBack(session);
+    } catch (IoTDBConnectionException e) {
+      // TException means the connection is broken, remove it and get a new one.
+      LOGGER.warn(SessionMessages.INSERT_ALIGNED_TABLETS_FAILED, e);
+      cleanSessionAndMayThrowConnectionException(session, FINAL_RETRY, e);
+    } catch (StatementExecutionException | RuntimeException e) {
+      putBack(session);
+      throw e;
+    } catch (Throwable e) {
+      LOGGER.error(SessionMessages.UNEXPECTED_ERROR_IN_INSERT_ALIGNED_TABLETS, e);
+      putBack(session);
+      throw new RuntimeException(e);
+    }
+  }
+
   /**
    * Insert data in batch format, which can reduce the overhead of network. This method is just like
    * jdbc batch insert, we pack some insert request in batch and send them to server If you want