[fix](mtmv) Generate mtmv cache should use ADMIN user, and rewritten plan should not check privilege (#40374) (#41450)

## Proposed changes

pr: https://github.com/apache/doris/pull/40374
commitId: f3d92e48

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

@@ -307,7 +307,13 @@ public class MTMV extends OlapTable {
         }
         // Concurrent situations may result in duplicate cache generation,
         // but we tolerate this in order to prevent nested use of readLock and write MvLock for the table
-        MTMVCache mtmvCache = MTMVCache.from(this, connectionContext, true);
+        MTMVCache mtmvCache;
+        try {
+            // Should new context with ADMIN user
+            mtmvCache = MTMVCache.from(this, MTMVPlanUtil.createMTMVContext(this), true);
+        } finally {
+            connectionContext.setThreadLocalInfo();
+        }
         writeMvLock();
         try {
             this.cache = mtmvCache;

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/AccessControllerManager.java

@@ -172,6 +172,15 @@ public class AccessControllerManager {
     }
 
     // ==== Column ====
+    // If param has ctx, we can skip auth by isSkipAuth field in ctx
+    public void checkColumnsPriv(ConnectContext ctx, String ctl, String qualifiedDb, String tbl, Set<String> cols,
+            PrivPredicate wanted) throws UserException {
+        if (ctx.isSkipAuth()) {
+            return;
+        }
+        checkColumnsPriv(ctx.getCurrentUserIdentity(), ctl, qualifiedDb, tbl, cols, wanted);
+    }
+
     public void checkColumnsPriv(UserIdentity currentUser, String
             ctl, String qualifiedDb, String tbl, Set<String> cols,
             PrivPredicate wanted) throws UserException {

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/UserAuthentication.java

@@ -17,7 +17,6 @@
 
 package org.apache.doris.nereids.rules.analysis;
 
-import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.ErrorCode;
@@ -59,14 +58,13 @@ public class UserAuthentication {
         }
         String ctlName = catalog.getName();
         AccessControllerManager accessManager = connectContext.getEnv().getAccessManager();
-        UserIdentity userIdentity = connectContext.getCurrentUserIdentity();
         if (CollectionUtils.isEmpty(columns)) {
-            if (!accessManager.checkTblPriv(userIdentity, ctlName, dbName, tableName, PrivPredicate.SELECT)) {
+            if (!accessManager.checkTblPriv(connectContext, ctlName, dbName, tableName, PrivPredicate.SELECT)) {
                 ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLE_ACCESS_DENIED_ERROR,
                         PrivPredicate.SELECT.getPrivs().toString(), tableName);
             }
         } else {
-            accessManager.checkColumnsPriv(userIdentity, ctlName, dbName, tableName, columns, PrivPredicate.SELECT);
+            accessManager.checkColumnsPriv(connectContext, ctlName, dbName, tableName, columns, PrivPredicate.SELECT);
         }
     }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtils.java

@@ -227,7 +227,7 @@ public class MaterializedViewUtils {
 
     /**
      * Optimize by rules, this support optimize by custom rules by define different rewriter according to different
-     * rules
+     * rules, this method is only for materialized view rewrite
      */
     public static Plan rewriteByRules(
             CascadesContext cascadesContext,
@@ -247,7 +247,12 @@ public class MaterializedViewUtils {
         CascadesContext rewrittenPlanContext = CascadesContext.initContext(
                 cascadesContext.getStatementContext(), rewrittenPlan,
                 cascadesContext.getCurrentJobContext().getRequiredProperties());
-        rewrittenPlan = planRewriter.apply(rewrittenPlanContext);
+        try {
+            rewrittenPlanContext.getConnectContext().setSkipAuth(true);
+            rewrittenPlan = planRewriter.apply(rewrittenPlanContext);
+        } finally {
+            rewrittenPlanContext.getConnectContext().setSkipAuth(false);
+        }
         Map<ExprId, Slot> exprIdToNewRewrittenSlot = Maps.newLinkedHashMap();
         for (Slot slot : rewrittenPlan.getOutput()) {
             exprIdToNewRewrittenSlot.put(slot.getExprId(), slot);

fe/fe-core/src/test/java/org/apache/doris/nereids/mv/MtmvCacheNewConnectContextTest.java

@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.mv;
+
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.mtmv.MTMVRelationManager;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.sqltest.SqlTestBase;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
+
+import mockit.Mock;
+import mockit.MockUp;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.BitSet;
+
+/**
+ * The connectContext would new instance when generate MTMVCache, after generate, the connectContext should
+ * reset the connectContext by earlier connectContext to avoid slot id error
+ * The test is for this.
+ * */
+public class MtmvCacheNewConnectContextTest extends SqlTestBase {
+
+    @Test
+    void testConnectContextIsCorrect() throws Exception {
+        ConnectContext tmp = ConnectContext.get();
+        connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION");
+        BitSet disableNereidsRules = connectContext.getSessionVariable().getDisableNereidsRules();
+        new MockUp<SessionVariable>() {
+            @Mock
+            public BitSet getDisableNereidsRules() {
+                return disableNereidsRules;
+            }
+        };
+        new MockUp<MTMVRelationManager>() {
+            @Mock
+            public boolean isMVPartitionValid(MTMV mtmv, ConnectContext ctx, boolean forceConsistent) {
+                return true;
+            }
+        };
+        connectContext.getSessionVariable().enableMaterializedViewRewrite = true;
+        connectContext.getSessionVariable().enableMaterializedViewNestRewrite = true;
+
+        createMvByNereids("create materialized view mv1 BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+                + "        DISTRIBUTED BY RANDOM BUCKETS 1\n"
+                + "        PROPERTIES ('replication_num' = '1') \n"
+                + "        as select T1.id from T1 inner join T2 "
+                + "        on T1.id = T2.id;");
+        CascadesContext c1 = createCascadesContext(
+                "select T1.id from T1 inner join T2 "
+                        + "on T1.id = T2.id "
+                        + "inner join T3 on T1.id = T3.id",
+                connectContext
+        );
+        PlanChecker.from(c1)
+                .analyze()
+                .rewrite()
+                .optimize()
+                .printlnBestPlanTree();
+
+        ConnectContext now = ConnectContext.get();
+        // The connectContext should not change
+        Assertions.assertSame(tmp, now);
+        dropMvByNereids("drop materialized view mv1");
+    }
+}