[BugFix] Fix possible dead lock in LoadJobStatsListener (#48146)

Signed-off-by: shuming.li <[email protected]>

Author: LiShuMing

fe/fe-core/src/main/java/com/starrocks/listener/LoadJobMVListener.java

@@ -130,18 +130,14 @@ private void doTriggerToRefreshRelatedMVs(Database db, Table table) throws DdlEx
                 mvIdIterator.remove();
                 continue;
             }
-            mvDb.readLock();
-            try {
-                if (materializedView.shouldTriggeredRefreshBy(db.getFullName(), table.getName())) {
-                    LOG.info("Trigger auto materialized view refresh because of base table {} has changed, " +
-                                    "db:{}, mv:{}", table.getName(), mvDb.getFullName(),
-                            materializedView.getName());
-                    GlobalStateMgr.getCurrentState().getLocalMetastore().refreshMaterializedView(
-                            mvDb.getFullName(), mvDb.getTable(mvId.getId()).getName(), false, null,
-                            Constants.TaskRunPriority.NORMAL.value(), true, false);
-                }
-            } finally {
-                mvDb.readUnlock();
+            // It's fine to no lock here, since it's not a critical operation and can be retried.
+            if (materializedView.shouldTriggeredRefreshBy(db.getFullName(), table.getName())) {
+                LOG.info("Trigger auto materialized view refresh because of base table {} has changed, " +
+                                "db:{}, mv:{}", table.getName(), mvDb.getFullName(),
+                        materializedView.getName());
+                GlobalStateMgr.getCurrentState().getLocalMetastore().refreshMaterializedView(
+                        mvDb.getFullName(), mvDb.getTable(mvId.getId()).getName(), false, null,
+                        Constants.TaskRunPriority.NORMAL.value(), true, false);
             }
         }
     }

fe/fe-core/src/main/java/com/starrocks/load/InsertOverwriteJobRunner.java

@@ -387,8 +387,11 @@ private void gc(boolean isReplay) {
 
     private void doCommit(boolean isReplay) {
         Database db = getAndWriteLockDatabase(dbId);
-        OlapTable targetTable = checkAndGetTable(db, tableId);
+        OlapTable tmpTargetTable = null;
         try {
+            // try exception to release write lock finally
+            final OlapTable targetTable = checkAndGetTable(db, tableId);
+            tmpTargetTable = targetTable;
             List<String> sourcePartitionNames = job.getSourcePartitionIds().stream()
                     .map(partitionId -> targetTable.getPartition(partitionId).getName())
                     .collect(Collectors.toList());
@@ -438,8 +441,9 @@ private void doCommit(boolean isReplay) {
             db.writeUnlock();
         }
 
-        // trigger listeners after insert overwrite committed.
-        GlobalStateMgr.getCurrentState().getOperationListenerBus().onInsertOverwriteJobCommitFinish(db, targetTable);
+        // trigger listeners after insert overwrite committed, trigger listeners after
+        // write unlock to avoid holding lock too long
+        GlobalStateMgr.getCurrentState().getOperationListenerBus().onInsertOverwriteJobCommitFinish(db, tmpTargetTable);
     }
 
     private void prepareInsert() {