[Pick](PreparedStatement) allow prepare mysql command in Nerids even … (#39960)

…if enableServeSidePreparedStatement is false to avoid fallback to
legacy planner

backport #39465

fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java

@@ -412,6 +412,9 @@ public class ConnectContext {
     }
 
     public void addPreparedStatementContext(String stmtName, PreparedStatementContext ctx) throws UserException {
+        if (!sessionVariable.enableServeSidePreparedStatement) {
+            throw new UserException("Failed to do prepared command, server side prepared statement is disabled");
+        }
         if (this.preparedStatementContextMap.size() > sessionVariable.maxPreparedStmtCount) {
             throw new UserException("Failed to create a server prepared statement"
                     + "possibly because there are too many active prepared statements on server already."

fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java

@@ -227,13 +227,8 @@ public abstract class ConnectProcessor {
         Exception nereidsParseException = null;
         long parseSqlStartTime = System.currentTimeMillis();
         List<StatementBase> cachedStmts = null;
-        // Currently we add a config to decide whether using PREPARED/EXECUTE command for nereids
-        // TODO: after implemented full prepared, we could remove this flag
-        boolean nereidsUseServerPrep = (sessionVariable.enableServeSidePreparedStatement
-                    && !sessionVariable.isEnableInsertGroupCommit())
-                        || mysqlCommand == MysqlCommand.COM_QUERY;
         CacheKeyType cacheKeyType = null;
-        if (nereidsUseServerPrep && sessionVariable.isEnableNereidsPlanner()) {
+        if (!sessionVariable.isEnableInsertGroupCommit() && sessionVariable.isEnableNereidsPlanner()) {
             if (wantToParseSqlFromSqlCache) {
                 cachedStmts = parseFromSqlCache(originStmt);
                 Optional<SqlCacheContext> sqlCacheContext = ConnectContext.get()

regression-test/suites/prepared_stmt_p0/prepared_stmt.groovy

@@ -58,6 +58,7 @@ suite("test_prepared_stmt", "nonConcurrent") {
         qt_sql """select * from  ${tableName} order by 1, 2, 3"""
         qt_sql """select * from  ${tableName} order by 1, 2, 3"""
         sql "set global max_prepared_stmt_count = 10000"
+        sql "set enable_fallback_to_original_planner = false"
 
         def stmt_read = prepareStatement "select * from ${tableName} where k1 = ? order by k1"
         assertEquals(stmt_read.class, com.mysql.cj.jdbc.ServerPreparedStatement);
@@ -239,5 +240,11 @@ suite("test_prepared_stmt", "nonConcurrent") {
         // not stable
         // qe_select16 stmt_read
         stmt_read.close()
+
+        stmt_read = prepareStatement "SELECT connection_id()" 
+        assertEquals(com.mysql.cj.jdbc.ServerPreparedStatement, stmt_read.class)
+        result = stmt_read.execute()
+        logger.info("connection_id: ${result}")
+        // qe_select16 stmt_read
     }
 }