[Enhancement] Count per-MV refresh metrics at job granularity

Migrate the per-MV refresh metrics (mv_refresh_jobs / mv_refresh_total_success_jobs /
mv_refresh_total_failed_jobs / mv_refresh_duration) from task-run granularity to job granularity so
they match information_schema.materialized_view_refresh_jobs:

- generateNextTaskRunIfNeeded() now returns whether it enqueued a successor batch; the refresh job is
  counted once, on its terminal task run (the run that enqueues no successor). MERGED runs are excluded.
- A successful batch that cannot enqueue its pending next batch (e.g. task-run queue full) leaves the
  refresh incomplete; the run fails through the normal failure path, so it is recorded as a failed job
  consistently in the metric, task history and materialized_view_refresh_jobs.
- mv_refresh_duration is the per-job wall-clock duration (ms), computed by reusing
  ShowMaterializedViewStatus.fromTaskRuns over the job's task runs (archived ones included) so it
  matches the system table.
- Docs (en/zh/ja) and unit tests updated.

mv_refresh_total_empty_jobs is left unchanged; its deprecation is deferred to a later PR.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Youngwb <yangwenbo_mailbox@163.com>

Author: Youngwb

docs/en/administration/management/monitoring/metrics-materialized_view.md

@@ -37,17 +37,17 @@ scrape_configs:
 ### mv_refresh_jobs
 
 - Type: Counter
-- Description: Total number of refresh jobs of the materialized view.
+- Description: Total number of refresh jobs triggered for the materialized view. A refresh job corresponds to one user-initiated or scheduled refresh; a single job may execute multiple task runs internally. Each job is counted once when it reaches a terminal state. MERGED task runs (sub-tasks merged into a later batch) are not counted.
 
 ### mv_refresh_total_success_jobs
 
 - Type: Counter
-- Description: Number of successful refresh jobs of the materialized view.
+- Description: Number of refresh jobs that completed successfully. Counted once per job.
 
 ### mv_refresh_total_failed_jobs
 
 - Type: Counter
-- Description: Number of failed refresh jobs of the materialized view.
+- Description: Number of refresh jobs that failed. Counted once per job.
 
 ### mv_refresh_total_empty_jobs
 
@@ -112,4 +112,4 @@ scrape_configs:
 ### mv_refresh_duration
 
 - Type: Histogram
-- Description: Duration of the successful materialized view refresh jobs.
+- Description: Wall-clock duration of refresh jobs, in milliseconds. For multi-batch jobs, measured from the first task run start to the final task run completion.

docs/ja/administration/management/monitoring/metrics-materialized_view.md

@@ -37,17 +37,17 @@ scrape_configs:
 ### mv_refresh_jobs
 
 - Type: Counter
-- Description: マテリアライズドビューのリフレッシュジョブの総数。
+- Description: マテリアライズドビューに対してトリガーされたリフレッシュジョブの総数。1つのリフレッシュジョブはユーザー起動またはスケジュールによる1回のリフレッシュに対応し、内部で複数の Task Run が実行される場合がある。各ジョブは終了状態に達した時点で1回カウントされる。MERGED 状態の Task Run（後のバッチにマージされたサブタスク）はカウントされない。
 
 ### mv_refresh_total_success_jobs
 
 - Type: Counter
-- Description: マテリアライズドビューの成功したリフレッシュジョブの数。
+- Description: 正常完了したリフレッシュジョブの数。ジョブ成功時に1回カウントされる。
 
 ### mv_refresh_total_failed_jobs
 
 - Type: Counter
-- Description: マテリアライズドビューの失敗したリフレッシュジョブの数。
+- Description: 失敗したリフレッシュジョブの数。ジョブ失敗時に1回カウントされる。
 
 ### mv_refresh_total_empty_jobs
 
@@ -112,4 +112,4 @@ scrape_configs:
 ### mv_refresh_duration
 
 - Type: Histogram
-- Description: 成功したマテリアライズドビューのリフレッシュジョブの期間。
+- Description: リフレッシュジョブの実時間（ミリ秒）。マルチバッチジョブの場合、最初の Task Run 開始から最後の Task Run 完了までを計測する。

docs/zh/administration/management/monitoring/metrics-materialized_view.md

@@ -37,17 +37,17 @@ scrape_configs:
 ### mv_refresh_jobs
 
 - 类型：Counter
-- 描述：物化视图刷新作业的总数。
+- 描述：物化视图触发的刷新作业总数。一个刷新作业对应一次用户发起或调度的刷新；单个作业内部可能执行多个 Task Run。每个作业在达到终止状态时计数一次。MERGED 状态的 Task Run（被合并到后续批次的子任务）不计入。
 
 ### mv_refresh_total_success_jobs
 
 - 类型：Counter
-- 描述：执行成功的物化视图刷新作业的数量。
+- 描述：执行成功的物化视图刷新作业的数量。每个作业成功时计数一次。
 
 ### mv_refresh_total_failed_jobs
 
 - 类型：Counter
-- 描述：执行失败的物化视图刷新作业的数量。
+- 描述：执行失败的物化视图刷新作业的数量。每个作业失败时计数一次。
 
 ### mv_refresh_total_empty_jobs
 
@@ -112,4 +112,4 @@ scrape_configs:
 ### mv_refresh_duration
 
 - 类型：Histogram
-- 描述：执行成功的物化视图刷新作业的持续时间。
+- 描述：刷新作业的挂钟时长，单位为毫秒。对于多批次作业，从第一个 Task Run 开始到最后一个 Task Run 完成为止计算。

fe/fe-core/src/main/java/com/starrocks/metric/MaterializedViewMetricsEntity.java

@@ -44,7 +44,7 @@ public final class MaterializedViewMetricsEntity implements IMaterializedViewMet
     private final List<Metric> metrics = Lists.newArrayList();
 
     // refresh
-    // increased once the materialized view's refresh job is triggered.
+    // increased once per refresh job, when it reaches a terminal state.
     public LongCounterMetric counterRefreshJobTotal;
     // increased only if the materialized view's refresh job is success refreshed.
     public LongCounterMetric counterRefreshJobSuccessTotal;
@@ -85,7 +85,7 @@ public final class MaterializedViewMetricsEntity implements IMaterializedViewMet
     public GaugeMetric<Integer> counterPartitionCount;
 
     // histogram(ms)
-    // record the materialized view's refresh job duration only if it's refreshed successfully.
+    // records the refresh job's wall-clock duration, once per job on its terminal run.
     public Histogram histRefreshJobDuration;
 
     public Optional<String> dbNameOpt = Optional.empty();
@@ -369,6 +369,7 @@ public void increaseRefreshJobStatus(Constants.TaskRunState status) {
         if (status == null || !status.isFinishState()) {
             return;
         }
+        // Counted once per refresh job, on its terminal task run.
         this.counterRefreshJobTotal.increase(1L);
         switch (status) {
             case MERGED:

fe/fe-core/src/main/java/com/starrocks/scheduler/MVTaskRunProcessor.java

@@ -17,6 +17,8 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Stopwatch;
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.Uninterruptibles;
 import com.starrocks.catalog.Database;
 import com.starrocks.catalog.MaterializedView;
@@ -34,6 +36,7 @@
 import com.starrocks.qe.ConnectContext;
 import com.starrocks.qe.QueryDetail;
 import com.starrocks.qe.QueryState;
+import com.starrocks.qe.ShowMaterializedViewStatus;
 import com.starrocks.qe.StmtExecutor;
 import com.starrocks.scheduler.mv.MVRefreshExecutor;
 import com.starrocks.scheduler.mv.MVRefreshProcessor;
@@ -55,6 +58,9 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.Logger;
 
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
@@ -194,22 +200,33 @@ public Constants.TaskRunState processTaskRun(TaskRunContext context) throws Exce
                 Preconditions.checkState(mv != null);
                 mvMetricsEntity = MaterializedViewMetricsRegistry.getInstance().getMetricsEntity(mv.getMvId());
                 this.taskRunState = retryProcessTaskRun(context);
+                boolean spawnedNext = false;
                 if (this.taskRunState == Constants.TaskRunState.SUCCESS) {
                     logger.info("Refresh materialized view {} finished successfully.", mv.getName());
                     // advance LAST_FRESHNESS_CONFIRMED_AT even when the refresh changed no version
                     mvRefreshProcessor.confirmFreshness();
-                    // if success, try to generate next task run
-                    mvRefreshProcessor.generateNextTaskRunIfNeeded();
+                    spawnedNext = mvRefreshProcessor.generateNextTaskRunIfNeeded();
+                    if (!spawnedNext && mvRefreshProcessor.hasNextBatchRun()) {
+                        // A pending next batch could not be enqueued (e.g. the task-run queue is full): the
+                        // refresh is incomplete, so fail it through the normal failure path below, keeping the
+                        // metric, task history and materialized_view_refresh_jobs consistent on FAILED.
+                        throw new DmlException("Materialized view %s refresh incomplete: " +
+                                "a pending batch could not be scheduled", mv.getName());
+                    }
                 } else {
                     logger.info("Refresh materialized view {} failed with state: {}.", mv.getName(), taskRunState);
                 }
-                // update metrics
-                mvMetricsEntity.increaseRefreshJobStatus(taskRunState);
+                // Count the job once on its terminal run; intermediate successful batches are not counted.
+                if (!spawnedNext) {
+                    mvMetricsEntity.increaseRefreshJobStatus(this.taskRunState);
+                    recordRefreshJobDuration(mvMetricsEntity);
+                }
                 connectContext.getState().setOk();
             }
         } catch (Exception e) {
             if (mvMetricsEntity != null) {
                 mvMetricsEntity.increaseRefreshJobStatus(Constants.TaskRunState.FAILED);
+                recordRefreshJobDuration(mvMetricsEntity);
             }
             connectContext.getState().setError(e.getMessage());
             throw e;
@@ -320,16 +337,13 @@ public Constants.TaskRunState doProcessTaskRun(TaskRunContext taskRunContext,
         final MVRefreshProcessor.ProcessExecPlan processExecPlan = mvRefreshProcessor.getProcessExecPlan(taskRunContext);
         if (processExecPlan.state() == Constants.TaskRunState.SKIPPED) {
             logger.info("MV {} refresh task skipped, no partitions to refresh", mv.getName());
-            long elapsed = watch.elapsed(TimeUnit.MILLISECONDS);
-            mvMetricsEntity.updateRefreshDuration(elapsed);
             return Constants.TaskRunState.SKIPPED;
         }
 
         // refresh materialized view
         Constants.TaskRunState result = mvRefreshProcessor.execProcessExecPlan(taskRunContext, processExecPlan, executor);
         long elapsed = watch.elapsed(TimeUnit.MILLISECONDS);
         logger.info("refresh mv success, cost time(ms): {}", DebugUtil.DECIMAL_FORMAT_SCALE_3.format(elapsed));
-        mvMetricsEntity.updateRefreshDuration(elapsed);
         return result;
     }
 
@@ -451,4 +465,43 @@ public String getExtraExplainInfo(StatementBase statement) {
     public RuntimeProfile getRuntimeProfile() {
         return runtimeProfile;
     }
+
+    // Records the job's wall-clock duration once, on the terminal run, reusing the exact roll-up the
+    // materialized_view_refresh_jobs system table uses so the metric and the table never diverge.
+    private void recordRefreshJobDuration(IMaterializedViewMetricsEntity metricsEntity) {
+        try {
+            if (mvTaskRunContext == null || mvTaskRunContext.getStatus() == null) {
+                return;
+            }
+            String jobId = mvTaskRunContext.getStatus().getStartTaskRunId();
+            if (Strings.isNullOrEmpty(jobId)) {
+                return;
+            }
+            List<TaskRunStatus> batch = Lists.newArrayList();
+            // Include archived runs: a long multi-batch refresh may have had early batches archived out of
+            // in-memory history before this terminal run, and the system table's roll-up also reads archived history.
+            String dbName = db == null ? null : db.getFullName();
+            if (dbName != null) {
+                String mvTaskName = TaskBuilder.getMvTaskName(mv.getId());
+                for (TaskRunStatus s : GlobalStateMgr.getCurrentState().getTaskManager().getTaskRunHistory()
+                        .lookupLastJobOfTasks(dbName, Collections.singleton(mvTaskName))) {
+                    if (jobId.equals(s.getStartTaskRunId()) && s.getState() != Constants.TaskRunState.MERGED) {
+                        batch.add(s);
+                    }
+                }
+            }
+            batch.add(mvTaskRunContext.getStatus());
+            // fromTaskRuns picks first/last by createTime — sort the same way.
+            batch.sort(Comparator.comparingLong(TaskRunStatus::getCreateTime));
+            ShowMaterializedViewStatus.RefreshJobStatus rjs = ShowMaterializedViewStatus.fromTaskRuns(batch);
+            long startBasis = rjs.getMvRefreshProcessTime() > 0
+                    ? rjs.getMvRefreshProcessTime() : rjs.getMvRefreshStartTime();
+            // The terminal run's finishTime may not be persisted yet; fall back to now().
+            long endMs = rjs.getMvRefreshEndTime() > 0 ? rjs.getMvRefreshEndTime() : System.currentTimeMillis();
+            long wallMs = Math.max(0L, endMs - startBasis);
+            metricsEntity.updateRefreshDuration(wallMs);
+        } catch (Exception e) {
+            logger.warn("skip refresh job duration metric for mv {}", mv.getName(), e);
+        }
+    }
 }

fe/fe-core/src/main/java/com/starrocks/scheduler/mv/MVRefreshProcessor.java

@@ -202,8 +202,9 @@ public abstract BaseTableSnapshotInfo buildBaseTableSnapshotInfo(BaseTableInfo b
 
     /**
      * Generate the next task run to be processed and set it to the nextTaskRun field.
+     * Returns true iff this call enqueued a successor task run, false otherwise.
      */
-    public abstract void generateNextTaskRunIfNeeded();
+    public abstract boolean generateNextTaskRunIfNeeded();
 
     /**
      * Whether this refresh will spawn another batch task run after the current one.

fe/fe-core/src/main/java/com/starrocks/scheduler/mv/hybrid/MVHybridRefreshProcessor.java

@@ -146,8 +146,8 @@ public BaseTableSnapshotInfo buildBaseTableSnapshotInfo(BaseTableInfo baseTableI
     }
 
     @Override
-    public void generateNextTaskRunIfNeeded() {
-        getCurrentProcessor().generateNextTaskRunIfNeeded();
+    public boolean generateNextTaskRunIfNeeded() {
+        return getCurrentProcessor().generateNextTaskRunIfNeeded();
     }
 
     @Override

fe/fe-core/src/main/java/com/starrocks/scheduler/mv/ivm/MVIVMRefreshProcessor.java

@@ -43,6 +43,7 @@
 import com.starrocks.scheduler.Constants;
 import com.starrocks.scheduler.ExecuteOption;
 import com.starrocks.scheduler.MvTaskRunContext;
+import com.starrocks.scheduler.SubmitResult;
 import com.starrocks.scheduler.TaskBuilder;
 import com.starrocks.scheduler.TaskManager;
 import com.starrocks.scheduler.TaskRun;
@@ -469,9 +470,9 @@ static final class AdaptiveDelta {
     }
 
     @Override
-    public void generateNextTaskRunIfNeeded() {
+    public boolean generateNextTaskRunIfNeeded() {
         if (!hasNextTaskRun || mvContext.getTaskRun().isKilled()) {
-            return;
+            return false;
         }
         TaskManager taskManager = GlobalStateMgr.getCurrentState().getTaskManager();
         Map<String, String> properties = mvContext.getProperties();
@@ -522,9 +523,11 @@ public void generateNextTaskRunIfNeeded() {
                     .setExecuteOption(option)
                     .build();
             nextTaskRun = taskRun;
-        } else {
-            taskManager.executeTask(taskName, option);
+            return true;
         }
+        // Report the job as continued only if the successor run was accepted; a rejected submit (e.g. queue
+        // full) means no successor runs, so the current run stays the job's terminal run.
+        return taskManager.executeTask(taskName, option).getStatus() == SubmitResult.SubmitStatus.SUBMITTED;
     }
 
     @Override

fe/fe-core/src/main/java/com/starrocks/scheduler/mv/pct/MVPCTRefreshProcessor.java

@@ -37,6 +37,7 @@
 import com.starrocks.scheduler.Constants;
 import com.starrocks.scheduler.ExecuteOption;
 import com.starrocks.scheduler.MvTaskRunContext;
+import com.starrocks.scheduler.SubmitResult;
 import com.starrocks.scheduler.TaskBuilder;
 import com.starrocks.scheduler.TaskManager;
 import com.starrocks.scheduler.TaskRun;
@@ -295,9 +296,9 @@ private InsertStmt prepareRefreshPlan(PCellSortedSet mvToRefreshedPartitions,
     }
 
     @Override
-    public void generateNextTaskRunIfNeeded() {
+    public boolean generateNextTaskRunIfNeeded() {
         if (!mvContext.hasNextBatchPartition() || mvContext.getTaskRun().isKilled()) {
-            return;
+            return false;
         }
 
         TaskManager taskManager = GlobalStateMgr.getCurrentState().getTaskManager();
@@ -370,9 +371,11 @@ public void generateNextTaskRunIfNeeded() {
                     .setExecuteOption(option)
                     .build();
             nextTaskRun = taskRun;
-        } else {
-            taskManager.executeTask(taskName, option);
+            return true;
         }
+        // Report the job as continued only if the successor run was accepted; a rejected submit (e.g. queue
+        // full) means no successor runs, so the current run stays the job's terminal run.
+        return taskManager.executeTask(taskName, option).getStatus() == SubmitResult.SubmitStatus.SUBMITTED;
     }
 
     @Override