Load: Fix the Session exit issue after type conversion (#16037)

* Load: Fix the Session exit issue after type conversion

* fix

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/SessionManager.java

@@ -411,6 +411,16 @@ public SessionInfo copySessionInfoForTreeModel(final SessionInfo sessionInfo) {
         IClientSession.SqlDialect.TREE);
   }
 
+  public SessionInfo getSessionInfoOfTreeModel(IClientSession session) {
+    return new SessionInfo(
+        session.getId(),
+        session.getUsername(),
+        ZoneId.systemDefault(),
+        session.getClientVersion(),
+        session.getDatabaseName(),
+        IClientSession.SqlDialect.TREE);
+  }
+
   public SessionInfo getSessionInfoOfTableModel(IClientSession session) {
     return new SessionInfo(
         session.getId(),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java

@@ -89,18 +89,24 @@ public Optional<TSStatus> convertForTableModel(final LoadTsFile loadTsFileTableS
   }
 
   private TSStatus executeForTableModel(final Statement statement, final String databaseName) {
-    final IClientSession session =
-        new InternalClientSession(
-            String.format(
-                "%s_%s",
-                LoadTsFileDataTypeConverter.class.getSimpleName(),
-                Thread.currentThread().getName()));
-    session.setUsername(AuthorityChecker.SUPER_USER);
-    session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0);
-    session.setZoneId(ZoneId.systemDefault());
-    session.setSqlDialect(IClientSession.SqlDialect.TABLE);
-
-    SESSION_MANAGER.registerSession(session);
+    final IClientSession session;
+    final boolean needToCreateSession = SESSION_MANAGER.getCurrSession() == null;
+    if (needToCreateSession) {
+      session =
+          new InternalClientSession(
+              String.format(
+                  "%s_%s",
+                  LoadTsFileDataTypeConverter.class.getSimpleName(),
+                  Thread.currentThread().getName()));
+      session.setUsername(AuthorityChecker.SUPER_USER);
+      session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0);
+      session.setZoneId(ZoneId.systemDefault());
+      session.setSqlDialect(IClientSession.SqlDialect.TABLE);
+
+      SESSION_MANAGER.registerSession(session);
+    } else {
+      session = SESSION_MANAGER.getCurrSession();
+    }
     try {
       return Coordinator.getInstance()
           .executeForTableModel(
@@ -114,7 +120,9 @@ private TSStatus executeForTableModel(final Statement statement, final String da
               IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold())
           .status;
     } finally {
-      SESSION_MANAGER.removeCurrSession();
+      if (needToCreateSession) {
+        SESSION_MANAGER.removeCurrSession();
+      }
     }
   }
 
@@ -134,17 +142,42 @@ public Optional<TSStatus> convertForTreeModel(final LoadTsFileStatement loadTsFi
   }
 
   private TSStatus executeForTreeModel(final Statement statement) {
-    return Coordinator.getInstance()
-        .executeForTreeModel(
-            isGeneratedByPipe ? new PipeEnrichedStatement(statement) : statement,
-            SESSION_MANAGER.requestQueryId(),
-            SESSION_MANAGER.getSessionInfo(SESSION_MANAGER.getCurrSession()),
-            "",
-            ClusterPartitionFetcher.getInstance(),
-            ClusterSchemaFetcher.getInstance(),
-            IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold(),
-            false)
-        .status;
+    final IClientSession session;
+    final boolean needToCreateSession = SESSION_MANAGER.getCurrSession() == null;
+    if (needToCreateSession) {
+      session =
+          new InternalClientSession(
+              String.format(
+                  "%s_%s",
+                  LoadTsFileDataTypeConverter.class.getSimpleName(),
+                  Thread.currentThread().getName()));
+      session.setUsername(AuthorityChecker.SUPER_USER);
+      session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0);
+      session.setZoneId(ZoneId.systemDefault());
+      session.setSqlDialect(IClientSession.SqlDialect.TREE);
+
+      SESSION_MANAGER.registerSession(session);
+    } else {
+      session = SESSION_MANAGER.getCurrSession();
+    }
+
+    try {
+      return Coordinator.getInstance()
+          .executeForTreeModel(
+              isGeneratedByPipe ? new PipeEnrichedStatement(statement) : statement,
+              SESSION_MANAGER.requestQueryId(),
+              SESSION_MANAGER.getSessionInfoOfTreeModel(session),
+              "",
+              ClusterPartitionFetcher.getInstance(),
+              ClusterSchemaFetcher.getInstance(),
+              IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold(),
+              false)
+          .status;
+    } finally {
+      if (needToCreateSession) {
+        SESSION_MANAGER.removeCurrSession();
+      }
+    }
   }
 
   public boolean isSuccessful(final TSStatus status) {