[enhance](mtmv)Mv refresh on commit(#34548)

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -215,6 +215,7 @@ buildMode
 refreshTrigger
     : ON MANUAL
     | ON SCHEDULE refreshSchedule
+    | ON COMMIT
     ;
 
 refreshSchedule

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

@@ -141,6 +141,8 @@
 import org.apache.doris.deploy.impl.AmbariDeployManager;
 import org.apache.doris.deploy.impl.K8sDeployManager;
 import org.apache.doris.deploy.impl.LocalFileDeployManager;
+import org.apache.doris.event.EventProcessor;
+import org.apache.doris.event.ReplacePartitionEvent;
 import org.apache.doris.ha.BDBHA;
 import org.apache.doris.ha.FrontendNodeType;
 import org.apache.doris.ha.HAProtocol;
@@ -529,6 +531,7 @@ public class Env {
     private TopicPublisherThread topicPublisherThread;
 
     private MTMVService mtmvService;
+    private EventProcessor eventProcessor;
 
     private InsertOverwriteManager insertOverwriteManager;
 
@@ -772,6 +775,7 @@ public Env(boolean isCheckpointCatalog) {
         this.topicPublisherThread = new TopicPublisherThread(
                 "TopicPublisher", Config.publish_topic_info_interval_ms, systemInfo);
         this.mtmvService = new MTMVService();
+        this.eventProcessor = new EventProcessor(mtmvService);
         this.insertOverwriteManager = new InsertOverwriteManager();
         this.dnsCache = new DNSCache();
         this.sqlCacheManager = new NereidsSqlCacheManager();
@@ -839,6 +843,10 @@ public MTMVService getMtmvService() {
         return mtmvService;
     }
 
+    public EventProcessor getEventProcessor() {
+        return eventProcessor;
+    }
+
     public InsertOverwriteManager getInsertOverwriteManager() {
         return insertOverwriteManager;
     }
@@ -5547,6 +5555,18 @@ public void replaceTempPartition(Database db, OlapTable olapTable, ReplacePartit
         long version = olapTable.getNextVersion();
         long versionTime = System.currentTimeMillis();
         olapTable.updateVisibleVersionAndTime(version, versionTime);
+        // Here, we only wait for the EventProcessor to finish processing the event,
+        // but regardless of the success or failure of the result,
+        // it does not affect the logic of replace the partition
+        try {
+            Env.getCurrentEnv().getEventProcessor().processEvent(
+                    new ReplacePartitionEvent(db.getCatalog().getId(), db.getId(),
+                            olapTable.getId()));
+        } catch (Throwable t) {
+            // According to normal logic, no exceptions will be thrown,
+            // but in order to avoid bugs affecting the original logic, all exceptions are caught
+            LOG.warn("produceEvent failed: ", t);
+        }
         // write log
         ReplacePartitionOperationLog info =
                 new ReplacePartitionOperationLog(db.getId(), db.getFullName(), olapTable.getId(), olapTable.getName(),

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -139,6 +139,7 @@
 import org.apache.doris.datasource.hive.HMSCachedClient;
 import org.apache.doris.datasource.hive.HiveMetadataOps;
 import org.apache.doris.datasource.property.constants.HMSProperties;
+import org.apache.doris.event.DropPartitionEvent;
 import org.apache.doris.nereids.trees.plans.commands.info.DropMTMVInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.TableNameInfo;
 import org.apache.doris.persist.AlterDatabasePropertyInfo;
@@ -1809,11 +1810,22 @@ public void dropPartition(Database db, OlapTable olapTable, DropPartitionClause
         long version = olapTable.getNextVersion();
         long versionTime = System.currentTimeMillis();
         olapTable.updateVisibleVersionAndTime(version, versionTime);
+        // Here, we only wait for the EventProcessor to finish processing the event,
+        // but regardless of the success or failure of the result,
+        // it does not affect the logic of deleting the partition
+        try {
+            Env.getCurrentEnv().getEventProcessor().processEvent(
+                    new DropPartitionEvent(db.getCatalog().getId(), db.getId(),
+                            olapTable.getId()));
+        } catch (Throwable t) {
+            // According to normal logic, no exceptions will be thrown,
+            // but in order to avoid bugs affecting the original logic, all exceptions are caught
+            LOG.warn("produceEvent failed: ", t);
+        }
         // log
         DropPartitionInfo info = new DropPartitionInfo(db.getId(), olapTable.getId(), partitionName, isTempPartition,
                 clause.isForceDrop(), recycleTime, version, versionTime);
         Env.getCurrentEnv().getEditLog().logDropPartition(info);
-
         LOG.info("succeed in dropping partition[{}], table : [{}-{}], is temp : {}, is force : {}",
                 partitionName, olapTable.getId(), olapTable.getName(), isTempPartition, clause.isForceDrop());
     }

fe/fe-core/src/main/java/org/apache/doris/event/DataChangeEvent.java

@@ -0,0 +1,24 @@
+// 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.event;
+
+public class DataChangeEvent extends TableEvent {
+    public DataChangeEvent(long ctlId, long dbId, long tableId) {
+        super(EventType.DATA_CHANGE, ctlId, dbId, tableId);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/event/DropPartitionEvent.java

@@ -0,0 +1,24 @@
+// 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.event;
+
+public class DropPartitionEvent extends TableEvent {
+    public DropPartitionEvent(long ctlId, long dbId, long tableId) {
+        super(EventType.DROP_PARTITION, ctlId, dbId, tableId);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/event/Event.java

@@ -0,0 +1,60 @@
+// 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.event;
+
+import org.apache.doris.catalog.Env;
+
+import java.util.Objects;
+
+public abstract class Event {
+    protected final long eventId;
+
+    // eventTime of the event. Used instead of calling getter on event everytime
+    protected final long eventTime;
+
+    // eventType from the NotificationEvent
+    protected final EventType eventType;
+
+    protected Event(EventType eventType) {
+        Objects.requireNonNull(eventType, "require eventType");
+        this.eventId = Env.getCurrentEnv().getNextId();
+        this.eventTime = System.currentTimeMillis();
+        this.eventType = eventType;
+    }
+
+    public long getEventId() {
+        return eventId;
+    }
+
+    public long getEventTime() {
+        return eventTime;
+    }
+
+    public EventType getEventType() {
+        return eventType;
+    }
+
+    @Override
+    public String toString() {
+        return "Event{"
+                + "eventId=" + eventId
+                + ", eventTime=" + eventTime
+                + ", eventType=" + eventType
+                + '}';
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/event/EventException.java

@@ -0,0 +1,34 @@
+// 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.event;
+
+public class EventException extends Exception {
+
+    public EventException(String msg, Throwable cause) {
+        super(msg, cause);
+    }
+
+    public EventException(String msg) {
+        super(msg);
+    }
+
+    public EventException(Exception e) {
+        super(e);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/event/EventListener.java

@@ -0,0 +1,23 @@
+// 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.event;
+
+public interface EventListener {
+
+    void processEvent(Event event) throws EventException;
+}

fe/fe-core/src/main/java/org/apache/doris/event/EventProcessor.java

@@ -0,0 +1,57 @@
+// 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.event;
+
+import com.google.common.collect.Sets;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.Objects;
+import java.util.Set;
+
+public class EventProcessor {
+
+    private static final Logger LOG = LogManager.getLogger(EventProcessor.class);
+
+    private Set<EventListener> listeners = Sets.newHashSet();
+
+    public EventProcessor(EventListener... args) {
+        for (EventListener listener : args) {
+            this.listeners.add(listener);
+        }
+    }
+
+    public boolean processEvent(Event event) {
+        Objects.requireNonNull(event);
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("processEvent: {}", event);
+        }
+        boolean result = true;
+        for (EventListener listener : listeners) {
+            try {
+                listener.processEvent(event);
+            } catch (EventException e) {
+                // A listener processing failure does not affect other listeners
+                LOG.warn("[{}] process event failed, event: {}, errMsg: {}", listener.getClass().getName(), event,
+                        e.getMessage());
+                result = false;
+            }
+        }
+        return result;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/event/EventType.java

@@ -0,0 +1,24 @@
+// 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.event;
+
+public enum EventType {
+    DATA_CHANGE,
+    REPLACE_PARTITION,
+    DROP_PARTITION
+}

fe/fe-core/src/main/java/org/apache/doris/event/ReplacePartitionEvent.java

@@ -0,0 +1,24 @@
+// 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.event;
+
+public class ReplacePartitionEvent extends TableEvent {
+    public ReplacePartitionEvent(long ctlId, long dbId, long tableId) {
+        super(EventType.REPLACE_PARTITION, ctlId, dbId, tableId);
+    }
+}