This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 9dc9dc75661 [feature](ivm) Report how a refresh task refreshed an IVM
MV, and count the refresh tasks on the Master (#69856)
9dc9dc75661 is described below
commit 9dc9dc75661dd0a1e5031b629bdc025a19c76839
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Nov 2 09:41:47 2026 +0000
[feature](ivm) Report how a refresh task refreshed an IVM MV, and count the
refresh tasks on the Master (#69856)
### What problem does this PR solve?
Issue Number: #69907
Related PR: None
Problem Summary:
**In short.** A refresh of an IVM materialized view can keep the MV up
incrementally or quietly recompute it, and nothing tells the two apart:
`tasks("type"="mv")` does not say which kinds of work a task committed,
and `/metrics` has no IVM series at all. This PR makes each refresh task
of an IVM MV report how it refreshed the MV (`IvmRefreshMethod` in
`tasks()`) and why it failed or fell back, keeps a refresh summary on
the MV for the IVM status table that follows, and counts the tasks on
the Master in five new counters.
**Background.** An IVM MV is refreshed by a task that runs one or more
attempts: the incremental attempt (the delta of its base tables'
streams, after rebuilding the MV partitions an invalidation marked), a
partition rebuild, and a whole-MV rebuild. A request that may fall back
(AUTO, INCREMENTAL FALLBACK, PARTITIONS FALLBACK) moves on to the next
attempt when one gives up, for example to COMPLETE after the delta meets
a MIN/MAX boundary. `tasks()` already reports `RefreshMode` (which
partitions the task covered), `IvmFallbackReason` (why the incremental
attempt gave up, or why a strict request failed) and
`IvmRebuiltPartitions`.
**The problem, and what it cost.**
- An operator cannot tell a refresh that kept the MV up incrementally
from one that recomputed it. `RefreshMode` says `COMPLETE` for any
whole-MV scope, and `IvmFallbackReason` is set even when the attempt
that replaced the delta wrote nothing, or when a strict request failed.
- No alert can watch IVM refreshes. The asynchronous MV counters
(`async_materialized_view_task_success_num`, `_failed_num`, `_skip_num`)
cover every MV and say nothing about the method, the fallbacks, or why
tasks fail. An MV that falls back to COMPLETE on every run, the
expensive failure mode of IVM, goes unnoticed until the load of the
cluster shows it.
- A task that the full MV task queue turns away never runs, and nothing
counts it.
- The task history keeps only `max_persistence_task_count` tasks, so how
many times in a row an MV has failed, or when it last fell back, cannot
be read back from it.
**How this PR fixes it.**
- `MTMVTask` notes each batch that commits data (the delta, a partition
rebuild, a whole-MV rebuild) and decides `IvmRefreshMethod` once, as the
task publishes its result: `NONE`, `INCREMENTAL`, `PARTITION_REBUILD`,
`COMPLETE`, or `MIXED` when more than one kind committed, typically the
rebuild of invalidated partitions before the delta. `tasks("type"="mv")`
gets the `IvmRefreshMethod` column, NULL while a task runs and for a
task of an MV that is not an IVM MV.
- A task falls back when the incremental attempt gives up and an attempt
replacing it commits data. Not fallbacks: building the first baseline of
an MV no refresh has succeeded on, the rebuild of invalidated partitions
before the delta, an explicit COMPLETE or PARTITIONS request, and a
fallback that commits nothing.
- A failed task reports why: the `IvmFailureReason` a request that may
not fall back failed with, or else the stage it failed in,
`QUERY_UNUSABLE`, `PARTITION_SYNC_FAILED`, `EXECUTION_FAILED` or
`OTHER`, told apart by where the failure was thrown rather than by its
message.
- The method, the failure reason and whether the task fell back are
journaled with the task, as new Gson fields that an older FE ignores.
The MV keeps an `IvmRefreshSummary`, updated from each task result in
`MTMV.addTaskResult`, which the Master and the replay of the same
journal record run alike, from what the record carries: consecutive
failures, the latest success and failure (with the reason and the error
message, cut to 512 characters), the latest method and run time, and the
fallback count with the latest reason. A canceled task leaves it as it
is. Nothing reads it in this PR; `information_schema.ivm_mv_status` and
the per-MV series of a follow-up PR do.
- Five counters, counted on the Master as each task of an IVM MV ends.
Every label combination is registered up front, so each of the 61 series
exists from 0:
| Series | Type | Labels | Value |
| --- | --- | --- | --- |
| `doris_fe_ivm_refresh_tasks_total` | counter |
`result`=success/no_change/failed/canceled,
`method`=none/incremental/partition_rebuild/complete/mixed | refresh
tasks that ended; `no_change` is a SUCCESS that committed nothing |
| `doris_fe_ivm_refresh_duration_seconds_total` | counter | `method` |
the run time, StartTime to FinishTime, of the tasks that succeeded or
failed; a canceled task has no finish time |
| `doris_fe_ivm_fallbacks_total` | counter | `reason`=the
`IvmFailureReason` values, `incomplete_refresh_snapshot` | tasks that
fell back |
| `doris_fe_ivm_refresh_failures_total` | counter | `reason`=the
`IvmFailureReason` values, the four stages | tasks that failed |
| `doris_fe_ivm_schedule_rejected_total` | counter | | tasks the full MV
task queue turned away |
Like the other FE event counters they live in the registry of every FE
and only the Master counts in them, so a query sums them over the FEs; a
restart or a new Master starts them from 0.
- `AbstractTask.onRejected`, a new hook that `TaskProcessor.addTask`
calls when the executor queue rejects a task. Both the instant dispatch
(`JobScheduler`) and the timer dispatch (`DispatchTaskHandler`) add
tasks through it, and `MTMVTask` counts the rejection for an IVM MV. A
timer-dispatched task that is turned away is canceled as before, so it
also counts as `canceled`.
**Results.** Before this PR, `tasks()` has no method column and
`/metrics` has none of these series. After it, on a local cluster, one
IVM MV refreshed through each case (regression
`test_ivm_refresh_method_metrics`):
| Refresh | `tasks()`: Status, IvmRefreshMethod, IvmFallbackReason |
Counters that moved, besides the run time |
| --- | --- | --- |
| first AUTO refresh of a new MV | SUCCESS, COMPLETE, NULL |
`tasks{success,complete}` |
| INCREMENTAL after an insert | SUCCESS, INCREMENTAL, NULL |
`tasks{success,incremental}` |
| INCREMENTAL with nothing new | SUCCESS, NONE, NULL |
`tasks{no_change,none}` |
| INCREMENTAL after deleting the row that holds a group's minimum |
FAILED, NONE, MIN_MAX_BOUNDARY_HIT | `tasks{failed,none}`,
`failures{min_max_boundary_hit}` |
| INCREMENTAL FALLBACK in the same state | SUCCESS, COMPLETE,
MIN_MAX_BOUNDARY_HIT | `tasks{success,complete}`,
`fallbacks{min_max_boundary_hit}` |
| COMPLETE | SUCCESS, COMPLETE, NULL | `tasks{success,complete}` |
| INCREMENTAL after truncating a base partition and inserting into
another | SUCCESS, MIXED, NULL | `tasks{success,mixed}` |
The first refresh of a new MV finds every partition still to be filled
and goes straight to COMPLETE; it counts no fallback. The MV holds what
its base table does after each case.
**Classes, and how they call each other.**
- `IvmRefreshMethod`, `IvmRefreshFailureStage`
(`org.apache.doris.mtmv.ivm`): the methods, and the stages a failure is
reported by.
- `MTMVTask`: notes committed batches (`recordCommitted`), decides
whether giving up the incremental attempt is a fallback
(`givingUpTheIncrementalAttemptIsAFallback`), moves the failure stage
along, settles the method and the failure reason in `after()`, and
counts the task (`recordIvmRefreshMetrics`) and a rejection
(`onRejected`).
- `IvmRefreshSummary` (`org.apache.doris.mtmv.ivm`): persisted on `MTMV`
and updated by `update(MTMVTask)` from `MTMV.addTaskResult`;
`MTMV.hasSucceededIvmRefresh` tells the first baseline apart.
- `MetricRepo`: registers the five counters. `DoubleCounterMetric` (new)
keeps a total in seconds without rounding each addition.
- `AbstractTask.onRejected` (new, empty by default), called by
`TaskProcessor.addTask`.
```
MTMVTask.run() worker thread
├─ executeIvmAttempt
│ ├─ rebuild invalidated partitions ──►
recordCommitted(PARTITION_REBUILD)
│ └─ delta (IvmIncrRefreshManager) ───► recordCommitted(INCREMENTAL)
│ gave up ─► fallingBack = givingUpTheIncrementalAttemptIsAFallback()
├─ PARTITIONS / COMPLETE attempts ───────► recordCommitted(...)
ivmFallback once fallingBack
└─ failureStage: QUERY_UNUSABLE ─► PARTITION_SYNC_FAILED ─►
EXECUTION_FAILED, or an IvmFailureReason
onSuccess / onFail / cancel ─► after()
├─ ivmRefreshMethod, ivmFailureReason settled
├─ recordIvmRefreshMetrics() ─► MetricRepo.COUNTER_IVM_*
Master only
└─ Env.addMTMVTaskResult ─► MTMV.addTaskResult ─►
IvmRefreshSummary.update(task)
(journaled as ADD_TASK; the replay calls
addTaskResult the same way)
TaskProcessor.addTask ── queue full ─► task.onRejected() ─► MTMVTask:
COUNTER_IVM_SCHEDULE_REJECTED
```
Nothing else changes: no thrift or proto change, no new config, and the
existing columns and series stay as they are.
---
.../main/java/org/apache/doris/catalog/MTMV.java | 23 ++
.../apache/doris/job/executor/TaskProcessor.java | 1 +
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 162 ++++++++++-
.../org/apache/doris/job/task/AbstractTask.java | 7 +
.../apache/doris/metric/DoubleCounterMetric.java | 43 +++
.../java/org/apache/doris/metric/MetricRepo.java | 67 +++++
.../doris/mtmv/ivm/IvmRefreshFailureStage.java | 37 +++
.../apache/doris/mtmv/ivm/IvmRefreshMethod.java | 55 ++++
.../apache/doris/mtmv/ivm/IvmRefreshSummary.java | 142 ++++++++++
.../doris/job/executor/TaskProcessorTest.java | 46 +++
.../apache/doris/metric/IvmRefreshMetricsTest.java | 161 +++++++++++
.../java/org/apache/doris/mtmv/MTMVTaskTest.java | 307 +++++++++++++++++++++
.../test/java/org/apache/doris/mtmv/MTMVTest.java | 93 +++++++
.../doris/mtmv/ivm/IvmBaselineRebuildTest.java | 3 +
.../doris/mtmv/ivm/IvmRefreshSummaryTest.java | 163 +++++++++++
.../ivm/test_ivm_refresh_method_metrics.out | 40 +++
.../ivm/test_ivm_refresh_method_metrics.groovy | 233 ++++++++++++++++
17 files changed, 1581 insertions(+), 2 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
index dad9765cf9b..c76ccd636e1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
@@ -56,6 +56,7 @@ import org.apache.doris.mtmv.MTMVSnapshotIf;
import org.apache.doris.mtmv.MTMVStatus;
import org.apache.doris.mtmv.MTMVUtil;
import org.apache.doris.mtmv.ivm.IvmInfo;
+import org.apache.doris.mtmv.ivm.IvmRefreshSummary;
import org.apache.doris.mtmv.ivm.IvmUtil;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.rules.analysis.SessionVarGuardRewriter;
@@ -126,6 +127,10 @@ public class MTMV extends OlapTable {
*/
@SerializedName("pst")
private Map<String, MTMVPartitionState> partitionStates =
Maps.newLinkedHashMap();
+ // What the refresh tasks of an IVM MV have come to; updated with each
task result, see addTaskResult. An MV
+ // that is not an IVM MV leaves it empty.
+ @SerializedName("irs")
+ private IvmRefreshSummary ivmRefreshSummary = new IvmRefreshSummary();
// Increased every time rewrite cache is invalidated to prevent publishing
stale in-flight cache builds.
private transient long rewriteCacheGeneration;
private long schemaChangeVersion;
@@ -320,6 +325,19 @@ public class MTMV extends OlapTable {
return getIvmInfo().isEnableIvm();
}
+ /**
+ * Whether a refresh task of this IVM MV has ever succeeded. Until one
has, the MV has no baseline for the
+ * incremental refresh to keep up, and the refresh that builds it is not
the incremental refresh falling back.
+ */
+ public boolean hasSucceededIvmRefresh() {
+ readMvLock();
+ try {
+ return ivmRefreshSummary.hasSucceeded();
+ } finally {
+ readMvUnlock();
+ }
+ }
+
/**
* Whether this MV is an IVM MV built as UNIQUE KEY + merge-on-write,
which is one of the two table
* models: it is maintained through the hidden row-id, and the layout its
refresh reads is pinned
@@ -447,6 +465,11 @@ public class MTMV extends OlapTable {
this.status.setRefreshState(MTMVRefreshState.FAIL);
}
this.jobInfo.addHistoryTask(task);
+ if (ivmInfo.isEnableIvm()) {
+ // From the task as the journal carries it, here and in the
replay alike, so that the summary does
+ // not depend on how many tasks the history keeps and a
follower that takes over has the same one.
+ ivmRefreshSummary.update(task);
+ }
compatiblePctSnapshot(partitionSnapshots);
// What this task wrote is described by the epochs just recorded,
so a partition the result
// left dirty is left out: its snapshot would otherwise come back
after an invalidation
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/executor/TaskProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/job/executor/TaskProcessor.java
index 5a46ef13864..b8d00abf4f0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/executor/TaskProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/executor/TaskProcessor.java
@@ -53,6 +53,7 @@ public class TaskProcessor {
return true;
} catch (RejectedExecutionException e) {
log.warn("Failed to add task to executor, task id: {}",
task.getTaskId(), e);
+ task.onRejected();
return false;
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
index 055d9fef675..9886cf718ea 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
@@ -72,6 +72,8 @@ import org.apache.doris.mtmv.ivm.IvmIncrRefreshContext;
import org.apache.doris.mtmv.ivm.IvmIncrRefreshManager;
import org.apache.doris.mtmv.ivm.IvmIncrRefreshResult;
import org.apache.doris.mtmv.ivm.IvmPlanSignature;
+import org.apache.doris.mtmv.ivm.IvmRefreshFailureStage;
+import org.apache.doris.mtmv.ivm.IvmRefreshMethod;
import org.apache.doris.mtmv.ivm.IvmRewriteContext;
import org.apache.doris.mtmv.ivm.IvmUtil;
import org.apache.doris.nereids.NereidsPlanner;
@@ -104,6 +106,7 @@ import org.apache.logging.log4j.Logger;
import java.math.BigDecimal;
import java.math.RoundingMode;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
@@ -114,9 +117,11 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
+import java.util.stream.Stream;
public class MTMVTask extends AbstractTask {
private static final Logger LOG = LogManager.getLogger(MTMVTask.class);
@@ -125,6 +130,17 @@ public class MTMVTask extends AbstractTask {
"MTMVTask.syncPartitionsIfNeeded.skip";
public static final String DEBUG_POINT_SKIP_PARTITION_SYNC_FILTER =
"MTMVTask.syncPartitionsIfNeeded.skip.filter";
+ // The fallback reason of an incremental attempt that found no refresh
snapshot to catch up from.
+ public static final String INCOMPLETE_REFRESH_SNAPSHOT =
"INCOMPLETE_REFRESH_SNAPSHOT";
+ // Every value ivmFallbackReason can take, and every value
ivmFailureReason can take, each of which the IVM
+ // refresh metrics register a series for.
+ public static final ImmutableList<String> IVM_FALLBACK_REASONS =
Stream.concat(
+ Arrays.stream(IvmFailureReason.values()).map(Enum::name),
Stream.of(INCOMPLETE_REFRESH_SNAPSHOT))
+ .collect(ImmutableList.toImmutableList());
+ public static final ImmutableList<String> IVM_FAILURE_REASONS =
Stream.concat(
+ Arrays.stream(IvmFailureReason.values()).map(Enum::name),
+ Arrays.stream(IvmRefreshFailureStage.values()).map(Enum::name))
+ .collect(ImmutableList.toImmutableList());
private static final Gson GSON = new Gson();
@@ -150,7 +166,8 @@ public class MTMVTask extends AbstractTask {
new Column("LastQueryId", ScalarType.createStringType()),
new Column("ComputeGroup", ScalarType.createStringType()),
new Column("IvmFallbackReason", ScalarType.createStringType()),
- new Column("IvmRebuiltPartitions", ScalarType.createStringType()));
+ new Column("IvmRebuiltPartitions", ScalarType.createStringType()),
+ new Column("IvmRefreshMethod", ScalarType.createStringType()));
public static final ImmutableMap<String, Integer> COLUMN_TO_INDEX;
@@ -304,6 +321,28 @@ public class MTMVTask extends AbstractTask {
// INCREMENTAL request reports so that "the request was incremental but
the work was not" is visible.
@SerializedName("irp")
private int ivmRebuiltPartitions;
+ // How this task refreshed an IVM MV, by the kinds of work that committed
data. Decided once, as the task
+ // ends (see after()), so it is null while the task runs and for a task of
a non-IVM MV.
+ @SerializedName("irm")
+ private IvmRefreshMethod ivmRefreshMethod;
+ // The reason a failed task of an IVM MV reports: the IvmFailureReason a
request that may not fall back failed
+ // with, or else the IvmRefreshFailureStage the failure happened in. Null
for any other task.
+ @SerializedName("ifl")
+ private String ivmFailureReason;
+ // Whether this task fell back from the incremental refresh and an attempt
that replaced it committed data. Its
+ // reason is ivmFallbackReason. A fallback that committed nothing did not
change how the MV was refreshed.
+ @SerializedName("ifb")
+ private boolean ivmFallback;
+ // The kinds of work this task has committed, which ivmRefreshMethod is
decided from. Concurrent because a STOP
+ // decides it from the cancel thread while the worker may still be
committing.
+ private transient Set<IvmRefreshMethod> committedRefreshKinds =
ConcurrentHashMap.newKeySet();
+ // Set once the incremental attempt has given up for a reason that counts
as a fallback: what an attempt
+ // commits after that is the fallback's work. Only the worker thread reads
and writes it.
+ private transient boolean fallingBack;
+ // What this task reports as its failure reason if it fails at this point;
see ivmFailureReason. Moved along as
+ // the task passes its stages, and set to an IvmFailureReason where a
request that may not fall back fails.
+ // Only the worker thread writes it, and it is read as the worker reports
the failure.
+ private transient String failureStage =
IvmRefreshFailureStage.OTHER.name();
private long mtmvSchemaChangeVersion;
// Published only after a signature-mismatch fallback succeeds and its
task result is accepted.
private transient String refreshedIvmPlanSignature;
@@ -343,6 +382,7 @@ public class MTMVTask extends AbstractTask {
}
// Every time a task is run, the relation is regenerated because
baseTables and baseViews may change,
// such as deleting a table and creating a view with the same name
+ failureStage = IvmRefreshFailureStage.QUERY_UNUSABLE.name();
MTMVPlanUtil.QueryAnalysisResult queryAnalysis =
MTMVPlanUtil.getBaseTableFromQuery(
mtmv.getQuerySql(), ctx);
this.relation =
MTMVPlanUtil.generateMTMVRelation(queryAnalysis.getAllLevelTables(),
@@ -356,12 +396,15 @@ public class MTMVTask extends AbstractTask {
// its base-table schema and partition definition. It is not part
of
// refresh fallback: incompatible MV definitions must fail
directly.
ensureQueryUsableIfNeeded(ctx, tableIfs);
+ failureStage = IvmRefreshFailureStage.OTHER.name();
RefreshRequest request = resolveRefreshRequest();
+ failureStage = IvmRefreshFailureStage.PARTITION_SYNC_FAILED.name();
try {
syncPartitionsIfNeeded(ctx, tableIfs);
} catch (PartitionPlanningException e) {
throw new JobException(e.getMessage(), e);
}
+ failureStage = IvmRefreshFailureStage.OTHER.name();
// Partition sync has decided which partitions exist, and nothing
has read a base table yet:
// this is the point where an entry and the partition it describes
become the same thing.
// Doing it any later would let a partition that sync has just
added be refreshed without an
@@ -372,6 +415,7 @@ public class MTMVTask extends AbstractTask {
List<RefreshAttemptType> attempts = buildAttempts(request,
queryAnalysis.containsOneRowRelation());
MTMVRefreshContext refreshContext = buildRefreshContext(tableIfs);
boolean disablePartitionRefresh = false;
+ failureStage = IvmRefreshFailureStage.EXECUTION_FAILED.name();
for (RefreshAttemptType attemptType : attempts) {
switch (attemptType) {
case IVM:
@@ -379,6 +423,7 @@ public class MTMVTask extends AbstractTask {
if (ivmResult == AttemptResultType.SUCCESS) {
return;
}
+ fallingBack =
givingUpTheIncrementalAttemptIsAFallback();
if (ivmResult ==
AttemptResultType.FALLBACK_TO_COMPLETE) {
disablePartitionRefresh = true;
}
@@ -407,6 +452,7 @@ public class MTMVTask extends AbstractTask {
throw new JobException("Unsupported refresh attempt
type: " + attemptType);
}
}
+ failureStage = IvmRefreshFailureStage.OTHER.name();
throw new JobException("No refresh attempt succeeded for mv=" +
mtmv.getName());
} catch (Throwable e) {
if (getStatus() == TaskStatus.RUNNING) {
@@ -603,6 +649,8 @@ public class MTMVTask extends AbstractTask {
if (request.allowFallback && mtmv.isIvm() &&
attempts.contains(RefreshAttemptType.IVM)
&& hasUnusableIvmStream()) {
ivmFallbackReason = IvmFailureReason.STREAM_UNSUPPORTED.name();
+ // The incremental attempt is given up before it runs, which is a
fallback all the same.
+ fallingBack = true;
LOG.warn("IVM stream is unusable, mv={}, taskId={}. Continuing
with COMPLETE refresh.",
mtmv.getName(), getTaskId());
return Lists.newArrayList(RefreshAttemptType.COMPLETE);
@@ -659,6 +707,15 @@ public class MTMVTask extends AbstractTask {
ivmRebuiltPartitions = Math.max(ivmRebuiltPartitions, replaced);
}
+ /**
+ * Whether the incremental attempt giving up makes this a task that falls
back: the attempts after it rebuild what
+ * the incremental refresh could not catch up. The exception is an MV no
refresh has succeeded on yet, which has no
+ * baseline to catch up -- building the first one is not the incremental
refresh giving up.
+ */
+ private boolean givingUpTheIncrementalAttemptIsAFallback() {
+ return !INCOMPLETE_REFRESH_SNAPSHOT.equals(ivmFallbackReason) ||
mtmv.hasSucceededIvmRefresh();
+ }
+
private boolean shouldUseCompleteForInitialIvmRefresh(boolean
containsOneRowRelation) {
if (!mtmv.isIvm() || mtmv.hasRefreshSnapshot()) {
return false;
@@ -807,7 +864,7 @@ public class MTMVTask extends AbstractTask {
// A strict INCREMENTAL request must reach IVM so the stream can
establish its initial baseline.
// Only fallback-enabled requests may rebuild a missing or invalidated
snapshot with COMPLETE.
if (!mtmv.hasRefreshSnapshot() && request.allowFallback) {
- ivmFallbackReason = "INCOMPLETE_REFRESH_SNAPSHOT";
+ ivmFallbackReason = INCOMPLETE_REFRESH_SNAPSHOT;
LOG.warn("IVM refresh fell back for mv={},
reason=INCOMPLETE_REFRESH_SNAPSHOT, taskId={}. "
+ "Continuing with COMPLETE refresh.", mtmv.getName(),
getTaskId());
return AttemptResultType.FALLBACK_TO_COMPLETE;
@@ -893,12 +950,14 @@ public class MTMVTask extends AbstractTask {
adoptPartitionsCreatedByTheRetry(dirtyPartitions);
currentRefreshContext = buildRefreshContext(tableIfs);
} catch (Exception e) {
+ failureStage =
IvmRefreshFailureStage.PARTITION_SYNC_FAILED.name();
throw new JobException("Failed to synchronize MV partitions
before IVM retry for mv="
+ mtmv.getName(), e);
}
LOG.warn("Retrying IVM refresh after synchronizing MV partitions,
mv={}, attempt={}/{}, taskId={}",
mtmv.getName(), partitionSyncRetryCount + 1,
ivmAttemptLimit, getTaskId());
}
+ failureStage = IvmFailureReason.MV_PARTITION_NOT_FOUND.name();
throw new JobException("IVM refresh could not recover missing MV
partition for mv="
+ mtmv.getName() + ", detail=" + ivmResult.getDetailMessage());
}
@@ -961,6 +1020,7 @@ public class MTMVTask extends AbstractTask {
+ ", detail=" + Util.getRootCauseMessage(e), e);
}
if (ivmResult.isSuccess()) {
+ recordCommitted(IvmRefreshMethod.INCREMENTAL);
this.partitionSnapshots.putAll(capturedSnapshots);
recordRefreshCompleted(incrementalScope);
commitCapturedEpochs(capturedEpochs);
@@ -1014,6 +1074,18 @@ public class MTMVTask extends AbstractTask {
completedPartitions.addAll(partitions);
}
+ /**
+ * Notes that a batch of this kind of work committed data, which is what
ivmRefreshMethod is decided from. What
+ * commits once the incremental attempt has given up is the fallback's
work, and it is what makes this a task
+ * that fell back: a fallback that commits nothing leaves the MV as the
incremental refresh would have.
+ */
+ private void recordCommitted(IvmRefreshMethod kind) {
+ committedRefreshKinds.add(kind);
+ if (fallingBack) {
+ ivmFallback = true;
+ }
+ }
+
/**
* Says durably that the parts of the scope that do not name a rebuild
requirement have to be rebuilt, and
* brings the epochs this phase records in line with it.
@@ -1206,6 +1278,7 @@ public class MTMVTask extends AbstractTask {
throws JobException {
ivmFallbackReason = ivmResult.getFailureReason().name();
if (!request.allowFallback) {
+ failureStage = ivmResult.getFailureReason().name();
throw new JobException(
"IVM incremental refresh failed for mv=" + mtmv.getName()
+ ", reason=" + ivmResult.getFailureReason()
@@ -1246,6 +1319,7 @@ public class MTMVTask extends AbstractTask {
if (mtmv.isIvm()
&& hasUnusableIvmStreamForPartitions(partitionPlan.context,
partitionPlan.partitions)) {
if (!request.allowFallback) {
+ failureStage = IvmFailureReason.STREAM_UNSUPPORTED.name();
throw new JobException("IVM stream is unusable for the
partitions of this refresh, mv="
+ mtmv.getName());
}
@@ -1349,6 +1423,8 @@ public class MTMVTask extends AbstractTask {
mtmv.getName(), getTaskId(), e);
throw new JobException(e.getMessage(), e);
}
+ recordCommitted(refreshMode == RefreshMode.COMPLETE
+ ? IvmRefreshMethod.COMPLETE :
IvmRefreshMethod.PARTITION_REBUILD);
recordRefreshCompleted(execPartitionNames);
// What this batch may record: everything it replaced, unless the
read behind it answered with a
// base table the MV does not partition by as of an older state
than that table is in now -- which
@@ -1905,6 +1981,8 @@ public class MTMVTask extends AbstractTask {
trow.addToColumnValue(new TCell().setStringVal(
ivmFallbackReason == null ? FeConstants.null_string :
ivmFallbackReason));
trow.addToColumnValue(new
TCell().setStringVal(String.valueOf(ivmRebuiltPartitions)));
+ trow.addToColumnValue(new TCell().setStringVal(
+ ivmRefreshMethod == null ? FeConstants.null_string :
ivmRefreshMethod.name()));
return trow;
}
@@ -1926,6 +2004,16 @@ public class MTMVTask extends AbstractTask {
private void after() {
if (mtmv != null) {
+ if (mtmv.isIvm()) {
+ // Decided here, once, and before the result is journaled: a
STOP publishes from the cancel thread
+ // while the worker may still commit a batch, and what the
journal carries has to be what the result
+ // was recorded and counted with.
+ ivmRefreshMethod = IvmRefreshMethod.of(committedRefreshKinds);
+ if (getStatus() == TaskStatus.FAILED) {
+ ivmFailureReason = failureStage;
+ }
+ recordIvmRefreshMetrics();
+ }
// A copy, because this is not always the thread that fills the
map: a STOP publishes from the
// cancel thread and `cancel(false)` does not wait for the
execution to stop, so the batches the
// worker commits afterwards would otherwise land in a result that
was already built -- and
@@ -1940,6 +2028,60 @@ public class MTMVTask extends AbstractTask {
}
+ /**
+ * Counts this task in the IVM refresh metrics as it ends. Only the Master
runs a task, and the replay of its
+ * result does not come here, so a follower never counts it again.
+ */
+ private void recordIvmRefreshMetrics() {
+ if (!MetricRepo.isInit) {
+ return;
+ }
+ String result;
+ switch (getStatus()) {
+ case SUCCESS:
+ result = ivmRefreshMethod == IvmRefreshMethod.NONE
+ ? MetricRepo.IVM_REFRESH_RESULT_NO_CHANGE :
MetricRepo.IVM_REFRESH_RESULT_SUCCESS;
+ break;
+ case FAILED:
+ result = MetricRepo.IVM_REFRESH_RESULT_FAILED;
+ break;
+ case CANCELED:
+ result = MetricRepo.IVM_REFRESH_RESULT_CANCELED;
+ break;
+ default:
+ throw new IllegalStateException("A task cannot end as " +
getStatus() + ", taskId=" + getTaskId());
+ }
+
MetricRepo.COUNTER_IVM_REFRESH_TASKS.get(result).get(ivmRefreshMethod).increase(1L);
+ if (getStatus() == TaskStatus.CANCELED) {
+ // A canceled task has no finish time to measure it by, and a
fallback it started is not one it finished.
+ return;
+ }
+ MetricRepo.COUNTER_IVM_REFRESH_DURATION_SECONDS.get(ivmRefreshMethod)
+ .increase((getFinishTimeMs() - getStartTimeMs()) / 1000.0);
+ if (ivmFallback) {
+
MetricRepo.COUNTER_IVM_FALLBACKS.get(ivmFallbackReason).increase(1L);
+ }
+ if (getStatus() == TaskStatus.FAILED) {
+
MetricRepo.COUNTER_IVM_REFRESH_FAILURES.get(ivmFailureReason).increase(1L);
+ }
+ }
+
+ /** Counts a task of an IVM MV that the full task queue turned away: it
never runs, so nothing else counts it. */
+ @Override
+ public void onRejected() {
+ MTMV rejectedMv;
+ try {
+ rejectedMv = MTMVUtil.getMTMV(dbId, mtmvId);
+ } catch (UserException e) {
+ // dropped since the task was created
+ LOG.warn("can not find mv of the rejected task {}", getTaskId(),
e);
+ return;
+ }
+ if (MetricRepo.isInit && rejectedMv.isIvm()) {
+ MetricRepo.COUNTER_IVM_SCHEDULE_REJECTED.increase(1L);
+ }
+ }
+
@Override
protected void closeOrReleaseResources() {
if (null != mtmv) {
@@ -2009,6 +2151,22 @@ public class MTMVTask extends AbstractTask {
return refreshedIvmPlanSignature;
}
+ public IvmRefreshMethod getIvmRefreshMethod() {
+ return ivmRefreshMethod;
+ }
+
+ public String getIvmFailureReason() {
+ return ivmFailureReason;
+ }
+
+ public boolean isIvmFallback() {
+ return ivmFallback;
+ }
+
+ public String getIvmFallbackReason() {
+ return ivmFallbackReason;
+ }
+
@Override
public String toString() {
return "MTMVTask{"
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/task/AbstractTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/task/AbstractTask.java
index f597998c0c1..94b6cff01af 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/task/AbstractTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/task/AbstractTask.java
@@ -163,6 +163,13 @@ public abstract class AbstractTask implements Task {
}
}
+ /**
+ * Called when the executor turns this task away because its queue is
full, so the task never runs. The caller
+ * goes on to fail or cancel it as it would any other task.
+ */
+ public void onRejected() {
+ }
+
/**
* Abstract method for implementing the task-specific cancellation logic.
* Subclasses must override this method to provide their own
implementation of how a task should be canceled.
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/metric/DoubleCounterMetric.java
b/fe/fe-core/src/main/java/org/apache/doris/metric/DoubleCounterMetric.java
new file mode 100644
index 00000000000..7597fac3a54
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/metric/DoubleCounterMetric.java
@@ -0,0 +1,43 @@
+// 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.metric;
+
+import java.util.concurrent.atomic.DoubleAdder;
+
+/**
+ * A counter of a quantity with a fractional part, such as a total in seconds,
which a LongCounterMetric would have to
+ * round each addition of.
+ */
+public class DoubleCounterMetric extends CounterMetric<Double> {
+
+ public DoubleCounterMetric(String name, MetricUnit unit, String
description) {
+ super(name, unit, description);
+ }
+
+ private final DoubleAdder value = new DoubleAdder();
+
+ @Override
+ public void increase(Double delta) {
+ value.add(delta);
+ }
+
+ @Override
+ public Double getValue() {
+ return value.sum();
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java
b/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java
index c54a93eaefe..e49eba0a02d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java
@@ -37,6 +37,7 @@ import org.apache.doris.job.common.JobStatus;
import org.apache.doris.job.common.TaskStatus;
import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
import org.apache.doris.job.extensions.insert.streaming.StreamingJobStatistic;
+import org.apache.doris.job.extensions.mtmv.MTMVTask;
import org.apache.doris.load.EtlJobType;
import org.apache.doris.load.loadv2.JobState;
import org.apache.doris.load.loadv2.LoadManager;
@@ -50,6 +51,7 @@ import
org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
import org.apache.doris.mtmv.MTMVRefreshEnum.RefreshMethod;
import org.apache.doris.mtmv.MTMVRefreshEnum.RefreshTrigger;
import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mtmv.ivm.IvmRefreshMethod;
import org.apache.doris.mysql.privilege.Auth;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.persist.EditLog;
@@ -272,6 +274,20 @@ public final class MetricRepo {
public static GaugeMetric<Integer>
GAUGE_ASYNC_MATERIALIZED_VIEW_TASK_RUNNING_NUM;
public static GaugeMetric<Integer>
GAUGE_ASYNC_MATERIALIZED_VIEW_TASK_PENDING_NUM;
+ // The refresh tasks of IVM materialized views, counted on the Master as
each one ends.
+ // The result label of COUNTER_IVM_REFRESH_TASKS: no_change is a SUCCESS
that committed nothing.
+ public static final String IVM_REFRESH_RESULT_SUCCESS = "success";
+ public static final String IVM_REFRESH_RESULT_NO_CHANGE = "no_change";
+ public static final String IVM_REFRESH_RESULT_FAILED = "failed";
+ public static final String IVM_REFRESH_RESULT_CANCELED = "canceled";
+ // by result, then by IvmRefreshMethod
+ public static Map<String, Map<IvmRefreshMethod, LongCounterMetric>>
COUNTER_IVM_REFRESH_TASKS;
+ public static Map<IvmRefreshMethod, DoubleCounterMetric>
COUNTER_IVM_REFRESH_DURATION_SECONDS;
+ // by MTMVTask.IVM_FALLBACK_REASONS and MTMVTask.IVM_FAILURE_REASONS
+ public static Map<String, LongCounterMetric> COUNTER_IVM_FALLBACKS;
+ public static Map<String, LongCounterMetric> COUNTER_IVM_REFRESH_FAILURES;
+ public static LongCounterMetric COUNTER_IVM_SCHEDULE_REJECTED;
+
// Statistics
public static GaugeMetric<Double> GAUGE_STATISTICS_UNHEALTHY_TABLE_RATE;
public static GaugeMetric<Double> GAUGE_STATISTICS_UNHEALTHY_COLUMN_RATE;
@@ -1146,6 +1162,44 @@ public final class MetricRepo {
}
};
DORIS_METRIC_REGISTER.addMetrics(GAUGE_ASYNC_MATERIALIZED_VIEW_TASK_PENDING_NUM);
+
+ // IVM refresh tasks, counted by MTMVTask as each task of an IVM MV
ends. Every combination of labels is
+ // registered here, so that each series exists from 0 rather than
appearing with its first event.
+ COUNTER_IVM_REFRESH_TASKS = Maps.newHashMap();
+ for (String result : new String[] {IVM_REFRESH_RESULT_SUCCESS,
IVM_REFRESH_RESULT_NO_CHANGE,
+ IVM_REFRESH_RESULT_FAILED, IVM_REFRESH_RESULT_CANCELED}) {
+ Map<IvmRefreshMethod, LongCounterMetric> byMethod =
Maps.newEnumMap(IvmRefreshMethod.class);
+ for (IvmRefreshMethod method : IvmRefreshMethod.values()) {
+ LongCounterMetric counter = new
LongCounterMetric("ivm_refresh_tasks_total", MetricUnit.NOUNIT,
+ "number of the refresh tasks of IVM materialized views
that ended, by result and by the"
+ + " kinds of work that committed data");
+ counter.addLabel(new MetricLabel("result", result))
+ .addLabel(new MetricLabel("method",
method.name().toLowerCase()));
+ DORIS_METRIC_REGISTER.addMetrics(counter);
+ byMethod.put(method, counter);
+ }
+ COUNTER_IVM_REFRESH_TASKS.put(result, byMethod);
+ }
+ COUNTER_IVM_REFRESH_DURATION_SECONDS =
Maps.newEnumMap(IvmRefreshMethod.class);
+ for (IvmRefreshMethod method : IvmRefreshMethod.values()) {
+ DoubleCounterMetric counter = new
DoubleCounterMetric("ivm_refresh_duration_seconds_total",
+ MetricUnit.SECONDS, "how long the refresh tasks of IVM
materialized views that succeeded or"
+ + " failed ran, from start to finish, in seconds");
+ counter.addLabel(new MetricLabel("method",
method.name().toLowerCase()));
+ DORIS_METRIC_REGISTER.addMetrics(counter);
+ COUNTER_IVM_REFRESH_DURATION_SECONDS.put(method, counter);
+ }
+ COUNTER_IVM_FALLBACKS =
addIvmRefreshReasonCounters("ivm_fallbacks_total",
+ "number of the refresh tasks of IVM materialized views that
fell back from the incremental refresh"
+ + " and committed data in the refresh that replaced
it, by the reason",
+ MTMVTask.IVM_FALLBACK_REASONS);
+ COUNTER_IVM_REFRESH_FAILURES =
addIvmRefreshReasonCounters("ivm_refresh_failures_total",
+ "number of the refresh tasks of IVM materialized views that
failed, by the reason",
+ MTMVTask.IVM_FAILURE_REASONS);
+ COUNTER_IVM_SCHEDULE_REJECTED = new
LongCounterMetric("ivm_schedule_rejected_total", MetricUnit.NOUNIT,
+ "number of the refresh tasks of IVM materialized views that
the full task queue turned away");
+ DORIS_METRIC_REGISTER.addMetrics(COUNTER_IVM_SCHEDULE_REJECTED);
+
GAUGE_MAX_TABLE_SIZE_BYTES = new
GaugeMetricImpl<>("max_table_size_bytes", MetricUnit.BYTES, "", 0L);
DORIS_METRIC_REGISTER.addMetrics(GAUGE_MAX_TABLE_SIZE_BYTES);
@@ -1761,6 +1815,19 @@ public final class MetricRepo {
return visitor.finish();
}
+ // One counter per reason, keyed by the reason as the task records it and
labeled with it in lower case.
+ private static Map<String, LongCounterMetric>
addIvmRefreshReasonCounters(String name, String description,
+ List<String> reasons) {
+ Map<String, LongCounterMetric> counters = Maps.newHashMap();
+ for (String reason : reasons) {
+ LongCounterMetric counter = new LongCounterMetric(name,
MetricUnit.NOUNIT, description);
+ counter.addLabel(new MetricLabel("reason", reason.toLowerCase()));
+ DORIS_METRIC_REGISTER.addMetrics(counter);
+ counters.put(reason, counter);
+ }
+ return counters;
+ }
+
public static <M extends Metric<?>> AutoMappedMetric<M>
addLabeledMetrics(String label, Supplier<M> metric) {
return new AutoMappedMetric<>(value -> {
M m = metric.get();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRefreshFailureStage.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRefreshFailureStage.java
new file mode 100644
index 00000000000..2da597b088d
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRefreshFailureStage.java
@@ -0,0 +1,37 @@
+// 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.mtmv.ivm;
+
+/**
+ * The stage of a refresh task of an IVM MV that a failure happened in, which
is the reason a failed task reports
+ * when no {@link IvmFailureReason} names it.
+ *
+ * <p>An {@link IvmFailureReason} names the failures of the incremental
refresh itself, which a request that may
+ * not fall back fails with. Every other failure is told apart by where it was
thrown rather than by its message,
+ * so that the reasons stay a fixed set.
+ */
+public enum IvmRefreshFailureStage {
+ /** The MV's query could not be analyzed against its base tables, or can
no longer be used. */
+ QUERY_UNUSABLE,
+ /** The MV's partitions could not be synchronized with the partitions of
its base tables. */
+ PARTITION_SYNC_FAILED,
+ /** A refresh attempt failed while it ran. */
+ EXECUTION_FAILED,
+ /** Any other stage: resolving the request, deciding the attempts, or
after every attempt gave up. */
+ OTHER
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRefreshMethod.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRefreshMethod.java
new file mode 100644
index 00000000000..3ba894099c4
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRefreshMethod.java
@@ -0,0 +1,55 @@
+// 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.mtmv.ivm;
+
+import java.util.Set;
+
+/**
+ * How a refresh task of an IVM MV actually refreshed it, judged by the kinds
of work that committed data.
+ *
+ * <p>The request does not say it: an AUTO or INCREMENTAL FALLBACK request may
end in the incremental refresh, a
+ * partition rebuild, a whole-MV rebuild, or several of them, and which of
them wrote data is what tells a refresh
+ * that kept the MV up incrementally from one that recomputed it. The refresh
mode a task reports is a different
+ * question -- which partitions it covers -- so the two are reported side by
side.
+ */
+public enum IvmRefreshMethod {
+ /** Nothing was committed: every partition was up to date, or the task
ended before writing anything. */
+ NONE,
+ /** Only the incremental refresh committed data. */
+ INCREMENTAL,
+ /**
+ * Only partition rebuilds committed data. The rebuild an incremental
attempt runs first, for the partitions the
+ * criterion says cannot be caught up, is one of them.
+ */
+ PARTITION_REBUILD,
+ /** Only a whole-MV rebuild committed data. */
+ COMPLETE,
+ /** More than one kind committed data, typically the rebuild of
invalidated partitions and then the delta. */
+ MIXED;
+
+ /** The method of a task whose committed work was of these kinds, which
are any of the three single kinds. */
+ public static IvmRefreshMethod of(Set<IvmRefreshMethod> committedKinds) {
+ if (committedKinds.isEmpty()) {
+ return NONE;
+ }
+ if (committedKinds.size() > 1) {
+ return MIXED;
+ }
+ return committedKinds.iterator().next();
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRefreshSummary.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRefreshSummary.java
new file mode 100644
index 00000000000..b0951b1ad9f
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRefreshSummary.java
@@ -0,0 +1,142 @@
+// 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.mtmv.ivm;
+
+import org.apache.doris.job.extensions.mtmv.MTMVTask;
+
+import com.google.gson.annotations.SerializedName;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * What the refresh tasks of an IVM MV have come to: consecutive failures, the
latest success and failure, how the
+ * latest task refreshed the MV and how long it took, and how often a refresh
fell back from the incremental path.
+ *
+ * <p>Kept on the MV rather than read from its task history, which holds only
the latest
+ * max_persistence_task_count tasks. It is updated from each task result as
the MV records it, on the Master and in
+ * the replay of the same journal record alike, and only from what that record
carries, so a follower that takes
+ * over keeps the same summary.
+ *
+ * <p>A canceled task leaves it as it is: it neither succeeded nor failed, and
it has no finish time to measure it
+ * by.
+ */
+public class IvmRefreshSummary {
+ // What the latest failure says, cut to this many characters.
+ public static final int MAX_ERROR_MESSAGE_LENGTH = 512;
+
+ // Failed tasks since the latest successful one.
+ @SerializedName("cf")
+ private int consecutiveFailures;
+ // Finish times in milliseconds, 0 when there has been none.
+ @SerializedName("lst")
+ private long lastSuccessTimeMs;
+ @SerializedName("lft")
+ private long lastFailureTimeMs;
+ // An IvmFailureReason or an IvmRefreshFailureStage; null for a failure
recorded before the task carried it.
+ @SerializedName("lfr")
+ private String lastFailureReason;
+ @SerializedName("lem")
+ private String lastErrorMessage;
+ // Of the latest task that succeeded or failed.
+ @SerializedName("lrm")
+ private IvmRefreshMethod lastRefreshMethod;
+ @SerializedName("lrd")
+ private long lastRefreshDurationMs;
+ @SerializedName("fc")
+ private long fallbackCount;
+ @SerializedName("lfbr")
+ private String lastFallbackReason;
+ @SerializedName("lfbt")
+ private long lastFallbackTimeMs;
+ // Whether a refresh has ever succeeded: until one has, the MV has no
baseline for the incremental refresh to
+ // keep up, and building it is not the incremental refresh falling back.
+ @SerializedName("s")
+ private boolean succeeded;
+
+ /** Takes in the result of a task, which has ended as SUCCESS, FAILED or
CANCELED. */
+ public void update(MTMVTask task) {
+ switch (task.getStatus()) {
+ case SUCCESS:
+ consecutiveFailures = 0;
+ lastSuccessTimeMs = task.getFinishTimeMs();
+ succeeded = true;
+ break;
+ case FAILED:
+ consecutiveFailures++;
+ lastFailureTimeMs = task.getFinishTimeMs();
+ lastFailureReason = task.getIvmFailureReason();
+ lastErrorMessage = StringUtils.abbreviate(task.getErrMsg(),
MAX_ERROR_MESSAGE_LENGTH);
+ break;
+ case CANCELED:
+ return;
+ default:
+ throw new IllegalStateException("A task result cannot be " +
task.getStatus()
+ + ", taskId=" + task.getTaskId());
+ }
+ lastRefreshMethod = task.getIvmRefreshMethod();
+ lastRefreshDurationMs = task.getFinishTimeMs() - task.getStartTimeMs();
+ if (task.isIvmFallback()) {
+ fallbackCount++;
+ lastFallbackReason = task.getIvmFallbackReason();
+ lastFallbackTimeMs = task.getFinishTimeMs();
+ }
+ }
+
+ public int getConsecutiveFailures() {
+ return consecutiveFailures;
+ }
+
+ public long getLastSuccessTimeMs() {
+ return lastSuccessTimeMs;
+ }
+
+ public long getLastFailureTimeMs() {
+ return lastFailureTimeMs;
+ }
+
+ public String getLastFailureReason() {
+ return lastFailureReason;
+ }
+
+ public String getLastErrorMessage() {
+ return lastErrorMessage;
+ }
+
+ public IvmRefreshMethod getLastRefreshMethod() {
+ return lastRefreshMethod;
+ }
+
+ public long getLastRefreshDurationMs() {
+ return lastRefreshDurationMs;
+ }
+
+ public long getFallbackCount() {
+ return fallbackCount;
+ }
+
+ public String getLastFallbackReason() {
+ return lastFallbackReason;
+ }
+
+ public long getLastFallbackTimeMs() {
+ return lastFallbackTimeMs;
+ }
+
+ public boolean hasSucceeded() {
+ return succeeded;
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/executor/TaskProcessorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/executor/TaskProcessorTest.java
index 526fa76515d..6d9c433c7b6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/executor/TaskProcessorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/executor/TaskProcessorTest.java
@@ -180,6 +180,52 @@ public class TaskProcessorTest {
}
}
+ /** A task the full queue turns away is told so, and only that one: the
others are accepted and run. */
+ @Test
+ public void testATaskTheFullQueueTurnsAwayIsToldSo() throws Exception {
+ CountDownLatch running = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger rejections = new AtomicInteger();
+ TaskProcessor taskProcessor = new TaskProcessor(1, 1, Thread::new);
+ try {
+ Assertions.assertTrue(taskProcessor.addTask(countingRejections(1L,
rejections, () -> {
+ running.countDown();
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ })));
+ Assertions.assertTrue(running.await(10, TimeUnit.SECONDS));
+ // The only worker is busy, and the queue has room for one more
task.
+ Assertions.assertTrue(taskProcessor.addTask(countingRejections(2L,
rejections, () -> { })));
+ Assertions.assertEquals(0, rejections.get());
+
+
Assertions.assertFalse(taskProcessor.addTask(countingRejections(3L, rejections,
() -> { })));
+ Assertions.assertEquals(1, rejections.get());
+ } finally {
+ release.countDown();
+ taskProcessor.shutdown();
+ }
+ }
+
+ private static AbstractTask countingRejections(long taskId, AtomicInteger
rejections, Runnable body) {
+ AbstractTask task = new InsertTask("task-" + taskId, null, null, null)
{
+ @Override
+ public void runTask() {
+ body.run();
+ }
+
+ @Override
+ public void onRejected() {
+ rejections.incrementAndGet();
+ }
+ };
+ task.setStatus(TaskStatus.PENDING);
+ task.setTaskId(taskId);
+ return task;
+ }
+
private static void installStatementScope(AtomicInteger closeCount,
CountDownLatch scopeInstalled) {
ConnectContext connectContext = new ConnectContext();
connectContext.setThreadLocalInfo();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/metric/IvmRefreshMetricsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/metric/IvmRefreshMetricsTest.java
new file mode 100644
index 00000000000..bdb5ad61b09
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/metric/IvmRefreshMetricsTest.java
@@ -0,0 +1,161 @@
+// 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.metric;
+
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.common.MetaNotFoundException;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.job.common.TaskStatus;
+import org.apache.doris.job.extensions.mtmv.MTMVTask;
+import org.apache.doris.job.extensions.mtmv.MTMVTask.MTMVTaskTriggerMode;
+import org.apache.doris.job.extensions.mtmv.MTMVTaskContext;
+import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mtmv.ivm.IvmFailureReason;
+import org.apache.doris.mtmv.ivm.IvmRefreshFailureStage;
+import org.apache.doris.mtmv.ivm.IvmRefreshMethod;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+/** The counters of the refresh tasks of IVM materialized views, and how a
task is counted in them as it ends. */
+public class IvmRefreshMetricsTest {
+
+ @BeforeAll
+ public static void setUp() {
+ FeConstants.runningUnitTest = true;
+ MetricRepo.init();
+ }
+
+ /**
+ * Every series is registered up front, so each one is there from 0: the
tasks by every result and method, the
+ * run time by method, and one series per reason a task can fall back or
fail with, labeled in lower case.
+ */
+ @Test
+ public void testEverySeriesIsThereFromZero() {
+ Assertions.assertEquals(4 * IvmRefreshMethod.values().length,
+ MetricRepo.getMetricsByName("ivm_refresh_tasks_total").size());
+ Assertions.assertEquals(IvmRefreshMethod.values().length,
+
MetricRepo.getMetricsByName("ivm_refresh_duration_seconds_total").size());
+ Assertions.assertEquals(IvmFailureReason.values().length + 1,
+ MetricRepo.getMetricsByName("ivm_fallbacks_total").size());
+ Assertions.assertEquals(IvmFailureReason.values().length +
IvmRefreshFailureStage.values().length,
+
MetricRepo.getMetricsByName("ivm_refresh_failures_total").size());
+
+ // Series no other case here counts in, so they are still 0 whatever
ran before.
+ String metrics = getPrometheusMetrics();
+ for (String type : new String[] {"ivm_refresh_tasks_total",
"ivm_refresh_duration_seconds_total",
+ "ivm_fallbacks_total", "ivm_refresh_failures_total",
"ivm_schedule_rejected_total"}) {
+ Assertions.assertTrue(metrics.contains("# TYPE doris_fe_" + type +
" counter\n"), type);
+ }
+ Assertions.assertTrue(metrics.contains(
+ "doris_fe_ivm_refresh_tasks_total{result=\"canceled\",
method=\"mixed\"} 0\n"), metrics);
+ Assertions.assertTrue(metrics.contains(
+
"doris_fe_ivm_refresh_duration_seconds_total{method=\"partition_rebuild\"}
0.0\n"), metrics);
+ Assertions.assertTrue(metrics.contains(
+
"doris_fe_ivm_fallbacks_total{reason=\"incomplete_refresh_snapshot\"} 0\n"),
metrics);
+ Assertions.assertTrue(metrics.contains(
+
"doris_fe_ivm_refresh_failures_total{reason=\"execution_failed\"} 0\n"),
metrics);
+ }
+
+ /**
+ * A task is counted by its result and method as it ends; one that ran to
completion adds its run time, one that
+ * fell back counts its reason, and one that failed counts why. A SUCCESS
that committed nothing is no_change, and
+ * a canceled task has no finish time to add and finished no fallback.
+ */
+ @Test
+ public void testATaskIsCountedAsItEnds() {
+ MTMVTask fellBack = endedTask(TaskStatus.SUCCESS,
IvmRefreshMethod.COMPLETE, 1000L, 3500L);
+ Deencapsulation.setField(fellBack, "ivmFallback", true);
+ Deencapsulation.setField(fellBack, "ivmFallbackReason",
IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name());
+ assertCounted(fellBack,
MetricRepo.COUNTER_IVM_REFRESH_TASKS.get("success").get(IvmRefreshMethod.COMPLETE));
+ Assertions.assertEquals(2.5,
+
MetricRepo.COUNTER_IVM_REFRESH_DURATION_SECONDS.get(IvmRefreshMethod.COMPLETE).getValue(),
1e-9);
+ Assertions.assertEquals(1L,
+
MetricRepo.COUNTER_IVM_FALLBACKS.get(IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name()).getValue());
+
+ assertCounted(endedTask(TaskStatus.SUCCESS, IvmRefreshMethod.NONE,
1000L, 1200L),
+
MetricRepo.COUNTER_IVM_REFRESH_TASKS.get("no_change").get(IvmRefreshMethod.NONE));
+ Assertions.assertEquals(0L,
MetricRepo.COUNTER_IVM_REFRESH_TASKS.get("success").get(IvmRefreshMethod.NONE)
+ .getValue());
+
+ MTMVTask failed = endedTask(TaskStatus.FAILED,
IvmRefreshMethod.INCREMENTAL, 1000L, 2000L);
+ Deencapsulation.setField(failed, "ivmFailureReason",
IvmRefreshFailureStage.QUERY_UNUSABLE.name());
+ assertCounted(failed,
MetricRepo.COUNTER_IVM_REFRESH_TASKS.get("failed").get(IvmRefreshMethod.INCREMENTAL));
+ Assertions.assertEquals(1L,
+
MetricRepo.COUNTER_IVM_REFRESH_FAILURES.get(IvmRefreshFailureStage.QUERY_UNUSABLE.name()).getValue());
+
+ MTMVTask canceled = endedTask(TaskStatus.CANCELED,
IvmRefreshMethod.PARTITION_REBUILD, 1000L, 2000L);
+ Deencapsulation.setField(canceled, "ivmFallback", true);
+ Deencapsulation.setField(canceled, "ivmFallbackReason",
IvmFailureReason.CHANGELOG_BROKEN.name());
+ assertCounted(canceled,
+
MetricRepo.COUNTER_IVM_REFRESH_TASKS.get("canceled").get(IvmRefreshMethod.PARTITION_REBUILD));
+ Assertions.assertEquals(0.0,
+
MetricRepo.COUNTER_IVM_REFRESH_DURATION_SECONDS.get(IvmRefreshMethod.PARTITION_REBUILD).getValue(),
+ 1e-9);
+ Assertions.assertEquals(0L,
+
MetricRepo.COUNTER_IVM_FALLBACKS.get(IvmFailureReason.CHANGELOG_BROKEN.name()).getValue());
+ }
+
+ /**
+ * A task the full task queue turns away never runs, so it is counted
where it is turned away -- if its MV is an
+ * IVM MV, and still there.
+ */
+ @Test
+ public void testATaskOfAnIvmMvTheQueueTurnsAwayIsCounted() {
+ MTMV ivmMv = Mockito.mock(MTMV.class);
+ Mockito.when(ivmMv.isIvm()).thenReturn(true);
+ long rejected = MetricRepo.COUNTER_IVM_SCHEDULE_REJECTED.getValue();
+ try (MockedStatic<MTMVUtil> mtmvUtil =
Mockito.mockStatic(MTMVUtil.class)) {
+ mtmvUtil.when(() -> MTMVUtil.getMTMV(1L, 10L)).thenReturn(ivmMv);
+ mtmvUtil.when(() -> MTMVUtil.getMTMV(1L,
20L)).thenReturn(Mockito.mock(MTMV.class));
+ mtmvUtil.when(() -> MTMVUtil.getMTMV(1L, 30L)).thenThrow(new
MetaNotFoundException("dropped"));
+
+ new MTMVTask(1L, 10L, new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM)).onRejected();
+ Assertions.assertEquals(rejected + 1,
MetricRepo.COUNTER_IVM_SCHEDULE_REJECTED.getValue());
+ new MTMVTask(1L, 20L, new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM)).onRejected();
+ new MTMVTask(1L, 30L, new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM)).onRejected();
+ Assertions.assertEquals(rejected + 1,
MetricRepo.COUNTER_IVM_SCHEDULE_REJECTED.getValue());
+ }
+ }
+
+ private static void assertCounted(MTMVTask task, LongCounterMetric
counter) {
+ long before = counter.getValue();
+ Deencapsulation.invoke(task, "recordIvmRefreshMetrics");
+ Assertions.assertEquals(before + 1, counter.getValue());
+ }
+
+ private static MTMVTask endedTask(TaskStatus status, IvmRefreshMethod
method, long startTimeMs,
+ long finishTimeMs) {
+ MTMVTask task = new MTMVTask();
+ task.setStatus(status);
+ task.setStartTimeMs(startTimeMs);
+ task.setFinishTimeMs(finishTimeMs);
+ Deencapsulation.setField(task, "ivmRefreshMethod", method);
+ return task;
+ }
+
+ private static String getPrometheusMetrics() {
+ MetricVisitor visitor = new PrometheusMetricVisitor();
+ MetricRepo.DORIS_METRIC_REGISTER.accept(visitor);
+ return visitor.finish();
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
index 44b70074c4f..496e44696b9 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
@@ -53,6 +53,8 @@ import org.apache.doris.mtmv.ivm.IvmIncrRefreshResult;
import org.apache.doris.mtmv.ivm.IvmInfo;
import org.apache.doris.mtmv.ivm.IvmPlanSignature;
import org.apache.doris.mtmv.ivm.IvmPlanSignatureGenerator;
+import org.apache.doris.mtmv.ivm.IvmRefreshFailureStage;
+import org.apache.doris.mtmv.ivm.IvmRefreshMethod;
import org.apache.doris.mtmv.ivm.IvmRewriteResult;
import org.apache.doris.mtmv.ivm.IvmUtil;
import org.apache.doris.nereids.CascadesContext;
@@ -663,6 +665,8 @@ public class MTMVTaskTest {
Assertions.assertEquals(Lists.newArrayList("COMPLETE"),
toNames(attempts));
Assertions.assertEquals(IvmFailureReason.STREAM_UNSUPPORTED.name(),
Deencapsulation.getField(task, "ivmFallbackReason"));
+ // The incremental attempt is given up before it runs: what COMPLETE
commits is a fallback.
+ Assertions.assertTrue((boolean) Deencapsulation.getField(task,
"fallingBack"));
}
@Test
@@ -1859,6 +1863,8 @@ public class MTMVTaskTest {
Assertions.assertTrue(failure.getMessage().contains(ChangelogUtils.CHANGELOG_OFFSET_EXPIRED));
Assertions.assertEquals(IvmFailureReason.CHANGELOG_BROKEN.name(),
Deencapsulation.getField(task, "ivmFallbackReason"));
+ // and that is the reason the failed task reports, rather than the
stage the rebuild ran in
+ Assertions.assertEquals(IvmFailureReason.CHANGELOG_BROKEN.name(),
Deencapsulation.getField(task, "failureStage"));
}
/**
@@ -2246,4 +2252,305 @@ public class MTMVTaskTest {
task.registerExecutor(null);
Assertions.assertNull(Deencapsulation.getField(task, "executor"));
}
+
+ /**
+ * The rebuild an incremental attempt runs first, for an invalidated
partition, commits data of its own. Followed
+ * by the delta, the task refreshed the MV in two ways, which is MIXED --
and it is not a fallback, because the
+ * incremental refresh never gave up.
+ */
+ @Test
+ public void testARebuildBeforeTheDeltaIsAMixedRefreshAndNotAFallback()
throws Exception {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ // An append-only detail MV captures no plan signature, which keeps
the mocked batch from needing one.
+ Mockito.when(mtmv.isIvmAppendOnlyDupTable()).thenReturn(true);
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+
Mockito.when(mtmv.getDatabase()).thenReturn(Mockito.mock(Database.class));
+ Mockito.when(mtmv.getRefreshPartitionNum()).thenReturn(1);
+
Mockito.when(mtmvPartitionInfo.getPctInfos()).thenReturn(Collections.emptyList());
+
Mockito.when(mtmv.getPartitionStates()).thenReturn(Maps.newHashMap(Map.of(
+ poneName, new MTMVPartitionState(1, 2),
+ ptwoName, new MTMVPartitionState(2, 2))));
+ MTMVRefreshContext refreshContext =
Mockito.mock(MTMVRefreshContext.class);
+
Mockito.when(refreshContext.getByPartitionName(Mockito.anyString())).thenReturn(Collections.emptyMap());
+ mtmvPartitionUtilStatic.when(() ->
MTMVPartitionUtil.generatePartitionSnapshots(
+ Mockito.same(refreshContext), Mockito.anySet(),
Mockito.anySet()))
+ .thenReturn(Collections.emptyMap());
+ mtmvPartitionUtilStatic.when(() ->
MTMVPartitionUtil.getMTMVNeedRefreshPartitions(
+ Mockito.same(refreshContext), Mockito.nullable(Set.class)))
+ .thenReturn(Lists.newArrayList(poneName, ptwoName));
+ MTMVTask task = new MTMVTask(mtmv, relation,
MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, null,
+ RefreshMode.INCREMENTAL, false, null));
+
+ UpdateMvByPartitionCommand command =
Mockito.mock(UpdateMvByPartitionCommand.class);
+ try (MockedStatic<MTMVPlanUtil> mtmvPlanUtilStatic =
Mockito.mockStatic(MTMVPlanUtil.class);
+ MockedStatic<UpdateMvByPartitionCommand> updateMvStatic
+ = Mockito.mockStatic(UpdateMvByPartitionCommand.class);
+ MockedConstruction<IvmIncrRefreshManager> delta =
Mockito.mockConstruction(
+ IvmIncrRefreshManager.class, (mock, context) ->
Mockito.when(mock.doRefresh(Mockito.any()))
+ .thenReturn(IvmIncrRefreshResult.success()))) {
+ mtmvPlanUtilStatic.when(() ->
MTMVPlanUtil.createMTMVContext(Mockito.eq(mtmv), Mockito.anyList()))
+ .thenAnswer(invocation -> contextWithQueryId());
+ updateMvStatic.when(() -> UpdateMvByPartitionCommand.from(
+ Mockito.eq(mtmv), Mockito.anySet(), Mockito.anyMap(),
Mockito.any(StatementContext.class)))
+ .thenReturn(command);
+
+ Object request = Deencapsulation.invoke(task,
"resolveRefreshRequest");
+ Object result = Deencapsulation.invoke(task, "executeIvmAttempt",
refreshContext, request,
+ contextWithQueryId(), Lists.newArrayList());
+
+ Assertions.assertEquals("SUCCESS", result.toString());
+ Assertions.assertEquals(1, delta.constructed().size());
+ }
+
+
Assertions.assertEquals(Sets.newHashSet(IvmRefreshMethod.PARTITION_REBUILD,
IvmRefreshMethod.INCREMENTAL),
+ Deencapsulation.getField(task, "committedRefreshKinds"));
+ endTask(task, TaskStatus.SUCCESS);
+ Assertions.assertEquals(IvmRefreshMethod.MIXED,
task.getIvmRefreshMethod());
+ Assertions.assertFalse(task.isIvmFallback());
+ }
+
+ /** An incremental attempt that finds every partition up to date commits
nothing, and reports NONE. */
+ @Test
+ public void testAnIncrementalAttemptWithNothingToCatchUpCommitsNothing()
throws Exception {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ MTMVRefreshContext refreshContext =
Mockito.mock(MTMVRefreshContext.class);
+ mtmvPartitionUtilStatic.when(() ->
MTMVPartitionUtil.getMTMVNeedRefreshPartitions(
+ Mockito.same(refreshContext),
Mockito.nullable(Set.class))).thenReturn(Lists.newArrayList());
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+
+ try (MockedConstruction<IvmIncrRefreshManager> delta =
Mockito.mockConstruction(IvmIncrRefreshManager.class)) {
+ Object request = Deencapsulation.invoke(task,
"resolveRefreshRequest");
+ Object result = Deencapsulation.invoke(task, "executeIvmAttempt",
refreshContext, request,
+ new ConnectContext(), Lists.newArrayList());
+
+ Assertions.assertEquals("SUCCESS", result.toString());
+ Assertions.assertTrue(delta.constructed().isEmpty());
+ }
+
+ endTask(task, TaskStatus.SUCCESS);
+ Assertions.assertEquals(IvmRefreshMethod.NONE,
task.getIvmRefreshMethod());
+ }
+
+ /**
+ * Each batch the partition executor commits notes the kind of work it was
-- a rebuild of the partitions it names,
+ * or of the whole MV -- and a batch that fails notes nothing.
+ */
+ @Test
+ public void testEachCommittedBatchNotesItsKindOfWork() throws Exception {
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+
Mockito.when(mtmv.getDatabase()).thenReturn(Mockito.mock(Database.class));
+ Mockito.when(mtmv.getRefreshPartitionNum()).thenReturn(1);
+
Mockito.when(mtmvPartitionInfo.getPctInfos()).thenReturn(Collections.emptyList());
+ MTMVRefreshContext refreshContext =
Mockito.mock(MTMVRefreshContext.class);
+ UpdateMvByPartitionCommand command =
Mockito.mock(UpdateMvByPartitionCommand.class);
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ try (MockedStatic<MTMVPlanUtil> mtmvPlanUtilStatic =
Mockito.mockStatic(MTMVPlanUtil.class);
+ MockedStatic<UpdateMvByPartitionCommand> updateMvStatic
+ =
Mockito.mockStatic(UpdateMvByPartitionCommand.class)) {
+ mtmvPlanUtilStatic.when(() ->
MTMVPlanUtil.createMTMVContext(Mockito.eq(mtmv), Mockito.anyList()))
+ .thenAnswer(invocation -> contextWithQueryId());
+ updateMvStatic.when(() -> UpdateMvByPartitionCommand.from(
+ Mockito.eq(mtmv), Mockito.anySet(), Mockito.anyMap(),
Mockito.any(StatementContext.class)))
+ .thenReturn(command);
+ AtomicInteger batches = new AtomicInteger();
+ mtmvPlanUtilStatic.when(() ->
MTMVPlanUtil.executeCommand(Mockito.any(ConnectContext.class),
+ Mockito.eq(command), Mockito.any(StatementContext.class),
Mockito.anyString(),
+ Mockito.any(Consumer.class))).thenAnswer(invocation -> {
+ if (batches.incrementAndGet() == 1) {
+ throw new RuntimeException("the first batch
failed");
+ }
+ return null;
+ });
+
+ Assertions.assertThrows(JobException.class, () ->
Deencapsulation.invoke(task,
+ "executePartitionBasedRefresh", refreshContext,
RefreshMode.PARTITIONS, new ConnectContext(),
+ Lists.newArrayList(poneName)));
+ Assertions.assertTrue(((Set<?>) Deencapsulation.getField(task,
"committedRefreshKinds")).isEmpty());
+
+ Deencapsulation.invoke(task, "executePartitionBasedRefresh",
refreshContext, RefreshMode.COMPLETE,
+ new ConnectContext(), Lists.newArrayList(poneName,
ptwoName));
+ Assertions.assertEquals(3, batches.get());
+ Deencapsulation.invoke(task, "executePartitionBasedRefresh",
refreshContext, RefreshMode.PARTITIONS,
+ new ConnectContext(), Lists.newArrayList(ptwoName));
+ }
+
+ Assertions.assertEquals(Sets.newHashSet(IvmRefreshMethod.COMPLETE,
IvmRefreshMethod.PARTITION_REBUILD),
+ Deencapsulation.getField(task, "committedRefreshKinds"));
+ }
+
+ /**
+ * A task falls back when the incremental attempt has given up and an
attempt replacing it commits data. What
+ * committed before it gave up -- the rebuild of invalidated partitions --
does not make it one, and neither does a
+ * fallback that commits nothing.
+ */
+ @Test
+ public void
testOnlyWhatCommitsAfterTheIncrementalAttemptGaveUpIsAFallback() {
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+
+ Deencapsulation.invoke(task, "recordCommitted",
IvmRefreshMethod.PARTITION_REBUILD);
+ Assertions.assertFalse(task.isIvmFallback());
+
+ Deencapsulation.setField(task, "fallingBack", true);
+ Assertions.assertFalse(task.isIvmFallback());
+ Deencapsulation.invoke(task, "recordCommitted",
IvmRefreshMethod.COMPLETE);
+ Assertions.assertTrue(task.isIvmFallback());
+
Assertions.assertEquals(Sets.newHashSet(IvmRefreshMethod.PARTITION_REBUILD,
IvmRefreshMethod.COMPLETE),
+ Deencapsulation.getField(task, "committedRefreshKinds"));
+ }
+
+ /**
+ * Giving up the incremental attempt is a fallback, except where the MV
has no baseline yet: an MV no refresh has
+ * succeeded on builds its first one, which is what
INCOMPLETE_REFRESH_SNAPSHOT says there. Once a refresh has
+ * succeeded, the same reason means the baseline was lost, and rebuilding
it is a fallback.
+ */
+ @Test
+ public void testBuildingTheFirstBaselineIsNotAFallback() {
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+
+ Deencapsulation.setField(task, "ivmFallbackReason",
IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name());
+ Assertions.assertTrue((boolean) Deencapsulation.invoke(task,
"givingUpTheIncrementalAttemptIsAFallback"));
+
+ Deencapsulation.setField(task, "ivmFallbackReason",
MTMVTask.INCOMPLETE_REFRESH_SNAPSHOT);
+ Mockito.when(mtmv.hasSucceededIvmRefresh()).thenReturn(false);
+ Assertions.assertFalse((boolean) Deencapsulation.invoke(task,
"givingUpTheIncrementalAttemptIsAFallback"));
+ Mockito.when(mtmv.hasSucceededIvmRefresh()).thenReturn(true);
+ Assertions.assertTrue((boolean) Deencapsulation.invoke(task,
"givingUpTheIncrementalAttemptIsAFallback"));
+ }
+
+ /**
+ * A failure that no IvmFailureReason names reports the stage it happened
in: an MV whose query no longer analyzes,
+ * one whose partitions cannot be aligned with its base tables', and a
refresh statement that fails.
+ */
+ @Test
+ public void testAFailureReportsTheStageItHappenedIn() throws Exception {
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ Mockito.when(mtmv.getRefreshPartitionNum()).thenReturn(1);
+
Mockito.when(mtmvPartitionInfo.getPctInfos()).thenReturn(Collections.emptyList());
+ UpdateMvByPartitionCommand command =
Mockito.mock(UpdateMvByPartitionCommand.class);
+ try (MockedStatic<MTMVPlanUtil> mtmvPlanUtilStatic =
Mockito.mockStatic(MTMVPlanUtil.class);
+ MockedStatic<MTMVRefreshContext> refreshContextStatic =
Mockito.mockStatic(MTMVRefreshContext.class);
+ MockedStatic<UpdateMvByPartitionCommand> updateMvStatic
+ =
Mockito.mockStatic(UpdateMvByPartitionCommand.class)) {
+ mtmvPlanUtilStatic.when(() ->
MTMVPlanUtil.createMTMVContext(Mockito.eq(mtmv), Mockito.anyList()))
+ .thenAnswer(invocation -> contextWithQueryId());
+ mtmvPlanUtilStatic.when(() ->
MTMVPlanUtil.generateMTMVRelation(Mockito.anySet(), Mockito.anySet()))
+ .thenReturn(relation);
+ refreshContextStatic.when(() -> MTMVRefreshContext.buildContext(
+ Mockito.eq(mtmv), Mockito.anyMap(), Mockito.anyMap()))
+ .thenReturn(Mockito.mock(MTMVRefreshContext.class));
+ updateMvStatic.when(() -> UpdateMvByPartitionCommand.from(
+ Mockito.eq(mtmv), Mockito.anySet(), Mockito.anyMap(),
Mockito.any(StatementContext.class)))
+ .thenReturn(command);
+ mtmvPlanUtilStatic.when(() ->
MTMVPlanUtil.executeCommand(Mockito.any(ConnectContext.class),
+ Mockito.eq(command), Mockito.any(StatementContext.class),
Mockito.anyString(),
+ Mockito.any(Consumer.class))).thenThrow(new
RuntimeException("the refresh statement failed"));
+
+ mtmvPlanUtilStatic.when(() ->
MTMVPlanUtil.getBaseTableFromQuery(Mockito.any(), Mockito.any()))
+ .thenThrow(new
org.apache.doris.nereids.exceptions.AnalysisException("Unknown table"));
+
Assertions.assertEquals(IvmRefreshFailureStage.QUERY_UNUSABLE.name(),
failureStageOfARunThatFails());
+
+ mtmvPlanUtilStatic.when(() ->
MTMVPlanUtil.getBaseTableFromQuery(Mockito.any(), Mockito.any()))
+ .thenReturn(new
MTMVPlanUtil.QueryAnalysisResult(Sets.newHashSet(), Sets.newHashSet(), false));
+ mtmvPartitionUtilStatic.when(() ->
MTMVPartitionUtil.alignMvPartition(Mockito.eq(mtmv), Mockito.anyMap()))
+ .thenThrow(new AnalysisException("the partition column
changed"));
+
Assertions.assertEquals(IvmRefreshFailureStage.PARTITION_SYNC_FAILED.name(),
failureStageOfARunThatFails());
+
+ mtmvPartitionUtilStatic.when(() ->
MTMVPartitionUtil.alignMvPartition(Mockito.eq(mtmv), Mockito.anyMap()))
+ .thenReturn(null);
+
Assertions.assertEquals(IvmRefreshFailureStage.EXECUTION_FAILED.name(),
failureStageOfARunThatFails());
+ }
+ }
+
+ /**
+ * How an IVM MV was refreshed, and why a failed task failed, are decided
as the task ends, from what it committed
+ * and the stage it failed in. A task of an MV that is not an IVM MV
reports neither.
+ */
+ @Test
+ public void testTheRefreshReportIsDecidedAsTheTaskEnds() {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ MTMVTask succeeded = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ Assertions.assertNull(succeeded.getIvmRefreshMethod());
+ Deencapsulation.invoke(succeeded, "recordCommitted",
IvmRefreshMethod.INCREMENTAL);
+ endTask(succeeded, TaskStatus.SUCCESS);
+ Assertions.assertEquals(IvmRefreshMethod.INCREMENTAL,
succeeded.getIvmRefreshMethod());
+ Assertions.assertNull(succeeded.getIvmFailureReason());
+
+ MTMVTask failed = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ Deencapsulation.setField(failed, "failureStage",
IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name());
+ endTask(failed, TaskStatus.FAILED);
+ Assertions.assertEquals(IvmRefreshMethod.NONE,
failed.getIvmRefreshMethod());
+ Assertions.assertEquals(IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name(),
failed.getIvmFailureReason());
+
+ Mockito.when(mtmv.isIvm()).thenReturn(false);
+ MTMVTask plain = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ Deencapsulation.invoke(plain, "recordCommitted",
IvmRefreshMethod.COMPLETE);
+ endTask(plain, TaskStatus.FAILED);
+ Assertions.assertNull(plain.getIvmRefreshMethod());
+ Assertions.assertNull(plain.getIvmFailureReason());
+ }
+
+ @Test
+ public void testTaskInfoContainsIvmRefreshMethodColumn() {
+ Mockito.when(mtmv.getQualifiedDbName()).thenReturn("test_db");
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ int columnIndex = MTMVTask.COLUMN_TO_INDEX.get("ivmrefreshmethod");
+
+ List<TCell> cells = task.getTvfInfo("job").getColumnValue();
+ Assertions.assertEquals(MTMVTask.SCHEMA.size(), cells.size());
+ // a task that has not ended, or of an MV that is not an IVM MV
+ Assertions.assertEquals(FeConstants.null_string,
cells.get(columnIndex).getStringVal());
+
+ Deencapsulation.setField(task, "ivmRefreshMethod",
IvmRefreshMethod.PARTITION_REBUILD);
+ cells = task.getTvfInfo("job").getColumnValue();
+ Assertions.assertEquals("PARTITION_REBUILD",
cells.get(columnIndex).getStringVal());
+ }
+
+ /** What the refresh report is decided from stays out of the journal; the
report itself goes into it. */
+ @Test
+ public void testTheRefreshReportIsJournaledWithTheTask() {
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ Deencapsulation.setField(task, "ivmRefreshMethod",
IvmRefreshMethod.COMPLETE);
+ Deencapsulation.setField(task, "ivmFailureReason",
IvmRefreshFailureStage.EXECUTION_FAILED.name());
+ Deencapsulation.setField(task, "ivmFallback", true);
+ Deencapsulation.setField(task, "fallingBack", true);
+ Deencapsulation.invoke(task, "recordCommitted",
IvmRefreshMethod.INCREMENTAL);
+
+ String json = GsonUtils.GSON.toJson(task);
+ Assertions.assertFalse(json.contains("committedRefreshKinds"), json);
+ Assertions.assertFalse(json.contains("fallingBack"), json);
+ Assertions.assertFalse(json.contains("failureStage"), json);
+ MTMVTask replayed = GsonUtils.GSON.fromJson(json, MTMVTask.class);
+ Assertions.assertEquals(IvmRefreshMethod.COMPLETE,
replayed.getIvmRefreshMethod());
+
Assertions.assertEquals(IvmRefreshFailureStage.EXECUTION_FAILED.name(),
replayed.getIvmFailureReason());
+ Assertions.assertTrue(replayed.isIvmFallback());
+ }
+
+ /** Runs a task that fails and returns the failure stage it reached. */
+ private String failureStageOfARunThatFails() {
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ task.setStatus(TaskStatus.RUNNING);
+ Assertions.assertThrows(JobException.class, task::run);
+ return Deencapsulation.getField(task, "failureStage");
+ }
+
+ /** Ends the task as {@code status} and has it publish its result, the way
onSuccess, onFail and a STOP do. */
+ private void endTask(MTMVTask task, TaskStatus status) {
+ Mockito.when(mtmv.getQualifiedDbName()).thenReturn("test_db");
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ task.setStatus(status);
+ task.setStartTimeMs(1000L);
+ task.setFinishTimeMs(3500L);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+
mockedEnv.when(Env::getCurrentEnv).thenReturn(Mockito.mock(Env.class));
+ Deencapsulation.invoke(task, "after");
+ }
+ }
+
+ private static ConnectContext contextWithQueryId() {
+ ConnectContext context = new ConnectContext();
+ context.setQueryId(new TUniqueId(1L, 2L));
+ return context;
+ }
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
index 62dd652f68f..3a99d13af1c 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
@@ -44,6 +44,10 @@ import
org.apache.doris.mtmv.MTMVRefreshEnum.MTMVRefreshState;
import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVState;
import org.apache.doris.mtmv.MTMVRefreshEnum.RefreshMethod;
import org.apache.doris.mtmv.MTMVRefreshEnum.RefreshTrigger;
+import org.apache.doris.mtmv.ivm.IvmFailureReason;
+import org.apache.doris.mtmv.ivm.IvmRefreshFailureStage;
+import org.apache.doris.mtmv.ivm.IvmRefreshMethod;
+import org.apache.doris.mtmv.ivm.IvmRefreshSummary;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.persist.AlterMTMV;
import org.apache.doris.persist.EditLog;
@@ -490,6 +494,8 @@ public class MTMVTest {
MTMVRelation relation = mtmv.getRelation();
MTMVTask task = new MTMVTask(mtmv, relation, null);
task.setStatus(TaskStatus.FAILED);
+ task.setStartTimeMs(1000L);
+ task.setFinishTimeMs(2000L);
AlterMTMV alterMTMV = new AlterMTMV(new TableNameInfo("db1", "mv1"),
MTMVAlterOpType.ADD_TASK);
alterMTMV.setTask(task);
alterMTMV.setRelation(relation);
@@ -505,6 +511,88 @@ public class MTMVTest {
Mockito.verify(editLogItem).await();
}
+ /**
+ * The refresh summary of an IVM MV follows each task result the MV
records, and a follower replaying the same
+ * journal records ends with the same summary -- however few tasks the
history keeps.
+ */
+ @Test
+ public void testTheRefreshSummaryIsTheSameOnTheMasterAndInTheReplay() {
+ MTMV master = buildSerializableMTMV();
+ master.getIvmInfo().setEnableIvm(true);
+ MTMV follower = buildSerializableMTMV();
+ follower.getIvmInfo().setEnableIvm(true);
+ List<AlterMTMV> journaled = Lists.newArrayList();
+ int maxPersistenceTaskCount = Config.max_persistence_task_count;
+ try {
+ Config.max_persistence_task_count = 1;
+ withMockedEditLog(journaled, () -> {
+ for (MTMVTask task : Lists.newArrayList(
+ endedTask(master, TaskStatus.SUCCESS,
IvmRefreshMethod.COMPLETE, null, null),
+ endedTask(master, TaskStatus.FAILED,
IvmRefreshMethod.NONE,
+ IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name(),
null),
+ endedTask(master, TaskStatus.CANCELED,
IvmRefreshMethod.PARTITION_REBUILD, null, null),
+ endedTask(master, TaskStatus.FAILED,
IvmRefreshMethod.NONE,
+
IvmRefreshFailureStage.EXECUTION_FAILED.name(), null),
+ endedTask(master, TaskStatus.SUCCESS,
IvmRefreshMethod.MIXED, null,
+ IvmFailureReason.CHANGELOG_BROKEN.name()))) {
+
Assertions.assertTrue(master.addTaskResult(taskResult(master, task), false));
+ }
+ for (AlterMTMV record : journaled) {
+ AlterMTMV replayed =
GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(record), AlterMTMV.class);
+ Assertions.assertTrue(follower.addTaskResult(replayed,
true));
+ }
+ });
+ } finally {
+ Config.max_persistence_task_count = maxPersistenceTaskCount;
+ }
+
+ Assertions.assertEquals(5, journaled.size());
+ IvmRefreshSummary summary = Deencapsulation.getField(master,
"ivmRefreshSummary");
+ Assertions.assertEquals(0, summary.getConsecutiveFailures());
+
Assertions.assertEquals(IvmRefreshFailureStage.EXECUTION_FAILED.name(),
summary.getLastFailureReason());
+ Assertions.assertEquals(IvmRefreshMethod.MIXED,
summary.getLastRefreshMethod());
+ Assertions.assertEquals(1L, summary.getFallbackCount());
+ Assertions.assertEquals(IvmFailureReason.CHANGELOG_BROKEN.name(),
summary.getLastFallbackReason());
+ Assertions.assertTrue(master.hasSucceededIvmRefresh());
+ IvmRefreshSummary replayed = Deencapsulation.getField(follower,
"ivmRefreshSummary");
+ Assertions.assertEquals(GsonUtils.GSON.toJson(summary),
GsonUtils.GSON.toJson(replayed));
+ }
+
+ /** An MV that is not an IVM MV records its task results without a refresh
summary. */
+ @Test
+ public void testAPlainMvKeepsNoRefreshSummary() {
+ MTMV mtmv = buildSerializableMTMV();
+
+ withMockedEditLog(Lists.newArrayList(), () ->
Assertions.assertTrue(mtmv.addTaskResult(taskResult(mtmv,
+ endedTask(mtmv, TaskStatus.FAILED, null,
IvmRefreshFailureStage.OTHER.name(), null)), false)));
+
+ IvmRefreshSummary summary = Deencapsulation.getField(mtmv,
"ivmRefreshSummary");
+ Assertions.assertEquals(0, summary.getConsecutiveFailures());
+ Assertions.assertEquals(0L, summary.getLastFailureTimeMs());
+ }
+
+ /** A task of the MV that has ended as {@code status}, with what it
reports about its refresh. */
+ private static MTMVTask endedTask(MTMV mtmv, TaskStatus status,
IvmRefreshMethod method, String failureReason,
+ String fallbackReason) {
+ MTMVTask task = new MTMVTask(mtmv, mtmv.getRelation(), null);
+ task.setStatus(status);
+ task.setStartTimeMs(1000L);
+ task.setFinishTimeMs(2000L);
+ Deencapsulation.setField(task, "ivmRefreshMethod", method);
+ Deencapsulation.setField(task, "ivmFailureReason", failureReason);
+ Deencapsulation.setField(task, "ivmFallback", fallbackReason != null);
+ Deencapsulation.setField(task, "ivmFallbackReason", fallbackReason);
+ return task;
+ }
+
+ private static AlterMTMV taskResult(MTMV mtmv, MTMVTask task) {
+ AlterMTMV alterMTMV = new AlterMTMV(new TableNameInfo("db1", "mv1"),
MTMVAlterOpType.ADD_TASK);
+ alterMTMV.setTask(task);
+ alterMTMV.setRelation(mtmv.getRelation());
+ alterMTMV.setPartitionSnapshots(Map.of());
+ return alterMTMV;
+ }
+
@Test
public void testRefreshPublishAdvancesCacheGeneration() {
MTMVCacheManager manager = new MTMVCacheManager();
@@ -1384,6 +1472,9 @@ public class MTMVTest {
withMockedEditLog(journaled, () -> {
MTMVTask task = new MTMVTask(mtmv, mtmv.getRelation(), null);
task.setStatus(TaskStatus.FAILED);
+ // A task that has failed has run, and has the times to say so.
+ task.setStartTimeMs(1000L);
+ task.setFinishTimeMs(2000L);
// Set on the task rather than through its getter: the getter
hands out a detached copy, which
// is the point of it -- a cancelled task's worker may still be
merging into the field.
Map<String, Long> taskCapturedEpochs = Maps.newConcurrentMap();
@@ -1435,6 +1526,8 @@ public class MTMVTest {
EditLogItem editLogItem = Mockito.mock(EditLogItem.class);
Mockito.when(env.getEditLog()).thenReturn(editLog);
Mockito.when(env.getMtmvService()).thenReturn(Mockito.mock(MTMVService.class));
+ // a successful task result invalidates the rewrite cache
+ Mockito.when(env.getMtmvCacheManager()).thenReturn(new
MTMVCacheManager());
Mockito.when(editLog.submitEdit(Mockito.eq(OperationType.OP_ALTER_MTMV),
Mockito.any(AlterMTMV.class)))
.thenAnswer(invocation -> {
journaled.add(invocation.getArgument(1));
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
index 9cd74995cf1..0ce6ee5b229 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
@@ -1546,6 +1546,9 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
private AlterMTMV taskResult(MTMV mtmv, TaskStatus status, long
schemaChangeVersion) {
MTMVTask task = new MTMVTask();
task.setStatus(status);
+ // A task that has ended has run, and has the times to say so.
+ task.setStartTimeMs(1000L);
+ task.setFinishTimeMs(2000L);
Deencapsulation.setField(task, "mtmvSchemaChangeVersion",
schemaChangeVersion);
AlterMTMV result = new AlterMTMV(
new TableNameInfo(mtmv.getQualifiedDbName(), mtmv.getName()),
MTMVAlterOpType.ADD_TASK);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmRefreshSummaryTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmRefreshSummaryTest.java
new file mode 100644
index 00000000000..6e06f666121
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmRefreshSummaryTest.java
@@ -0,0 +1,163 @@
+// 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.mtmv.ivm;
+
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.job.common.TaskStatus;
+import org.apache.doris.job.extensions.mtmv.MTMVTask;
+
+import com.google.common.base.Strings;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class IvmRefreshSummaryTest {
+
+ /**
+ * Consecutive failures are the failed tasks since the latest successful
one: a success clears them, including one
+ * that had nothing to commit, and a canceled task leaves them as they are.
+ */
+ @Test
+ public void testConsecutiveFailuresAreTheFailuresSinceTheLatestSuccess() {
+ IvmRefreshSummary summary = new IvmRefreshSummary();
+
+ summary.update(failed(1000L, 2000L,
IvmRefreshFailureStage.EXECUTION_FAILED.name(), "first"));
+ summary.update(failed(3000L, 4000L,
IvmRefreshFailureStage.EXECUTION_FAILED.name(), "second"));
+ Assertions.assertEquals(2, summary.getConsecutiveFailures());
+ summary.update(canceled());
+ Assertions.assertEquals(2, summary.getConsecutiveFailures());
+
+ summary.update(succeeded(5000L, 6000L, IvmRefreshMethod.NONE));
+ Assertions.assertEquals(0, summary.getConsecutiveFailures());
+ summary.update(failed(7000L, 8000L,
IvmRefreshFailureStage.OTHER.name(), "third"));
+ Assertions.assertEquals(1, summary.getConsecutiveFailures());
+ }
+
+ /**
+ * The latest failure keeps its time, reason and message, and the message
is cut to a bounded length; the latest
+ * success does not clear them. How the latest task refreshed the MV, and
how long it took, follow every task that
+ * succeeded or failed.
+ */
+ @Test
+ public void testTheLatestSuccessAndFailureAreKept() {
+ IvmRefreshSummary summary = new IvmRefreshSummary();
+
+ String longMessage = Strings.repeat("x",
IvmRefreshSummary.MAX_ERROR_MESSAGE_LENGTH * 2);
+ summary.update(failed(1000L, 1500L,
IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name(), longMessage));
+ Assertions.assertEquals(1500L, summary.getLastFailureTimeMs());
+ Assertions.assertEquals(IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name(),
summary.getLastFailureReason());
+ Assertions.assertEquals(IvmRefreshSummary.MAX_ERROR_MESSAGE_LENGTH,
summary.getLastErrorMessage().length());
+
Assertions.assertTrue(longMessage.startsWith(summary.getLastErrorMessage().substring(0,
100)));
+ Assertions.assertEquals(IvmRefreshMethod.NONE,
summary.getLastRefreshMethod());
+ Assertions.assertEquals(500L, summary.getLastRefreshDurationMs());
+ Assertions.assertEquals(0L, summary.getLastSuccessTimeMs());
+
+ summary.update(succeeded(2000L, 4500L, IvmRefreshMethod.COMPLETE));
+ Assertions.assertEquals(4500L, summary.getLastSuccessTimeMs());
+ Assertions.assertEquals(IvmRefreshMethod.COMPLETE,
summary.getLastRefreshMethod());
+ Assertions.assertEquals(2500L, summary.getLastRefreshDurationMs());
+ Assertions.assertEquals(1500L, summary.getLastFailureTimeMs());
+ Assertions.assertEquals(IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name(),
summary.getLastFailureReason());
+
+ // A canceled task has no finish time, so it changes neither the
latest method nor the latest duration.
+ summary.update(canceled());
+ Assertions.assertEquals(IvmRefreshMethod.COMPLETE,
summary.getLastRefreshMethod());
+ Assertions.assertEquals(2500L, summary.getLastRefreshDurationMs());
+ }
+
+ /** Fallbacks count the tasks that succeeded or failed after falling back;
a canceled one is not counted. */
+ @Test
+ public void testFallbacksCountTheTasksThatFellBack() {
+ IvmRefreshSummary summary = new IvmRefreshSummary();
+
+ summary.update(succeeded(1000L, 2000L, IvmRefreshMethod.INCREMENTAL));
+ Assertions.assertEquals(0L, summary.getFallbackCount());
+
+ MTMVTask fellBack = succeeded(3000L, 4000L, IvmRefreshMethod.COMPLETE);
+ fellBack(fellBack, IvmFailureReason.MIN_MAX_BOUNDARY_HIT.name());
+ summary.update(fellBack);
+ MTMVTask fellBackAndFailed = failed(5000L, 6000L,
IvmRefreshFailureStage.EXECUTION_FAILED.name(), "failed");
+ fellBack(fellBackAndFailed, IvmFailureReason.CHANGELOG_BROKEN.name());
+ summary.update(fellBackAndFailed);
+ MTMVTask fellBackAndCanceled = canceled();
+ fellBack(fellBackAndCanceled,
IvmFailureReason.PLAN_SIGNATURE_MISMATCH.name());
+ summary.update(fellBackAndCanceled);
+
+ Assertions.assertEquals(2L, summary.getFallbackCount());
+ Assertions.assertEquals(IvmFailureReason.CHANGELOG_BROKEN.name(),
summary.getLastFallbackReason());
+ Assertions.assertEquals(6000L, summary.getLastFallbackTimeMs());
+ }
+
+ /** Only a task that succeeded gives the MV the baseline the incremental
refresh keeps up. */
+ @Test
+ public void testOnlyASuccessMarksTheMvRefreshed() {
+ IvmRefreshSummary summary = new IvmRefreshSummary();
+
+ summary.update(failed(1000L, 2000L,
IvmRefreshFailureStage.OTHER.name(), "failed"));
+ summary.update(canceled());
+ Assertions.assertFalse(summary.hasSucceeded());
+
+ summary.update(succeeded(3000L, 4000L, IvmRefreshMethod.COMPLETE));
+ summary.update(failed(5000L, 6000L,
IvmRefreshFailureStage.OTHER.name(), "failed"));
+ Assertions.assertTrue(summary.hasSucceeded());
+ }
+
+ /** A result is a task that has ended. */
+ @Test
+ public void testATaskThatHasNotEndedIsNotAResult() {
+ MTMVTask running = new MTMVTask();
+ running.setStatus(TaskStatus.RUNNING);
+
+ Assertions.assertThrows(IllegalStateException.class, () -> new
IvmRefreshSummary().update(running));
+ }
+
+ private static MTMVTask succeeded(long startTimeMs, long finishTimeMs,
IvmRefreshMethod method) {
+ MTMVTask task = ended(TaskStatus.SUCCESS, startTimeMs, finishTimeMs);
+ Deencapsulation.setField(task, "ivmRefreshMethod", method);
+ return task;
+ }
+
+ private static MTMVTask failed(long startTimeMs, long finishTimeMs, String
reason, String message) {
+ MTMVTask task = ended(TaskStatus.FAILED, startTimeMs, finishTimeMs);
+ Deencapsulation.setField(task, "ivmRefreshMethod",
IvmRefreshMethod.NONE);
+ Deencapsulation.setField(task, "ivmFailureReason", reason);
+ task.setErrMsg(message);
+ return task;
+ }
+
+ // A STOP sets neither a finish time nor a failure reason.
+ private static MTMVTask canceled() {
+ MTMVTask task = new MTMVTask();
+ task.setStatus(TaskStatus.CANCELED);
+ task.setStartTimeMs(1L);
+ Deencapsulation.setField(task, "ivmRefreshMethod",
IvmRefreshMethod.NONE);
+ return task;
+ }
+
+ private static MTMVTask ended(TaskStatus status, long startTimeMs, long
finishTimeMs) {
+ MTMVTask task = new MTMVTask();
+ task.setStatus(status);
+ task.setStartTimeMs(startTimeMs);
+ task.setFinishTimeMs(finishTimeMs);
+ return task;
+ }
+
+ private static void fellBack(MTMVTask task, String reason) {
+ Deencapsulation.setField(task, "ivmFallback", true);
+ Deencapsulation.setField(task, "ivmFallbackReason", reason);
+ }
+}
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_method_metrics.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_method_metrics.out
new file mode 100644
index 00000000000..b20ae419b55
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_method_metrics.out
@@ -0,0 +1,40 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !first_baseline_task --
+SUCCESS COMPLETE NONE
+
+-- !incremental_task --
+SUCCESS INCREMENTAL NONE
+
+-- !no_change_task --
+SUCCESS NONE NONE
+
+-- !strict_failure_task --
+FAILED NONE MIN_MAX_BOUNDARY_HIT
+
+-- !fallback_task --
+SUCCESS COMPLETE MIN_MAX_BOUNDARY_HIT
+
+-- !mv_after_fallback --
+2026-01-10 20 20 1
+2026-02-10 30 30 1
+2026-03-10 40 50 2
+
+-- !base_after_fallback --
+2026-01-10 20 20 1
+2026-02-10 30 30 1
+2026-03-10 40 50 2
+
+-- !complete_task --
+SUCCESS COMPLETE NONE
+
+-- !mixed_task --
+SUCCESS MIXED NONE
+
+-- !mv_after_mixed --
+2026-02-10 30 60 2
+2026-03-10 40 50 2
+
+-- !base_after_mixed --
+2026-02-10 30 60 2
+2026-03-10 40 50 2
+
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_method_metrics.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_method_metrics.groovy
new file mode 100644
index 00000000000..21f52437a05
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_method_metrics.groovy
@@ -0,0 +1,233 @@
+// 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.
+
+import org.apache.doris.regression.util.Http
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+/**
+ * How a refresh task of an IVM MV says the way it refreshed the MV, and how
the Master FE counts these tasks.
+ *
+ * <p>tasks("type"="mv") reports IvmRefreshMethod, the kinds of work that
committed data: INCREMENTAL, PARTITION_REBUILD,
+ * COMPLETE, MIXED when more than one of them did, NONE when nothing did. The
Master counts each task as it ends in
+ * doris_fe_ivm_refresh_tasks_total{result,method}, adds how long it ran to
doris_fe_ivm_refresh_duration_seconds_total,
+ * counts a task that fell back from the incremental refresh -- and committed
data in the refresh that replaced it -- in
+ * doris_fe_ivm_fallbacks_total{reason}, and a failed one in
doris_fe_ivm_refresh_failures_total{reason}.
+ *
+ * <p>Cases, in order, on one MV:
+ * <ol>
+ * <li>the first refresh finds every partition still to be filled and builds
the MV with COMPLETE, which is not a
+ * fallback;</li>
+ * <li>an incremental refresh of new rows is INCREMENTAL, and one with
nothing new commits nothing (no_change);</li>
+ * <li>a strict INCREMENTAL refresh that meets a MIN/MAX boundary fails for
that reason and commits nothing;</li>
+ * <li>the same refresh allowed to fall back rebuilds the MV, which is a
fallback for that reason;</li>
+ * <li>an explicit COMPLETE refresh is COMPLETE, and not a fallback;</li>
+ * <li>a truncated base partition makes its MV partition need a rebuild,
which an incremental refresh does before the
+ * delta: MIXED, and not a fallback either.</li>
+ * </ol>
+ *
+ * <p>nonConcurrent: the counters are the Master's, for every IVM MV, so the
exact moves asserted here hold only while no
+ * other suite refreshes one.
+ */
+suite("test_ivm_refresh_method_metrics", "nonConcurrent") {
+ // The MV is checked against its base table below, so a query must not be
answered from it; and a base-table write
+ // must not be redirected into it, or the incremental path under test goes
unused.
+ sql "SET enable_materialized_view_rewrite = false"
+ sql "SET enable_dml_materialized_view_rewrite = false"
+
+ String mvName = "ivm_refresh_method_metrics_mv"
+ String masterMetricsUrl =
"http://${getMasterIp()}:${getMasterPort()}/metrics"
+
+ // The value of every doris_fe_ivm_ series the Master exports, by the
series.
+ def ivmSeries = {
+ Map<String, Double> values = [:]
+ for (String line : Http.GET(masterMetricsUrl, false,
false).readLines()) {
+ if (line.startsWith("doris_fe_ivm_")) {
+ int split = line.lastIndexOf(' ')
+ values.put(line.substring(0, split),
Double.parseDouble(line.substring(split + 1)))
+ }
+ }
+ return values
+ }
+
+ def latestTaskId = {
+ def tasks = sql """
+ SELECT TaskId FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}' AND MvName = '${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """
+ return tasks.isEmpty() ? null : tasks[0][0].toString()
+ }
+
+ def waitForNewTask = { previousTaskId ->
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}' AND MvName =
'${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != previousTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ return taskResult[0].TaskId.toString()
+ }
+
+ // Runs a refresh and waits for its task to end. Returns the task, the
counters that moved meanwhile and by how much,
+ // and the run times added, which vary from run to run and are returned
apart, by method.
+ def refresh = { String statement ->
+ String previousTaskId = latestTaskId()
+ Map<String, Double> before = ivmSeries()
+ sql statement
+ String taskId = waitForNewTask(previousTaskId)
+ Map<String, Double> after = ivmSeries()
+ Map<String, Double> counted = [:]
+ Map<String, Double> seconds = [:]
+ String durationPrefix =
'doris_fe_ivm_refresh_duration_seconds_total{method="'
+ after.each { series, value ->
+ double moved = value - before[series]
+ if (moved == 0) {
+ return
+ }
+ if (series.startsWith(durationPrefix)) {
+ seconds.put(series.substring(durationPrefix.length(),
series.length() - 2), moved)
+ } else {
+ counted.put(series, moved)
+ }
+ }
+ return [taskId, counted, seconds]
+ }
+
+ // How the task ended and refreshed the MV, in one row. An unset column
comes back as the literal two-character
+ // string "\N", which does not survive the .out round trip, so it is
folded into a printable token: a method or a
+ // reason is an enum name, which "\N" is not.
+ def taskQuery = { String taskId ->
+ """
+ SELECT Status,
+ CASE WHEN IvmRefreshMethod REGEXP '^[A-Z_]+\$' THEN
IvmRefreshMethod ELSE 'UNSET' END,
+ CASE WHEN IvmFallbackReason REGEXP '^[A-Z_]+\$' THEN
IvmFallbackReason ELSE 'NONE' END
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """
+ }
+
+ def tasks = { String result, String method ->
+ "doris_fe_ivm_refresh_tasks_total{result=\"${result}\",
method=\"${method}\"}".toString()
+ }
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ivm_refresh_method_metrics_t"""
+ sql """
+ CREATE TABLE ivm_refresh_method_metrics_t (
+ dt DATE NOT NULL,
+ id BIGINT NOT NULL,
+ v INT
+ )
+ UNIQUE KEY(dt, id)
+ PARTITION BY RANGE(dt) (
+ PARTITION p1 VALUES [('2026-01-01'), ('2026-02-01')),
+ PARTITION p2 VALUES [('2026-02-01'), ('2026-03-01')),
+ PARTITION p3 VALUES [('2026-03-01'), ('2026-04-01'))
+ )
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "changelog.enable" = "true",
+ "changelog.need_historical_value" = "true"
+ )
+ """
+ sql """INSERT INTO ivm_refresh_method_metrics_t VALUES
+ ('2026-01-10', 1, 10), ('2026-01-10', 2, 20),
+ ('2026-02-10', 3, 30),
+ ('2026-03-10', 4, 40)"""
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL FALLBACK ON MANUAL
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(dt) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT dt, MIN(v) AS min_v, MAX(v) AS max_v, COUNT(*) AS cnt
+ FROM ivm_refresh_method_metrics_t
+ GROUP BY dt
+ """
+
+ // Every series is there from 0, before a task has counted in it.
+ Map<String, Double> initial = ivmSeries()
+ assertTrue(initial.containsKey(tasks("canceled", "mixed")),
initial.toString())
+
assertTrue(initial.containsKey('doris_fe_ivm_fallbacks_total{reason="incomplete_refresh_snapshot"}'))
+
assertTrue(initial.containsKey('doris_fe_ivm_refresh_failures_total{reason="partition_sync_failed"}'))
+ assertTrue(initial.containsKey('doris_fe_ivm_schedule_rejected_total'))
+
+ // 1. No partition of a new MV has been filled, so the first refresh goes
straight to COMPLETE: it builds the first
+ // baseline, and the incremental refresh never gave up on anything.
+ def (baselineTask, baselineCounted, baselineSeconds) = refresh("REFRESH
MATERIALIZED VIEW ${mvName} AUTO")
+ qt_first_baseline_task taskQuery(baselineTask)
+ assertEquals([(tasks("success", "complete")): 1.0d], baselineCounted)
+ assertEquals(["complete"], baselineSeconds.keySet().toList())
+
+ // 2. New rows are caught up incrementally, and a refresh with nothing new
to catch up commits nothing.
+ sql """INSERT INTO ivm_refresh_method_metrics_t VALUES ('2026-03-10', 5,
50)"""
+ def (deltaTask, deltaCounted, deltaSeconds) = refresh("REFRESH
MATERIALIZED VIEW ${mvName} INCREMENTAL")
+ qt_incremental_task taskQuery(deltaTask)
+ assertEquals([(tasks("success", "incremental")): 1.0d], deltaCounted)
+ assertEquals(["incremental"], deltaSeconds.keySet().toList())
+
+ def (idleTask, idleCounted, idleSeconds) = refresh("REFRESH MATERIALIZED
VIEW ${mvName} INCREMENTAL")
+ qt_no_change_task taskQuery(idleTask)
+ assertEquals([(tasks("no_change", "none")): 1.0d], idleCounted)
+
+ // 3. Deleting the row that holds a group's minimum leaves the delta
unable to say what the new minimum is. A strict
+ // INCREMENTAL request fails for that reason, and has committed nothing.
+ sql """DELETE FROM ivm_refresh_method_metrics_t WHERE dt = '2026-01-10'
AND id = 1"""
+ def (strictTask, strictCounted, strictSeconds) = refresh("REFRESH
MATERIALIZED VIEW ${mvName} INCREMENTAL")
+ qt_strict_failure_task taskQuery(strictTask)
+ assertEquals([(tasks("failed", "none")): 1.0d,
+
'doris_fe_ivm_refresh_failures_total{reason="min_max_boundary_hit"}': 1.0d],
strictCounted)
+
+ // 4. Allowed to fall back, the same refresh rebuilds the MV: a fallback,
for the reason the delta gave up.
+ def (fallbackTask, fallbackCounted, fallbackSeconds) =
+ refresh("REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL FALLBACK")
+ qt_fallback_task taskQuery(fallbackTask)
+ assertEquals([(tasks("success", "complete")): 1.0d,
+
'doris_fe_ivm_fallbacks_total{reason="min_max_boundary_hit"}': 1.0d],
fallbackCounted)
+ order_qt_mv_after_fallback """SELECT dt, min_v, max_v, cnt FROM
${mvName}"""
+ order_qt_base_after_fallback """
+ SELECT dt, MIN(v), MAX(v), COUNT(*) FROM ivm_refresh_method_metrics_t
GROUP BY dt
+ """
+
+ // 5. A COMPLETE request rebuilds the MV because it was asked to, which is
not a fallback.
+ def (completeTask, completeCounted, completeSeconds) = refresh("REFRESH
MATERIALIZED VIEW ${mvName} COMPLETE")
+ qt_complete_task taskQuery(completeTask)
+ assertEquals([(tasks("success", "complete")): 1.0d], completeCounted)
+
+ // 6. Truncating a base partition leaves its MV partition needing a
rebuild, which no delta can make: an incremental
+ // refresh rebuilds it first and then catches up the other partitions,
refreshing the MV in two ways at once.
+ sql """TRUNCATE TABLE ivm_refresh_method_metrics_t PARTITION(p1)"""
+ sql """INSERT INTO ivm_refresh_method_metrics_t VALUES ('2026-02-10', 6,
60)"""
+ def (mixedTask, mixedCounted, mixedSeconds) = refresh("REFRESH
MATERIALIZED VIEW ${mvName} INCREMENTAL")
+ qt_mixed_task taskQuery(mixedTask)
+ assertEquals([(tasks("success", "mixed")): 1.0d], mixedCounted)
+ assertEquals(["mixed"], mixedSeconds.keySet().toList())
+ order_qt_mv_after_mixed """SELECT dt, min_v, max_v, cnt FROM ${mvName}"""
+ order_qt_base_after_mixed """
+ SELECT dt, MIN(v), MAX(v), COUNT(*) FROM ivm_refresh_method_metrics_t
GROUP BY dt
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
