fix(spi): harden SQL execution and connection lifecycle

chat2db-community-server/chat2db-community-plugins/chat2db-community-mysql/src/main/java/ai/chat2db/plugin/mysql/enums/type/MysqlColumnTypeEnum.java

@@ -299,7 +299,7 @@ private String buildExt(TableColumn column, MysqlColumnTypeEnum type) {
         if (!type.columnType.isSupportExtent() || StringUtils.isEmpty(column.getExtent())) {
             return "";
         }
-        return column.getComment();
+        return column.getExtent();
     }
 
     private String buildDefaultValue(TableColumn column, MysqlColumnTypeEnum type) {

chat2db-community-server/chat2db-community-plugins/chat2db-community-mysql/src/test/java/ai/chat2db/plugin/mysql/builder/MysqlSqlBuilderTest.java

@@ -14,6 +14,7 @@
 import ai.chat2db.community.domain.api.model.result.ResultOperation;
 import ai.chat2db.spi.model.datasource.ConnectInfo;
 import ai.chat2db.spi.sql.Chat2DBContext;
+import ai.chat2db.plugin.mysql.enums.type.MysqlColumnTypeEnum;
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
@@ -24,6 +25,20 @@
 
 class MysqlSqlBuilderTest {
 
+    @Test
+    void shouldUseEnumExtentInsteadOfColumnComment() {
+        TableColumn column = TableColumn.builder()
+                .name("status")
+                .columnType("ENUM")
+                .extent("('draft','published')")
+                .comment("workflow status")
+                .build();
+
+        String sql = MysqlColumnTypeEnum.ENUM.buildCreateColumnSql(column);
+
+        assertTrue(sql.contains("('draft','published')"), sql);
+    }
+
     @Test
     void shouldBuildCreateDatabaseWithCharsetAndCollation() {
         MysqlSqlBuilder builder = new MysqlSqlBuilder();

chat2db-community-server/chat2db-community-plugins/chat2db-community-sqlserver/src/main/java/ai/chat2db/plugin/sqlserver/builder/SqlServerSqlBuilder.java

@@ -294,7 +294,7 @@ protected void buildTableName(String databaseName, String schemaName, String tab
         if (StringUtils.isNotBlank(databaseName)) {
             script.append(SQLConstants.OPEN_SQUARE_BRACKET).append(databaseName).append(SQLConstants.CLOSE_SQUARE_BRACKET).append('.');
         }
-        if (StringUtils.isNotBlank(databaseName)) {
+        if (StringUtils.isNotBlank(schemaName)) {
             script.append(SQLConstants.OPEN_SQUARE_BRACKET).append(schemaName).append(SQLConstants.CLOSE_SQUARE_BRACKET).append('.');
         }
         if (!(tableName.startsWith(SQLConstants.OPEN_SQUARE_BRACKET) && tableName.endsWith(SQLConstants.CLOSE_SQUARE_BRACKET))) {

chat2db-community-server/chat2db-community-plugins/chat2db-community-sqlserver/src/test/java/ai/chat2db/plugin/sqlserver/builder/SqlServerSqlBuilderTest.java

@@ -29,6 +29,14 @@ void shouldUseTopWhenLimitingSingleRowDeleteAndUpdate() {
                 builder.appendSingleRowLimit("UPDATE", "[t]", where, "UPDATE [t] set [a] = 1" + where));
     }
 
+    @Test
+    void shouldIncludeSchemaWhenDatabaseNameIsBlank() {
+        ExposedSqlServerSqlBuilder builder = new ExposedSqlServerSqlBuilder();
+
+        assertEquals("[dbo].[orders]", builder.tableName(null, "dbo", "orders"));
+        assertEquals("[analytics].[dbo].[orders]", builder.tableName("analytics", "dbo", "orders"));
+    }
+
     @Test
     void shouldQuoteAndEscapeQualifiedViewName() {
         SqlServerSqlBuilder builder = new SqlServerSqlBuilder();
@@ -45,4 +53,12 @@ void shouldQuoteAndEscapeQualifiedViewName() {
                         + "'order] schema', 'VIEW', 'select] view'",
                 builder.buildCreateView(view));
     }
+
+    private static final class ExposedSqlServerSqlBuilder extends SqlServerSqlBuilder {
+        private String tableName(String databaseName, String schemaName, String tableName) {
+            StringBuilder script = new StringBuilder();
+            buildTableName(databaseName, schemaName, tableName, script);
+            return script.toString();
+        }
+    }
 }

chat2db-community-server/chat2db-community-spi/src/main/java/ai/chat2db/spi/DefaultDBManager.java

@@ -126,12 +126,14 @@ public Connection getConnection(ConnectInfo connectInfo) {
                 try {
                     connection.setCatalog(connectInfo.getDatabaseName());
                 } catch (SQLException e) {
+                    log.warn("Failed to set catalog to '{}': {}", connectInfo.getDatabaseName(), e.getMessage());
                 }
             }
             if (StringUtils.isNotBlank(connectInfo.getSchemaName())) {
                 try {
                     connection.setSchema(connectInfo.getSchemaName());
                 } catch (SQLException e) {
+                    log.warn("Failed to set schema to '{}': {}", connectInfo.getSchemaName(), e.getMessage());
                 }
             }
         }

chat2db-community-server/chat2db-community-spi/src/main/java/ai/chat2db/spi/DefaultSQLExecutor.java

@@ -46,6 +46,7 @@
 
 import java.sql.*;
 import java.util.*;
+import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -57,6 +58,8 @@ public class DefaultSQLExecutor implements ICommandExecutor {
 
     private static final int STREAMING_ROW_BATCH_SIZE = 200;
 
+    private static final Executor DIRECT_ABORT_EXECUTOR = Runnable::run;
+
 
     private static final DefaultSQLExecutor INSTANCE = new DefaultSQLExecutor();
 
@@ -280,9 +283,10 @@ public Long count(String sql, Connection connection) throws SQLException {
             boolean query = stmt.execute();
             if (query) {
                 long n = 0;
-                ResultSet rs = stmt.getResultSet();
-                while (rs.next()) {
-                    n++;
+                try (ResultSet rs = stmt.getResultSet()) {
+                    while (rs.next()) {
+                        n++;
+                    }
                 }
                 return n;
             } else {
@@ -1476,19 +1480,83 @@ public void fetchAllTableRecords(FetchAllTableRecordsRequest request) {
         String sql = request.getSql();
         int batchSize = request.getBatchSize();
         IResultSetConsumer consumer = request.getConsumer();
+        final boolean manageTransaction;
+        try {
+            manageTransaction = connection.getAutoCommit();
+        } catch (SQLException ex) {
+            throw new RuntimeException("Failed to read original autoCommit", ex);
+        }
+        boolean transactionStarted = false;
+        boolean transactionResolved = false;
+        boolean discardRequired = false;
+        Exception failure = null;
         try (PreparedStatement stmt = connection.prepareStatement(sql,
                 ResultSet.TYPE_FORWARD_ONLY,
                 ResultSet.CONCUR_READ_ONLY
         )) {
-            connection.setAutoCommit(false);
+            if (manageTransaction) {
+                transactionStarted = true;
+                connection.setAutoCommit(false);
+            }
             stmt.setFetchSize(batchSize);
-            ResultSet rs = stmt.executeQuery();
-            consumer.accept(rs);
-            connection.commit();
-            connection.setAutoCommit(true);
+            try (ResultSet rs = stmt.executeQuery()) {
+                consumer.accept(rs);
+            }
+            if (transactionStarted) {
+                connection.commit();
+                transactionResolved = true;
+            }
         } catch (Exception e) {
-            log.error("Failed to fetch table records. Query: {}", sql, e);
-            throw new RuntimeException(e);
+            failure = e;
+        }
+
+        if (failure != null && transactionStarted && !transactionResolved) {
+            try {
+                connection.rollback();
+                transactionResolved = true;
+            } catch (Exception rollbackEx) {
+                failure.addSuppressed(rollbackEx);
+                discardRequired = true;
+            }
+        }
+
+        if (transactionStarted && !discardRequired) {
+            try {
+                connection.setAutoCommit(true);
+            } catch (Exception restoreEx) {
+                if (failure == null) {
+                    failure = restoreEx;
+                } else {
+                    failure.addSuppressed(restoreEx);
+                }
+                discardRequired = true;
+            }
+        }
+
+        if (discardRequired) {
+            discardConnection(connection, failure);
+        }
+        if (failure != null) {
+            log.error("Failed to fetch table records. Query: {}", sql, failure);
+            throw new RuntimeException(failure);
+        }
+    }
+
+    private void discardConnection(Connection connection, Throwable failure) {
+        ConnectInfo connectInfo = Chat2DBContext.getConnectInfo();
+        if (connectInfo != null && connectInfo.getConnection() == connection) {
+            connectInfo.setConnection(null);
+        }
+        try {
+            connection.abort(DIRECT_ABORT_EXECUTOR);
+            return;
+        } catch (Exception abortEx) {
+            failure.addSuppressed(abortEx);
+        }
+        try {
+            connection.close();
+        } catch (Exception closeEx) {
+            failure.addSuppressed(closeEx);
         }
     }
 

chat2db-community-server/chat2db-community-spi/src/main/java/ai/chat2db/spi/model/datasource/ConnectInfo.java

@@ -139,6 +139,9 @@ public void setProject(String project) {
 
     private Date lastAccessTime;
 
+    // Runtime-only marker used to reject leases created before a pool invalidation.
+    private transient volatile Long poolGeneration;
+
 
     public String getDbVersion() {
         return dbVersion;
@@ -213,7 +216,7 @@ public boolean equals(Object o) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(dataSourceId, consoleId, databaseName);
+        return Objects.hash(dataSourceId, gmtModified);
     }
 
     public Long getDataSourceId() {
@@ -490,6 +493,14 @@ public void setLastAccessTime(Date lastAccessTime) {
         this.lastAccessTime = lastAccessTime;
     }
 
+    public Long poolGeneration() {
+        return poolGeneration;
+    }
+
+    public void updatePoolGeneration(Long poolGeneration) {
+        this.poolGeneration = poolGeneration;
+    }
+
     private AtomicBoolean inUse = new AtomicBoolean(false);
 
     public boolean trySetInUse() {

chat2db-community-server/chat2db-community-spi/src/main/java/ai/chat2db/spi/sql/Chat2DBContext.java

@@ -48,12 +48,23 @@ public class Chat2DBContext {
         }
     }
 
+    private static IPlugin getPlugin(String dbType) {
+        if (StringUtils.isBlank(dbType)) {
+            throw new IllegalArgumentException("Database type must not be blank. Registered types: " + PLUGIN_MAP.keySet());
+        }
+        IPlugin plugin = PLUGIN_MAP.get(dbType);
+        if (plugin == null) {
+            throw new IllegalArgumentException("Unsupported database type: " + dbType + ". Registered types: " + PLUGIN_MAP.keySet());
+        }
+        return plugin;
+    }
+
     public static DriverConfig getDefaultDriverConfig(String dbType) {
-        return PLUGIN_MAP.get(dbType).getDBConfig().getDefaultDriverConfig();
+        return getPlugin(dbType).getDBConfig().getDefaultDriverConfig();
     }
 
     public static ISqlBuilder getSqlBuilder() {
-        return PLUGIN_MAP.get(getConnectInfo().getDbType()).getDbMetaData().getSqlBuilder();
+        return getPlugin(getConnectInfo().getDbType()).getDbMetaData().getSqlBuilder();
     }
 
 
@@ -62,34 +73,34 @@ public static ConnectInfo getConnectInfo() {
     }
 
     public static IDbMetaData getDbMetaData() {
-        return PLUGIN_MAP.get(getConnectInfo().getDbType()).getDbMetaData();
+        return getPlugin(getConnectInfo().getDbType()).getDbMetaData();
     }
 
     public static IDbMetaData getDbMetaData(String dbType) {
         if (StringUtils.isBlank(dbType)) {
             return getDbMetaData();
         }
-        return PLUGIN_MAP.get(dbType).getDbMetaData();
+        return getPlugin(dbType).getDbMetaData();
     }
 
     public static DBConfig getDBConfig(String dbType) {
-        return PLUGIN_MAP.get(dbType).getDBConfig();
+        return getPlugin(dbType).getDBConfig();
     }
 
     public static DBConfig getDBConfig() {
         ConnectInfo connectInfo = getConnectInfo();
         if (connectInfo == null) {
             return null;
         }
-        return PLUGIN_MAP.get(connectInfo.getDbType()).getDBConfig();
+        return getPlugin(connectInfo.getDbType()).getDBConfig();
     }
 
     public static IDbManager getDbManager() {
-        return PLUGIN_MAP.get(getConnectInfo().getDbType()).getDbManager();
+        return getPlugin(getConnectInfo().getDbType()).getDbManager();
     }
 
     public static IDbManager getDbManager(String dbType) {
-        return PLUGIN_MAP.get(dbType).getDbManager();
+        return getPlugin(dbType).getDbManager();
     }
 
     public static IAccountManager getAccountManager() {