[Feature](Prepared Statement) fix and enable enable_server_side_prepared_statement by default #36581 (#36818)

picked from #36581

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/PrepareCommand.java

@@ -20,6 +20,7 @@ package org.apache.doris.nereids.trees.plans.commands;
 import org.apache.doris.mysql.MysqlCommand;
 import org.apache.doris.nereids.trees.expressions.Placeholder;
 import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.qe.ConnectContext;
@@ -102,6 +103,10 @@ public class PrepareCommand extends Command {
             LOG.debug("add prepared statement {}, isBinaryProtocol {}",
                     name, ctx.getCommand() == MysqlCommand.COM_STMT_PREPARE);
         }
+        if (logicalPlan instanceof InsertIntoTableCommand
+                    && ((InsertIntoTableCommand) logicalPlan).getLabelName().isPresent()) {
+            throw new org.apache.doris.common.UserException("Only support prepare InsertStmt without label now");
+        }
         ctx.addPreparedStatementContext(name,
                 new PreparedStatementContext(this, ctx, ctx.getStatementContext(), name));
         if (ctx.getCommand() == MysqlCommand.COM_STMT_PREPARE) {

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

@@ -35,6 +35,7 @@ import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.Type;
 import org.apache.doris.cluster.ClusterNamespace;
 import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.datasource.CatalogIf;
@@ -394,12 +395,18 @@ public class ConnectContext {
         this.preparedStmtCtxs.put(stmtName, ctx);
     }
 
-    public void addPreparedStatementContext(String stmtName, PreparedStatementContext ctx) {
+    public void addPreparedStatementContext(String stmtName, PreparedStatementContext ctx) throws UserException {
+        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."
+                    + "set max_prepared_stmt_count with larger number than " + sessionVariable.maxPreparedStmtCount);
+        }
         this.preparedStatementContextMap.put(stmtName, ctx);
     }
 
     public void removePrepareStmt(String stmtName) {
         this.preparedStmtCtxs.remove(stmtName);
+        this.preparedStatementContextMap.remove(stmtName);
     }
 
     public PrepareStmtContext getPreparedStmt(String stmtName) {

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

@@ -227,7 +227,8 @@ public abstract class ConnectProcessor {
         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
+        boolean nereidsUseServerPrep = (sessionVariable.enableServeSidePreparedStatement
+                    && !sessionVariable.isEnableInsertGroupCommit())
                         || mysqlCommand == MysqlCommand.COM_QUERY;
         if (nereidsUseServerPrep && sessionVariable.isEnableNereidsPlanner()) {
             if (wantToParseSqlFromSqlCache) {

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

@@ -20,6 +20,8 @@ package org.apache.doris.qe;
 import org.apache.doris.analysis.BinaryPredicate;
 import org.apache.doris.analysis.Expr;
 import org.apache.doris.analysis.LiteralExpr;
+import org.apache.doris.analysis.SlotRef;
+import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.Status;
@@ -149,7 +151,13 @@ public class PointQueryExecutor implements CoordInterface {
         KeyTuple.Builder kBuilder = KeyTuple.newBuilder();
         for (Expr expr : shortCircuitQueryContext.scanNode.getConjuncts()) {
             BinaryPredicate predicate = (BinaryPredicate) expr;
-            kBuilder.addKeyColumnRep(predicate.getChild(1).getStringValue());
+            Expr left = predicate.getChild(0);
+            Expr right = predicate.getChild(1);
+            // ignore delete sign conjuncts only collect key conjuncts
+            if (left instanceof SlotRef && ((SlotRef) left).getColumnName().equalsIgnoreCase(Column.DELETE_SIGN)) {
+                continue;
+            }
+            kBuilder.addKeyColumnRep(right.getStringValue());
         }
         requestBuilder.addKeyTuples(kBuilder);
     }

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

@@ -134,6 +134,7 @@ public class SessionVariable implements Serializable, Writable {
     public static final String ENABLE_EXCHANGE_NODE_PARALLEL_MERGE = "enable_exchange_node_parallel_merge";
 
     public static final String ENABLE_SERVER_SIDE_PREPARED_STATEMENT = "enable_server_side_prepared_statement";
+    public static final String MAX_PREPARED_STMT_COUNT = "max_prepared_stmt_count";
     public static final String PREFER_JOIN_METHOD = "prefer_join_method";
 
     public static final String ENABLE_FOLD_CONSTANT_BY_BE = "enable_fold_constant_by_be";
@@ -1361,7 +1362,12 @@ public class SessionVariable implements Serializable, Writable {
 
     @VariableMgr.VarAttr(name = ENABLE_SERVER_SIDE_PREPARED_STATEMENT, needForward = true, description = {
             "是否启用开启服务端prepared statement", "Set whether to enable server side prepared statement."})
-    public boolean enableServeSidePreparedStatement = false;
+    public boolean enableServeSidePreparedStatement = true;
+
+    @VariableMgr.VarAttr(name = MAX_PREPARED_STMT_COUNT,  flag = VariableMgr.GLOBAL,
+            needForward = true, description = {
+                "服务端prepared statement最大个数", "the maximum prepared statements server holds."})
+    public int maxPreparedStmtCount = 100000;
 
     // Default value is false, which means the group by and having clause
     // should first use column name not alias. According to mysql.