[fix](group commit) support full prepare of group commit in nereids (#…

…46533)

fe/fe-core/src/main/java/org/apache/doris/analysis/NativeInsertStmt.java

@@ -154,7 +154,6 @@ public class NativeInsertStmt extends InsertStmt {
 
     // Used for group commit insert
     private boolean isGroupCommit = false;
-    private int baseSchemaVersion = -1;
     private TUniqueId loadId = null;
     private long tableId = -1;
     public boolean isGroupCommitStreamLoadSql = false;
@@ -1304,7 +1303,8 @@ public GroupCommitPlanner planForGroupCommit(TUniqueId queryId) throws UserExcep
         OlapTable olapTable = (OlapTable) getTargetTable();
         olapTable.readLock();
         try {
-            if (groupCommitPlanner != null && olapTable.getBaseSchemaVersion() == baseSchemaVersion) {
+            if (groupCommitPlanner != null
+                    && olapTable.getBaseSchemaVersion() == groupCommitPlanner.baseSchemaVersion) {
                 if (LOG.isDebugEnabled()) {
                     LOG.debug("reuse group commit plan, table={}", olapTable);
                 }
@@ -1322,7 +1322,6 @@ public GroupCommitPlanner planForGroupCommit(TUniqueId queryId) throws UserExcep
                     targetColumnNames, queryId, ConnectContext.get().getSessionVariable().getGroupCommit());
             // save plan message to be reused for prepare stmt
             loadId = queryId;
-            baseSchemaVersion = olapTable.getBaseSchemaVersion();
             return groupCommitPlanner;
         } finally {
             olapTable.readUnlock();
@@ -1334,7 +1333,7 @@ public TUniqueId getLoadId() {
     }
 
     public int getBaseSchemaVersion() {
-        return baseSchemaVersion;
+        return groupCommitPlanner.baseSchemaVersion;
     }
 
     public void setIsFromDeleteOrUpdateStmt(boolean isFromDeleteOrUpdateStmt) {

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnvFactory.java

@@ -22,11 +22,9 @@
 import org.apache.doris.analysis.DescriptorTable;
 import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.catalog.CloudTabletStatMgr;
-import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.DynamicPartitionProperty;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.EnvFactory;
-import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.Partition;
 import org.apache.doris.catalog.Replica;
 import org.apache.doris.catalog.ReplicaAllocation;
@@ -37,12 +35,10 @@
 import org.apache.doris.cloud.load.CloudBrokerLoadJob;
 import org.apache.doris.cloud.load.CloudLoadManager;
 import org.apache.doris.cloud.load.CloudRoutineLoadManager;
-import org.apache.doris.cloud.planner.CloudGroupCommitPlanner;
 import org.apache.doris.cloud.qe.CloudCoordinator;
 import org.apache.doris.cloud.system.CloudSystemInfoService;
 import org.apache.doris.cloud.transaction.CloudGlobalTransactionMgr;
 import org.apache.doris.common.MetaNotFoundException;
-import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.MasterDaemon;
 import org.apache.doris.common.util.PropertyAnalyzer;
 import org.apache.doris.datasource.InternalCatalog;
@@ -51,7 +47,6 @@
 import org.apache.doris.load.loadv2.LoadManager;
 import org.apache.doris.load.routineload.RoutineLoadManager;
 import org.apache.doris.nereids.stats.StatsErrorEstimator;
-import org.apache.doris.planner.GroupCommitPlanner;
 import org.apache.doris.planner.PlanFragment;
 import org.apache.doris.planner.Planner;
 import org.apache.doris.planner.ScanNode;
@@ -62,8 +57,6 @@
 import org.apache.doris.thrift.TUniqueId;
 import org.apache.doris.transaction.GlobalTransactionMgrIface;
 
-import org.apache.thrift.TException;
-
 import java.lang.reflect.Type;
 import java.util.List;
 import java.util.Map;
@@ -168,12 +161,6 @@ public Coordinator createCoordinator(Long jobId, TUniqueId queryId, DescriptorTa
                                 enableProfile);
     }
 
-    @Override
-    public GroupCommitPlanner createGroupCommitPlanner(Database db, OlapTable table, List<String> targetColumnNames,
-            TUniqueId queryId, String groupCommit) throws UserException, TException {
-        return  new CloudGroupCommitPlanner(db, table, targetColumnNames, queryId, groupCommit);
-    }
-
     @Override
     public RoutineLoadManager createRoutineLoadManager() {
         return new CloudRoutineLoadManager();

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

@@ -24,7 +24,9 @@
 import org.apache.doris.nereids.glue.LogicalPlanAdapter;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.commands.insert.OlapGroupCommitInsertExecutor;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.planner.GroupCommitPlanner;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.PointQueryExecutor;
 import org.apache.doris.qe.PreparedStatementContext;
@@ -72,27 +74,35 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         executor.setParsedStmt(planAdapter);
         // If it's not a short circuit query or schema version is different(indicates schema changed) or
         // has nondeterministic functions in statement, then need to do reanalyze and plan
-        boolean isShortCircuit = executor.getContext().getStatementContext().isShortCircuitQuery();
-        boolean hasShortCircuitContext = preparedStmtCtx.shortCircuitQueryContext.isPresent();
-        boolean schemaVersionMismatch = hasShortCircuitContext
-                    && preparedStmtCtx.shortCircuitQueryContext.get().tbl.getBaseSchemaVersion()
-                    != preparedStmtCtx.shortCircuitQueryContext.get().schemaVersion;
-        boolean needAnalyze = !isShortCircuit || schemaVersionMismatch || !hasShortCircuitContext
-                        || executor.getContext().getStatementContext().hasNondeterministic();
-        if (needAnalyze) {
-            // execute real statement
-            preparedStmtCtx.shortCircuitQueryContext = Optional.empty();
-            statementContext.setShortCircuitQueryContext(null);
-            executor.execute();
-            if (executor.getContext().getStatementContext().isShortCircuitQuery()) {
-                // cache short-circuit plan
-                preparedStmtCtx.shortCircuitQueryContext = Optional.of(
-                        new ShortCircuitQueryContext(executor.planner(), (Queriable) executor.getParsedStmt()));
-                statementContext.setShortCircuitQueryContext(preparedStmtCtx.shortCircuitQueryContext.get());
-            }
+        if (executor.getContext().getStatementContext().isShortCircuitQuery()
+                && preparedStmtCtx.shortCircuitQueryContext.isPresent()
+                && preparedStmtCtx.shortCircuitQueryContext.get().tbl.getBaseSchemaVersion()
+                == preparedStmtCtx.shortCircuitQueryContext.get().schemaVersion && !executor.getContext()
+                .getStatementContext().hasNondeterministic()) {
+            PointQueryExecutor.directExecuteShortCircuitQuery(executor, preparedStmtCtx, statementContext);
             return;
         }
-        PointQueryExecutor.directExecuteShortCircuitQuery(executor, preparedStmtCtx, statementContext);
+        if (ctx.getSessionVariable().enableGroupCommitFullPrepare) {
+            if (preparedStmtCtx.groupCommitPlanner.isPresent()) {
+                OlapGroupCommitInsertExecutor.fastAnalyzeGroupCommit(ctx, prepareCommand);
+            } else {
+                OlapGroupCommitInsertExecutor.analyzeGroupCommit(ctx, prepareCommand);
+            }
+            if (ctx.isGroupCommit()) {
+                GroupCommitPlanner.executeGroupCommitInsert(ctx, preparedStmtCtx, statementContext);
+                return;
+            }
+        }
+        // execute real statement
+        preparedStmtCtx.shortCircuitQueryContext = Optional.empty();
+        statementContext.setShortCircuitQueryContext(null);
+        executor.execute();
+        if (executor.getContext().getStatementContext().isShortCircuitQuery()) {
+            // cache short-circuit plan
+            preparedStmtCtx.shortCircuitQueryContext = Optional.of(
+                    new ShortCircuitQueryContext(executor.planner(), (Queriable) executor.getParsedStmt()));
+            statementContext.setShortCircuitQueryContext(preparedStmtCtx.shortCircuitQueryContext.get());
+        }
     }
 
     /**

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

@@ -47,6 +47,7 @@
 import org.apache.doris.nereids.trees.plans.commands.Command;
 import org.apache.doris.nereids.trees.plans.commands.ForwardWithSync;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.UnboundLogicalSink;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalEmptyRelation;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalHiveTableSink;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalIcebergTableSink;
@@ -67,6 +68,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Throwables;
 import com.google.common.collect.Lists;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -423,6 +425,35 @@ public boolean isExternalTableSink() {
         return !(getLogicalQuery() instanceof UnboundTableSink);
     }
 
+    /**
+     * get the target table of the insert command
+     */
+    public TableIf getTable(ConnectContext ctx) throws Exception {
+        TableIf targetTableIf = InsertUtils.getTargetTable(originLogicalQuery, ctx);
+        if (!Env.getCurrentEnv().getAccessManager()
+                .checkTblPriv(ConnectContext.get(), targetTableIf.getDatabase().getCatalog().getName(),
+                        targetTableIf.getDatabase().getFullName(), targetTableIf.getName(),
+                        PrivPredicate.LOAD)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "LOAD",
+                    ConnectContext.get().getQualifiedUser(), ConnectContext.get().getRemoteIP(),
+                    targetTableIf.getDatabase().getFullName() + "." + targetTableIf.getName());
+        }
+        return targetTableIf;
+    }
+
+    /**
+     * get the target columns of the insert command
+     */
+    public List<String> getTargetColumns() {
+        if (originLogicalQuery instanceof UnboundTableSink) {
+            UnboundLogicalSink<? extends Plan> unboundTableSink = (UnboundTableSink<? extends Plan>) originLogicalQuery;
+            return CollectionUtils.isEmpty(unboundTableSink.getColNames()) ? null : unboundTableSink.getColNames();
+        } else {
+            throw new AnalysisException(
+                    "the root of plan should be [UnboundTableSink], but it is " + originLogicalQuery.getType());
+        }
+    }
+
     @Override
     public Plan getExplainPlan(ConnectContext ctx) {
         Optional<CascadesContext> analyzeContext = Optional.of(

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

@@ -32,6 +32,7 @@
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.trees.plans.algebra.InlineTable;
 import org.apache.doris.nereids.trees.plans.algebra.OneRowRelation;
+import org.apache.doris.nereids.trees.plans.commands.PrepareCommand;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
 import org.apache.doris.planner.GroupCommitPlanner;
@@ -68,15 +69,40 @@ public OlapGroupCommitInsertExecutor(ConnectContext ctx, Table table,
 
     /**
      * check if the sql can run in group commit mode
-     * @param logicalPlan plan of sql
      */
-    public static void analyzeGroupCommit(LogicalPlan logicalPlan) {
-        ConnectContext ctx = ConnectContext.get();
-        if (ctx.getSessionVariable().isEnableInsertGroupCommit() && logicalPlan instanceof InsertIntoTableCommand) {
-            LogicalPlan logicalQuery = ((InsertIntoTableCommand) logicalPlan).getLogicalQuery();
-            TableIf targetTableIf = InsertUtils.getTargetTable(logicalQuery, ctx);
-            OlapGroupCommitInsertExecutor.analyzeGroupCommit(ctx, targetTableIf, logicalQuery,
-                    Optional.empty());
+    public static void fastAnalyzeGroupCommit(ConnectContext ctx, LogicalPlan logicalPlan) {
+        try {
+            if (ctx.getSessionVariable().isEnableInsertGroupCommit() && !ctx.isTxnModel() && !ctx.getSessionVariable()
+                    .isEnableUniqueKeyPartialUpdate()) {
+                ctx.setGroupCommit(true);
+            }
+        } catch (Throwable e) {
+            LOG.warn("analyze group commit failed", e);
+        }
+    }
+
+    /**
+     * check if the sql can run in group commit mode
+     */
+    public static void analyzeGroupCommit(ConnectContext ctx, LogicalPlan logicalPlan) {
+        try {
+            if (ctx.isGroupCommit()) {
+                return;
+            }
+            if (!ctx.getSessionVariable().isEnableInsertGroupCommit()) {
+                return;
+            }
+            if (logicalPlan instanceof PrepareCommand) {
+                logicalPlan = ((PrepareCommand) logicalPlan).getLogicalPlan();
+            }
+            if (logicalPlan instanceof InsertIntoTableCommand) {
+                LogicalPlan logicalQuery = ((InsertIntoTableCommand) logicalPlan).getLogicalQuery();
+                TableIf targetTableIf = InsertUtils.getTargetTable(logicalQuery, ctx);
+                OlapGroupCommitInsertExecutor.analyzeGroupCommit(ctx, targetTableIf, logicalQuery,
+                        Optional.empty());
+            }
+        } catch (Throwable e) {
+            LOG.warn("analyze group commit failed", e);
         }
     }