morningman opened a new issue, #69907:
URL: https://github.com/apache/doris/issues/69907
This issue tracks monitoring for the consumers of a table's changelog: table
streams, and IVM materialized views (MVs). It adds Prometheus metrics on the FE
and the BE, two system tables and new columns on existing ones, and fixes the
correctness problems found while designing them. Each sub task is one PR to
master, labeled `incremental-computation`.
Some context. A table with `changelog.enable` keeps its changelog for
`changelog.ttl_seconds` (one day by default). A table stream keeps a
consumption offset, a TSO, for each partition of its base table: a DML that
consumes the stream reads the changes after the offset and moves the offset
when it commits. An IVM MV stays up to date the same way, through an internal
stream on each base table (except those in `excluded_trigger_tables`), and
falls back to rebuilding partitions or the whole MV when the delta cannot be
applied. Today all of this is visible only through system tables queried one by
one (`table_streams`, `table_stream_consumption`, `mv_infos()`, `jobs()`,
`tasks()`). Nothing in `/metrics` watches it.
## 1. Problems to solve
Five questions users ask, referred to below as S1 to S5.
**S1. Is my MV fresh?** Does a report on an IVM MV show the latest data, how
far behind is it, and why? `mv_infos()` has a sync flag but no delay in
seconds. Users also need to tell apart a refresh that is committed but not
visible yet, an MV partition that needs a rebuild, and a change to a table in
`excluded_trigger_tables`, which by design triggers no refresh.
**S2. Has my MV stopped refreshing?** Are refreshes failing, how many times
in a row and why, is a task stuck running or queued? `tasks()` keeps a bounded
history. The `doris_fe_async_materialized_view_*` metrics mix IVM MVs with the
other async MVs and one MV with another, time only the successful tasks, and do
not count the tasks rejected because the task queue was full.
**S3. Has incremental refresh quietly turned into recomputation?** Every
task succeeds, yet refreshes get slower and the cluster gets busier. The
`RefreshMode` column of `tasks()` tells which partitions a task covered
(`COMPLETE`, `PARTIAL`, `NOT_REFRESH`), not whether it applied a delta or
rebuilt the MV, and nothing counts fallbacks or their reasons.
**S4. Is a stream consumer falling behind, will the TTL catch up with it,
and has it already lost changes?** `table_stream_consumption` shows the offset
of each partition, but its `LAG` is a raw TSO difference (the low 18 bits of a
TSO are a logical counter), neither rows nor milliseconds, and
`CONSUMPTION_STATUS` is the consumed TSO as text, not a health state. Nothing
tells how old the oldest unconsumed change is, or how long until the TTL
expires an offset. A `DETAIL` or `APPEND_ONLY` read from an expired offset
succeeds and skips the expired changes without a trace.
**S5. How much space does the changelog take, and is it cleaned up?** `SHOW
DATA` lists `ChangelogSize` on demand, which no alert or trend can use. On the
BE the TTL cleanup shows only as bvars on brpc's `/vars`, which count what was
found and submitted, not what was cleaned up or how many bytes came back.
All of this has to hold at scale. Each IVM MV brings an internal stream per
base table, so a cluster with thousands of MVs has several times as many
streams. A problem in any of them has to show without the `/metrics` of the
Master FE growing with every object.
Designing the monitoring also turned up six defects that users hit directly.
They are fixed here as well; the IDs are the sub tasks in section 4.
- **T0.** Once the TTL expired the offset of an IVM MV's internal stream,
every refresh of the MV failed, even one allowed to fall back, until someone
ran a manual `COMPLETE` refresh.
- **T10.** An IVM MV over a duplicate-key table silently missed rows once
its offset expired: the refresh succeeded, later refreshes did not bring the
rows back, and a base-table query that transparent rewrite answered from the MV
returned the same wrong result.
- **T11.** An IVM MV silently missed the rows of a base partition its
internal stream had no offset for (one added after the MV was built, or one
empty at the time) once those rows were older than the TTL.
- **T9.** A base partition idle for longer than the TTL makes the refresh
after its next write rebuild the whole MV, although nothing that refresh needs
has expired: a dimension table updated weekly, a backfill into an older
partition, a quiet weekend.
- **T12.** A user stream read as `MIN_DELTA` (the default on a unique-key
table), and a stream on a duplicate-key table, silently skip the rows of a
partition they have no offset for once those rows are older than the TTL, and
the offset moves past them for good. `MIN_DELTA` even emits `UPDATE_BEFORE` and
`DELETE` rows for rows the consumer never received.
- **T8.** A table name containing `"`, `\` or a line feed makes the whole
`/metrics` response of the Master FE unparseable in cloud mode, where four
per-table series carry the table name. The MV and stream names on the new
series would do the same.
## 2. How users will use it
The names below are as designed; until the PR of a part is merged, its
review may still change them. Nothing has to be turned on: with the changelog
feature enabled, the monitor runs on the Master FE by default.
### Alert from Prometheus
Keep scraping `/metrics` of every FE and BE. Only the Master FE exports the
state series, and another FE exports none of them rather than 0, so they need
no filter by node; counters are summed over the FEs after `increase`. Every
alert signal has a cluster-level series that covers all objects, so it fires
however many MVs and streams there are. The first 1,000 MVs and the first 1,000
user streams by ID also get a few series of their own, for per-object alerts.
```promql
# S1 An MV is more than 5 minutes behind (for: 2m)
doris_fe_ivm_mv_data_delay_seconds > 300
# S1 Some MV, including those beyond the quota, is more than 15 minutes
behind (for: 5m)
doris_fe_ivm_mvs_by_data_delay{le="+Inf"} - ignoring(le)
doris_fe_ivm_mvs_by_data_delay{le="900"} > 0
# S1 Some MV needs a rebuild, or its state is unknown (for: 10m)
doris_fe_ivm_mvs{state=~"rebuild_required|unknown"} > 0
# S2 Three failures in a row; failures by reason over all MVs
doris_fe_ivm_mv_consecutive_failures >= 3
sum by (reason) (increase(doris_fe_ivm_refresh_failures_total[15m])) > 0
# S2 A task running for over an hour, a task queued for over 10 minutes, a
rejected task
doris_fe_ivm_task_oldest_running_age_seconds > 3600
doris_fe_ivm_task_oldest_pending_age_seconds > 600
sum(increase(doris_fe_ivm_schedule_rejected_total[10m])) > 0
# S3 Fallbacks in the last hour; one MV fell back 3 times or more
sum by (reason) (increase(doris_fe_ivm_fallbacks_total[1h])) > 0
increase(doris_fe_ivm_mv_fallbacks_total[1h]) >= 3
# S4 A stream cannot resume; less than 30 minutes before the TTL reaches an
offset
# (with kind="ivm" it warns that the next refresh of an MV can only
recompute)
doris_fe_changelog_streams{state="cannot_resume"} > 0
doris_fe_changelog_stream_min_retention_headroom_seconds < 1800
# S4 A consumption skipped expired changes (one stream / all of them)
increase(doris_fe_changelog_stream_gap_events_total[10m]) > 0
sum(increase(doris_fe_changelog_stream_gap_commits_total[10m])) > 0
# S4 A stream has had a change waiting for more than 10 minutes
doris_fe_changelog_stream_pending_age_seconds > 600
# S5 A TTL cleanup failed
sum(increase(doris_be_changelog_ttl_cleanup_total{result="failed"}[30m])) > 0
# The monitoring itself is stale: no round completed for 2 minutes, or no
Master exports it
time() - doris_fe_changelog_ivm_monitor_last_sample_timestamp_seconds > 120
absent(doris_fe_changelog_ivm_monitor_last_sample_timestamp_seconds)
```
The thresholds are examples, to be set from the freshness target, the usual
refresh time and the TTL. Delay and gap alerts are best grouped per MV or per
stream, with a link to the queries below, so that thousands of partitions do
not alert at once. S2 alerts can be silenced while a job is paused on purpose;
the delay still shows.
### Find the cause with SQL
The alert tells that something is wrong and how badly; the system tables
tell which MV or stream, which base table, which partition, and why. Grafana's
MySQL data source shows them as table panels.
```sql
-- S1, S2: which MVs are not fresh, and why
SELECT DB_NAME, MV_NAME, FRESHNESS_STATE, DATA_DELAY_SECONDS,
DELAY_PRECISION,
LAGGING_REASON, CONSECUTIVE_FAILURES, LAST_FAILURE_REASON, SAMPLE_TIME
FROM information_schema.ivm_mv_status
WHERE FRESHNESS_STATE <> 'FRESH'
ORDER BY DATA_DELAY_SECONDS DESC;
-- S1: which base table holds an MV back
SELECT BASE_TABLE_DB, BASE_TABLE_NAME, IS_EXCLUDED_TRIGGER, DEPENDENCY_STATE,
DELAY_SECONDS, LAGGING_PARTITIONS, WORST_PARTITION, STREAM_ID
FROM information_schema.ivm_dependencies
WHERE DB_NAME = 'demo' AND MV_NAME = 'mv_orders'
ORDER BY DELAY_SECONDS DESC;
-- S1, S4: the partitions behind, and how close the TTL is
SELECT UNIT, CONSUMED_TSO, SOURCE_END_TSO, PENDING_AGE_SECONDS,
PENDING_AGE_PRECISION,
CAN_RESUME, RETENTION_HEADROOM_SECONDS, LAST_GAP_TIME
FROM information_schema.table_stream_consumption
WHERE STREAM_ID = 12345
ORDER BY PENDING_AGE_SECONDS DESC
LIMIT 20;
-- S2, S3: how the recent tasks refreshed, why they fell back, and the query
ID for the profile
SELECT TaskId, Status, RefreshMode, IvmRefreshMethod, IvmFallbackReason,
DurationMs, ErrorMsg, LastQueryId
FROM tasks("type" = "mv")
WHERE MvDatabaseName = 'demo' AND MvName = 'mv_orders'
ORDER BY CreateTime DESC
LIMIT 20;
-- S4: the user streams at risk
SELECT DB_NAME, STREAM_NAME, MONITOR_STATE, MAX_PENDING_AGE_SECONDS,
MIN_RETENTION_HEADROOM_SECONDS, CAN_RESUME, GAP_EVENTS, LAST_GAP_TIME
FROM information_schema.table_streams
WHERE IVM_MV_ID IS NULL AND MONITOR_STATE <> 'CAUGHT_UP'
ORDER BY MIN_RETENTION_HEADROOM_SECONDS;
-- S5: the changelog of one table
SHOW DATA FROM demo.orders;
```
### Dashboards
Three pages, shipped with the documentation (T7):
1. **Overview**: MVs and streams by state, the largest delay and the delay
buckets, the smallest headroom, fallbacks and failures by reason over the last
hour, the oldest queued and the longest running task, the changelog size, the
health of the monitor, and a table of the MVs in `ivm_mv_status` that are not
fresh.
2. **One MV**: delay and state over time, consecutive failures, fallbacks,
the last duration; tables of its rows in `ivm_dependencies` and its recent
`tasks()`.
3. **One stream**: pending age, headroom, state and gaps over time; a table
of its worst partitions in `table_stream_consumption`.
### Reference
<details><summary>Metrics: 21 cluster-level and 9 per-object on the FE, 3 on
the BE</summary>
**FE, cluster-level** (about 110 series, however many objects there are)
| Metric | Type | Labels | Meaning | For |
| --- | --- | --- | --- | --- |
| `doris_fe_changelog_streams` | gauge | `kind` = user, ivm; `state` = the 7
stream states | streams in each state | S4 |
| `doris_fe_changelog_stream_max_pending_age_seconds` | gauge | `kind` | the
largest known pending age | S4 |
| `doris_fe_changelog_stream_min_retention_headroom_seconds` | gauge |
`kind` | the smallest known retention headroom, may be negative | S4, S3 |
| `doris_fe_changelog_read_window_events_total` | counter | `outcome` =
expired_rejected, expired_clipped; `consumer` = stream, ivm, query | TTL
rejections and clips when a read is planned, once per planning, retries and
uncommitted queries included | S4 |
| `doris_fe_changelog_stream_gap_commits_total` | counter | | consuming
transactions committed after a clipped read | S4 |
| `doris_fe_changelog_retained_bytes` | gauge | `scope` = single_replica,
all_replicas | the changelog the tables keep; cloud mode exports single_replica
only | S5 |
| `doris_fe_ivm_mvs` | gauge | `state` = the 5 freshness states | IVM MVs in
each state | S1 |
| `doris_fe_ivm_mv_max_data_delay_seconds` | gauge | | the largest known
data delay | S1 |
| `doris_fe_ivm_mvs_by_data_delay` | gauge | `le` = 60, 300, 900, 3600, +Inf
| MVs whose delay is at most `le` seconds, cumulative, known values only | S1 |
| `doris_fe_ivm_mvs_excluded_dependency_outdated` | gauge | | MVs with an
excluded dependency changed and not covered | S1 |
| `doris_fe_ivm_refresh_tasks_total` | counter | `result` = success,
no_change, failed, canceled; `method` = none, incremental, partition_rebuild,
complete, mixed | finished refresh tasks, each counted once | S2, S3 |
| `doris_fe_ivm_refresh_duration_seconds_total` | counter | `method` | run
time of the tasks that succeeded or failed; divided by the above, the average |
S3 |
| `doris_fe_ivm_fallbacks_total` | counter | `reason` | tasks that fell back
| S3 |
| `doris_fe_ivm_refresh_failures_total` | counter | `reason` | tasks that
failed | S2 |
| `doris_fe_ivm_schedule_rejected_total` | counter | | refresh tasks
rejected because the task queue was full | S2 |
| `doris_fe_ivm_task_oldest_pending_age_seconds` | gauge | | how long the
oldest queued refresh task has waited, 0 if none | S2 |
| `doris_fe_ivm_task_oldest_running_age_seconds` | gauge | | how long the
longest running refresh task has run, 0 if none | S2 |
| `doris_fe_changelog_ivm_monitor_last_sample_timestamp_seconds` | gauge | |
when the latest round completed | all |
| `doris_fe_changelog_ivm_monitor_sample_duration_seconds` | gauge | | how
long the latest round took | all |
| `doris_fe_changelog_ivm_monitor_sample_failures_total` | counter | |
rounds that failed | all |
| `doris_fe_changelog_ivm_monitor_objects_not_exported` | gauge | `type` =
mv, stream | objects beyond the quota, which have no per-object series | all |
`doris_fe_changelog_read_window_events_total` is counted by every FE that
plans a read. The five refresh counters (`refresh_tasks`,
`refresh_duration_seconds`, `fallbacks`, `refresh_failures`,
`schedule_rejected`) are exported by every FE and counted on the Master only,
so sum them over the FEs. Everything else comes from the Master only. The
counters held by the Master start from 0 after a restart or a Master switch, as
any counter reset.
**FE, per object** (from the Master, under the quota)
| Metric | Type | Meaning |
| --- | --- | --- |
| `doris_fe_ivm_mv_data_delay_seconds` | gauge | data delay, -1 when unknown
|
| `doris_fe_ivm_mv_freshness_state` | gauge | the freshness state as a
number |
| `doris_fe_ivm_mv_consecutive_failures` | gauge | consecutive failures |
| `doris_fe_ivm_mv_fallbacks_total` | counter | fallbacks; persisted, so it
stays monotonic across restarts and Master switches |
| `doris_fe_ivm_mv_last_refresh_duration_seconds` | gauge | run time of the
latest task that succeeded or failed |
| `doris_fe_changelog_stream_state` | gauge | the stream state as a number |
| `doris_fe_changelog_stream_pending_age_seconds` | gauge | the largest
pending age over the partitions, -1 when unknown |
| `doris_fe_changelog_stream_retention_headroom_seconds` | gauge | the
smallest headroom over the partitions with a TTL; not exported when no
partition has one or it is unknown |
| `doris_fe_changelog_stream_gap_events_total` | counter | committed
consumptions that skipped expired changes; persisted |
The MV series carry `db_id`, `db_name`, `mv_id` and `mv_name`; the stream
series, exported for user streams only, carry `db_id`, `db_name`, `stream_id`
and `stream_name`.
**BE**
| Metric | Type | Labels | Meaning |
| --- | --- | --- | --- |
| `doris_be_changelog_ttl_cleanup_total` | counter | `result` = success,
failed | TTL cleanups that finished |
| `doris_be_changelog_ttl_reclaimed_bytes_total` | counter | | bytes of
expired changelog the cleanups dropped |
| `doris_be_changelog_ttl_last_success_timestamp_seconds` | gauge | | when a
cleanup last succeeded |
</details>
<details><summary>System tables: two new tables, and new columns on
<code>table_streams</code>, <code>table_stream_consumption</code> and
<code>tasks()</code></summary>
**`information_schema.ivm_mv_status`** (new): one row per IVM MV, from the
latest round, filtered by the MV's SHOW privilege as `mv_infos()` is.
| Column | Meaning |
| --- | --- |
| `DB_NAME`, `MV_NAME`, `MV_ID` | the MV |
| `FRESHNESS_STATE` | the freshness state |
| `DATA_DELAY_SECONDS`, `DELAY_PRECISION` | the data delay, and `PRECISE`,
`UPPER_BOUND` or `UNKNOWN` |
| `LAGGING_REASON` | why the MV lags |
| `LAGGING_DEPENDENCIES` | trigger dependencies with changes the MV does not
cover yet |
| `DIRTY_PARTITIONS` | MV partitions in the maintained range that need a
rebuild |
| `PENDING_PUBLISH_TXNS` | refresh transactions committed and not visible
yet |
| `EXCLUDED_OUTDATED_DEPENDENCIES`, `MAX_EXCLUDED_DELAY_SECONDS` | excluded
dependencies that changed since the MV covered them, and the largest of their
delays |
| `MIN_RETENTION_HEADROOM_SECONDS` | the smallest headroom of its internal
streams; a low one means the next refresh may only recompute |
| `JOB_STATUS` | the state of the refresh job |
| `CURRENT_TASK_ID`, `CURRENT_TASK_STATUS`, `CURRENT_TASK_AGE_SECONDS` | the
task running or queued |
| `CONSECUTIVE_FAILURES` | consecutive failures |
| `LAST_SUCCESS_TIME`, `LAST_FAILURE_TIME`, `LAST_FAILURE_REASON`,
`LAST_ERROR_MESSAGE` | the error message is cut to 512 characters |
| `LAST_REFRESH_METHOD`, `LAST_REFRESH_DURATION_MS` | of the latest task
that succeeded or failed |
| `FALLBACK_COUNT`, `LAST_FALLBACK_REASON`, `LAST_FALLBACK_TIME` | fallbacks
|
| `SAMPLE_TIME` | the round the values come from |
The failure, fallback and duration columns come from a summary persisted on
the MV, so the bounded task history does not limit them.
**`information_schema.ivm_dependencies`** (new): one row per IVM MV and
direct base table, from the latest round, with the same privilege check.
| Column | Meaning |
| --- | --- |
| `DB_NAME`, `MV_NAME`, `MV_ID` | the MV |
| `BASE_TABLE_CTL`, `BASE_TABLE_DB`, `BASE_TABLE_NAME`, `BASE_TABLE_ID` |
the base table |
| `IS_EXCLUDED_TRIGGER` | the table is in `excluded_trigger_tables` |
| `STREAM_NAME`, `STREAM_ID` | the internal stream, NULL for an excluded
dependency; join `table_stream_consumption` on `STREAM_ID` for the partitions |
| `DEPENDENCY_STATE` | `USABLE`, `MISSING`, `DISABLED`, `STALE`,
`CANNOT_RESUME` or `EXCLUDED` |
| `COVERED` | the visible data of the MV covers every change of this
dependency |
| `LAGGING_PARTITIONS`, `WORST_PARTITION` | the source partitions not
covered, and the one with the largest delay |
| `DELAY_SECONDS`, `DELAY_PRECISION` | the delay of this dependency |
| `COVERED_VERSION`, `CURRENT_VERSION` | for an excluded dependency: the
version the MV's refresh snapshots cover, and the visible one |
| `MIN_RETENTION_HEADROOM_SECONDS`, `CAN_RESUME` | for a trigger dependency |
| `SAMPLE_TIME` | the round the values come from |
**`information_schema.table_streams`**, new columns: one row per stream,
from the latest round.
| Column | Meaning |
| --- | --- |
| `IVM_MV_ID` | the MV that owns an internal stream; NULL for a user stream |
| `MONITOR_STATE` | the stream state |
| `PENDING_PARTITIONS` | partitions not caught up |
| `MAX_PENDING_AGE_SECONDS`, `PENDING_AGE_PRECISION` | the largest pending
age over the partitions, and its precision |
| `CAN_RESUME` | every partition with a TTL can resume; NULL when none has
one |
| `MIN_RETENTION_HEADROOM_SECONDS` | the smallest headroom over the
partitions with a TTL |
| `GAP_EVENTS`, `LAST_GAP_TIME` | how many committed consumptions skipped
expired changes, and when the latest committed; persisted |
| `MONITOR_SAMPLE_TIME` | the round the values come from; NULL before the
first one |
**`information_schema.table_stream_consumption`**, new columns: one row per
stream and source partition, computed at query time like the existing columns,
the pending age from the sampler's change points. `CONSUMPTION_STATUS`, `LAG`
and `LAST_CONSUMPTION_TIME` stay as they are.
| Column | Meaning |
| --- | --- |
| `SOURCE_END_TSO` | the visible end of the source partition; NULL when it
has no data |
| `CONSUMED_TSO` | the committed offset, inclusive; NULL when there is none |
| `OFFSET_STATE` | `CONSUMED`, `INITIAL_SNAPSHOT_PENDING` or `NO_OFFSET` |
| `CAUGHT_UP` | the offset is at the visible end |
| `POSITION_GAP_SECONDS` | the physical time between the offset and the end;
for trends only, it is not the pending age |
| `PENDING_AGE_SECONDS`, `PENDING_AGE_PRECISION` | the pending age, 0 when
caught up |
| `RETENTION_APPLICABLE` | the table has a TTL and the partition has an
offset |
| `RETENTION_FLOOR_TSO`, `NEXT_REQUIRED_TSO`, `CAN_RESUME`,
`RETENTION_HEADROOM_SECONDS` | see "Definitions" below; NULL when not
applicable |
| `LAST_GAP_TIME` | when a consumption that skipped expired changes of this
partition last committed |
**`tasks("type"="mv")`**, new column `IvmRefreshMethod`: how the task
refreshed an IVM MV, `NONE`, `INCREMENTAL`, `PARTITION_REBUILD`, `COMPLETE` or
`MIXED`; NULL while the task runs and for an MV that is not an IVM MV.
`RefreshMode`, `IvmFallbackReason` and `IvmRebuiltPartitions` stay as they are.
</details>
<details><summary>FE configs</summary>
All are mutable and take effect on the Master FE.
| Config | Default | Meaning |
| --- | --- | --- |
| `enable_changelog_ivm_monitor` | true | with `enable_feature_changelog`;
when off, nothing is sampled, no state series is exported, and the sampled
columns are NULL |
| `changelog_ivm_monitor_interval_seconds` | 15 | the sampling interval,
which is also the precision of the pending age and the data delay |
| `changelog_ivm_monitor_max_partitions_per_round` | 50000 | beyond it, the
objects are sampled in turns |
| `changelog_ivm_monitor_max_points_per_partition` | 64 | change points kept
per partition |
| `changelog_ivm_monitor_max_points` | 1000000 | change points kept in total
|
| `prom_output_ivm_mv_metrics_limit` | 1000 | MVs with per-object series; 0
for none |
| `prom_output_table_stream_metrics_limit` | 1000 | user streams with
per-object series; 0 for none |
| `enable_ivm_stream_keepalive` | true | when off, no new keep-alive TSO is
recorded; the recorded ones still count |
| `ivm_stream_keepalive_interval_second` | 600 | the keep-alive round; keep
it well under half the shortest `changelog.ttl_seconds` of the IVM base tables |
</details>
## 3. Design
### Metrics detect, system tables locate
The two carriers can do different things:
| | Metrics (`/metrics` to Prometheus) | System tables
(`information_schema`, TVFs) |
| --- | --- | --- |
| Alerting | evaluated continuously, unattended | needs an external program
that runs SQL |
| History and rates | kept by the time-series database | the current state
only |
| Events between two looks | a counter misses none | a snapshot misses them
unless they are persisted |
| Cost | per series, paid on every scrape | per query, with filters |
| Exact values | float64, which cannot hold a TSO (about 4.7×10¹⁷, above
2⁵³) | exact BIGINT |
| Text | finite enums only | error messages, query IDs, dependency lists |
| Privileges | no per-object privilege check on the FE's `/metrics` |
filtered by object privilege |
So, in this order:
1. A signal that needs unattended alerting or a trend is a metric, and
always has a cluster-level series first, so that no number of objects hides a
problem.
2. An event that can happen between two samples (a clipped read, a fallback,
a failure, a rejected task) is a counter. When it has to be traced back to an
object, the object also persists a summary, a count and the latest time, which
a system table shows.
3. Details that grow with partitions, tablets, tasks or transactions, and
exact TSOs and versions, error text and dependency lists, go into system tables
only.
4. The key health values of an object (data delay, state, consecutive
failures) are in both: the system table has every object with the reason, and
the metrics carry a few golden signals per object, under a quota.
Both sides come from the same sampler and the same computation, so they
cannot disagree.
### A bounded number of series
Exporting every signal for every object would cost about 80 series per IVM
MV (`reason × phase` and `requested × actual × result` alone have dozens of
combinations) and 23 per stream. With 2,000 MVs over 3 base tables each (6,000
internal streams) and 1,000 user streams, that is about 320,000 series and 50
MB per scrape of the Master FE, whose `/metrics` is rendered under one lock
(`MetricRepo.getMetric` is `static synchronized`). Doris already caps the
per-table series of cloud mode with `prom_output_table_metrics_limit` (#50586),
and the BE exports tablet-level metrics only on `?with_tablet=true`. So:
- **Cluster-level aggregates by default**: counts by state, the worst values
(the largest delay, the smallest headroom), delay buckets, and event counters
over finite enums. About 110 series, however many objects there are, and every
object counts in them.
- **A few golden signals per object**: 5 series per IVM MV and 4 per user
stream. Reasons, phases and methods appear at cluster level only, and there is
no histogram per object. A state is a number, not a label value, so a change of
state does not start a new series; a commit-triggered MV can switch between
fresh and lagging every minute.
- **No series per internal stream**: the delay and the headroom of an MV
already take the worst of its internal streams, whose details are in
`ivm_dependencies` and `table_stream_consumption`. The cluster-level stream
series tell user and internal streams apart with `kind`.
- **A quota per object type**: `prom_output_ivm_mv_metrics_limit` and
`prom_output_table_stream_metrics_limit`, named after
`prom_output_table_metrics_limit`, 1,000 each, 0 to turn them off. Beyond the
quota the objects with the lowest IDs are exported, which is a stable choice,
and `doris_fe_changelog_ivm_monitor_objects_not_exported{type}` tells how many
are left out. They still count in the cluster-level series.
- **Labels carry stable identities only**: the database and object ID and
name, as the per-job series of routine load do. No partition, tablet, task,
transaction, offset or error text. Label values are escaped (T8).
With the default quotas that is at most about 9,100 series, about 1 MB.
Not chosen: exporting the worst N objects only (they move in and out
quickly, which breaks the `for:` of an alert), exporting the unhealthy objects
only (a missing series could then mean healthy, not sampled or dropped), one
series per object with the state in a label (a new series at every state
change), and putting the per-object series behind a separate scrape parameter
as the BE's `with_tablet` does (every small cluster would need a second scrape
job).
### One sampler on the Master FE
`ChangelogIvmMonitor` is a daemon of the Master FE, as `TableStreamManager`
and `CloudTabletStatMgr` are. Every `changelog_ivm_monitor_interval_seconds`
(15 by default) it runs a round:
1. Under each object's lock it only copies: offsets; the IDs, visible
versions, TSOs and visible times of the base partitions; the partition states
and refresh snapshots of the MVs; the job and task states. This is what
`table_stream_consumption` does today, and no RPC is sent under a metadata lock.
2. Outside the locks, in cloud mode, it reads every stream partition of the
round in one batched MetaService call, as the system table does.
3. It updates the change points (below), computes the values of every
object, and publishes an immutable snapshot by replacing one reference. A round
that fails keeps the previous snapshot and counts the failure, and
`last_sample_timestamp` shows how old the snapshot is.
4. Beyond `changelog_ivm_monitor_max_partitions_per_round` partitions, the
objects are sampled in turns, each with its own sample time.
`/metrics` only renders the latest snapshot, through a `MetricVisitor`
method beside `visitCloudTableStats`, instead of registering the series again
on every scrape. Only the Master exports the state series. Another FE exports
none of them rather than 0, because a 0 from every other FE, as the routine
load lag gauges give today, drags a `min` over the FEs to 0. A new Master
exports none until its first round completes, which the `absent()` alert above
covers. Events are counted on the FE where they happen (refresh task events on
the Master, TTL rejections and clips on the FE that plans the query), never in
replay.
### Definitions
**Resume and retention headroom** are computed with the functions the
planner uses (`ChangelogUtils`, `TSOTimestamp`), not an approximation of them:
```text
cutoff = calculateCutoff(current TSO, ttl_seconds)
retention_floor = nextTso(cutoff)
next_required = nextTso(C) C is the offset, inclusive
can_resume = next_required >= retention_floor
headroom_seconds = (physical(next_required) - physical(retention_floor)) /
1000, may be negative
```
They apply when the base table has a TTL and the partition has an offset. A
stream caught up on a partition nobody writes still drifts out of the
retention: its next `MIN_DELTA` read is rejected and its next `DETAIL` or
`APPEND_ONLY` read clipped, and a user stream reports that as `cannot_resume`.
An IVM internal stream uses its effective offset `max(C, K)` with the
keep-alive TSO K (below), so an idle one does not.
**Pending age** is the age of the oldest change that the source has made
visible and the offset does not cover, 0 when caught up. A `Partition` keeps
only its latest visible version, time and TSO, and compaction merges the TSO
ranges of rowsets, so the time a change first became visible cannot be
recovered afterwards. The sampler observes it instead. Whenever the visible end
of a tracked partition has advanced since the previous round, it records a
change point: the new end TSO and version, and the window in which those
changes became visible, from the previous round to this one. For a consumer at
offset C, the first point beyond C bounds the age, and the upper end of the
bound is reported, so a delay is never reported too small.
- One new version in a round gives the exact time (the visible time of the
version, or in cloud mode the physical time of its TSO); several are off by one
sampling interval at most. Both are `PRECISE`.
- After a Master switch or a restart there is no history: a consumer already
behind reports `now - physical(C)`, `UPPER_BOUND`, until it catches up once.
- An offset that cannot be read (a cloud read state `UNKNOWN`, a failed
round) gives no value, `UNKNOWN`.
- A partition with data and no offset yet counts from the creation of the
stream, `UPPER_BOUND`. A stream whose initial snapshot is not consumed yet is
`initializing` and has no pending age.
A partition keeps only the points beyond the smallest offset of its
consumers, so consumers that are caught up keep none. Beyond
`changelog_ivm_monitor_max_points_per_partition` the two oldest adjacent points
merge, which only widens the bound, and `changelog_ivm_monitor_max_points`
(about 32 MB) caps the total. Change points are not persisted: losing them all
only falls back to the `UPPER_BOUND` rule. A value exact across restarts would
need new persisted metadata on the commit and publish paths and in MetaService;
this does without.
**MV data delay** is the largest of:
1. The delay of each trigger dependency over the maintained range (the MV's
partition sync range, when it has one), computed as above with the internal
stream at the MV's *visible coverage* instead of its current offset. The offset
moves when the refresh transaction is COMMITTED, before the new data of the MV
is VISIBLE, so in shared-nothing mode the sampler takes, for the refresh
transactions committed and not visible yet, the offsets from before them
(recorded in `OlapTableStreamUpdate`). In cloud mode a commit is visible at
once.
2. The age of a dirty MV partition in the maintained range: from the round
before the one that first saw it dirty (`PRECISE`), or, with no history, from
the MV's latest successful refresh (`UPPER_BOUND`).
It is 0 when everything is covered and no partition is dirty, and has no
value for an MV that never refreshed successfully (`initializing`). Its
precision is the worst of its parts. A chain of IVM MVs is measured against the
direct dependencies only. Tables in `excluded_trigger_tables` are left out of
it on purpose: their changes are not meant to trigger a refresh, and counting
them would keep such MVs over the threshold for good. Their coverage is judged
from the versions in the MV's refresh snapshots and reported on its own
(`EXCLUDED_OUTDATED_DEPENDENCIES`, `MAX_EXCLUDED_DELAY_SECONDS`,
`doris_fe_ivm_mvs_excluded_dependency_outdated`).
**States**: the first one that matches, with the number the per-object gauge
exports.
| Stream state | Number | When |
| --- | --- | --- |
| `disabled` | 6 | `ENABLED = false` |
| `stale` | 5 | `IS_STALE = true`, including a dropped base table |
| `unknown` | 4 | the round failed, or an offset cannot be read |
| `cannot_resume` | 3 | a partition with a TTL cannot resume: the next read
is rejected (`MIN_DELTA`) or clipped (`DETAIL`, `APPEND_ONLY`) |
| `initializing` | 2 | the initial snapshot is not consumed yet |
| `lagging` | 1 | a partition has changes beyond its offset, or data and no
offset |
| `caught_up` | 0 | otherwise |
| MV freshness state | Number | When |
| --- | --- | --- |
| `unknown` | 4 | the round failed, an internal stream is missing, or an
offset cannot be read |
| `initializing` | 3 | never refreshed successfully |
| `rebuild_required` | 2 | a dirty partition in the maintained range |
| `lagging` | 1 | changes not covered yet, including a refresh being
published |
| `fresh` | 0 | otherwise |
`LAGGING_REASON`, in the system table only, is the first that matches of
`REBUILD_REQUIRED`, `PUBLISHING` (a refresh committed and not visible yet),
`RUNNING`, `QUEUED`, `REFRESH_FAILED`, `JOB_PAUSED`, `WAITING_SCHEDULE` (a
scheduled refresh not due yet) and `WAITING_TRIGGER` (a commit-triggered or
manual refresh, no task yet); `NONE` when fresh.
**Refresh result, method and fallback.** Each finished task counts once:
- `result`: `success`, `no_change` (a success that committed no MV data),
`failed` or `canceled`. Consecutive failures go up on `failed`, back to 0 on
`success` and `no_change`, and stay on `canceled`.
- `method`, from the attempts that committed data: `none`, `incremental`,
`partition_rebuild`, `complete`, or `mixed`, typically the rebuild of dirty
partitions before the delta. It is a different dimension from `RefreshMode`.
- A **fallback** is a task that planned an incremental attempt, moved to a
partition rebuild or a complete rebuild for an IVM reason, and committed data.
Its reason is `IvmFallbackReason`: a value of `IvmFailureReason`, or
`INCOMPLETE_REFRESH_SNAPSHOT`. Not fallbacks: building the first baseline, an
explicit `COMPLETE` or `PARTITIONS` request, rebuilding dirty partitions before
the delta, a fallback that committed nothing, and a canceled task.
- A **failure reason** is the IVM reason when falling back is not allowed,
and otherwise the stage that threw: `PARTITION_SYNC_FAILED`, `QUERY_UNUSABLE`,
`EXECUTION_FAILED` or `OTHER`, told apart by where the failure was thrown
rather than by its message. An expired offset is `CHANGELOG_BROKEN` (T0).
Label values on metrics are lower case.
**Unknown values.** System tables give NULL, with a state or precision
column that says why. Per-object gauges give -1 for an unknown delay or age, as
the per-job lag of streaming jobs does; a headroom can be negative, so an
unknown or inapplicable one is not exported. Cluster-level maxima and minima
count the known values only, and an object with unknown values counts in
`state="unknown"`. No NaN.
### What is persisted, and where events are counted
- **Refresh summary of an MV.** `MTMVTask` persists its method, its failure
reason and whether it fell back. The MV keeps a summary that
`MTMV.addTaskResult` updates, the same code on the Master and in replay, so it
is deterministic, survives restarts and Master switches, and does not depend on
the bounded task history: consecutive failures, the latest success and failure
with the reason and the message, the latest method and duration, and the
fallback count with the latest reason and time. A canceled task leaves it as it
is. The refresh counters count in `MTMVTask.after()` on the Master and never in
replay; a task the full task queue rejects counts in
`doris_fe_ivm_schedule_rejected_total`.
- **TTL rejections, clips and gaps.** The two places that apply the TTL, the
stream scan (`NormalizeOlapTableStreamScan.applyChangelogTtl`) and the direct
incremental query (`BindRelation.applyChangelogTtl`), count each rejection or
clip in `doris_fe_changelog_read_window_events_total`, except a clip of a
partition with no change after the offset, which loses nothing. A consuming DML
records the partitions it clipped in `OlapTableStreamUpdate`; when the
transaction is COMMITTED, the stream counts a gap event and the time of the
latest gap of each of those partitions, journaled and replayed (replay does not
count `doris_fe_changelog_stream_gap_commits_total`). In cloud mode MetaService
keeps them, in two new optional fields of `TableStreamOffsetPB`
(`gap_event_count`, `last_gap_time_ms`) that it updates in the transaction that
moves the offset.
- **TTL cleanup on the BE.** A TTL cleanup is a cumulative compaction of a
changelog tablet that leaves the expired rowsets out of the merge. It counts
once when the compaction finishes (`CumulativeCompaction::execute_compact`, and
in cloud mode `CloudCumulativeCompaction::record_compaction_success` or
`record_compaction_failure`), and only when its input held expired rowsets; a
compaction that never started, for example on a busy lock, does not count. The
reclaimed bytes are the `total_disk_size` of the expired rowsets a successful
cleanup dropped. The existing bvars stay.
- **Changelog size** sums `OlapTable.Statistics` over the tables with a
changelog, which `TabletStatMgr` or `CloudTabletStatMgr` refreshes periodically.
### Keep-alive TSO for idle IVM partitions (T9)
An offset moves only when a refresh consumes the partition, and a refresh
consumes only the partitions with new data, so the offset of a partition nobody
writes stays at its last change. The TTL check compares that offset with `now -
TTL` and cannot know that nothing was written in between, so the first write
after a day of idleness makes the next refresh reject the offset and rebuild
the whole MV.
The Master records a keep-alive TSO K for the idle partitions of IVM
internal streams, meaning that the partition holds no change in (C, K], and the
TTL check of an IVM read uses `max(C, K)`. The offset C itself does not move.
- A daemon runs a round every `ivm_stream_keepalive_interval_second` (600 by
default) over the partitions that are caught up, past the initial snapshot, on
a base table with a TTL, and with less than half the TTL left.
- In shared-nothing mode K is a TSO taken under the base table read lock: a
commit takes its TSO under the write lock, so any later commit gets a larger
one. A partition with a committed transaction that is not visible yet is
skipped.
- In cloud mode K is the committed watermark of the TSO service. The round
first advances MetaService's TSO fence to K, so that no commit at or below K
can still land, then reads the offsets in one batched call.
- K is kept in FE metadata with the offset it was taken against, and a read
uses it only while the offset is still that one.
- A round that fails records nothing and leaves the partition to the next
round. The refresh after a write stays incremental when it starts within half
the TTL minus one interval after the write, 11 hours 50 minutes with the
defaults.
User streams get no keep-alive, so the monitoring reports an idle user
stream as `cannot_resume`.
### Compatibility
- **Persisted state** is new GSON fields only, which an older FE ignores, so
a rollback is safe, and the new cloud proto fields are optional. One exception:
#69831 merges the per-partition maps of `OlapTableStream` into one map. Table
streams are not released yet, so a stream from an older image has to be dropped
and created again.
- **New columns on existing tables.** The BE scanners check how many columns
the FE returns. `tso_status` once added columns by checking whether the request
carries `columns_name`, which works only once, and these tables get columns
twice (T2, then T5). So the FE fills the columns by the names the request lists
(NULL for a name it does not know, and only the original columns for a request
without names), and a new BE accepts an FE that returns fewer trailing columns
and fills them with NULL. A cluster with mixed versions can query both tables
whichever is upgraded first.
- **New tables** append values to `TSchemaTableType` and `TSchemaTableName`
and never reuse one (`gensrc/AGENTS.md`).
- **Privileges.** The two new tables are filtered by the MV's SHOW
privilege; the extended tables keep their current visibility rules.
## 4. Sub tasks
`[x]` merged, `[ ]` in review or not started. The IDs are stable references,
not the order of work.
### Fixes
- [x] **T0. Fall back to COMPLETE when an IVM stream offset has expired.**
#69772, merged 2026-10-28.
- [x] **T10. Reject an expired offset in every IVM read, including one of a
duplicate-key base table.** #69785, merged 2026-10-29.
- [x] **T11. Read a base partition with no offset from the base table in an
IVM refresh.** #69778, merged 2026-10-29.
- [ ] **T9. Keep the offsets of idle IVM stream partitions alive**, in
shared-nothing and cloud mode. #69831, in review.
- [ ] **T12. Read a partition with no offset from the base table for user
streams read as `MIN_DELTA`, and for streams on duplicate-key tables.** Within
the TTL the output is the same as before; past it, the rows come back.
`APPEND_ONLY` and `DETAIL` streams on unique-key tables stay as they are,
because reading the base table would change their output within the TTL. The PR
follows #69831, which it shares a test fixture with.
- [ ] **T8. Escape label values in the Prometheus output of the FE**, with
one function for the existing series and the new ones. #69837, in review.
### Monitoring
- [ ] **T1. Monitor framework.** `ChangelogIvmMonitor` on the Master, the
immutable snapshot and its rendering in `/metrics`;
`doris_fe_changelog_retained_bytes{scope}` and the three series on the health
of the monitor; `enable_changelog_ivm_monitor` and
`changelog_ivm_monitor_interval_seconds`. #69848, in review.
- [ ] **T3. How a refresh refreshed.** `IvmRefreshMethod` in `tasks()`; the
method, failure reason and fallback persisted with each task; the refresh
summary of an MV; `doris_fe_ivm_refresh_tasks_total`,
`doris_fe_ivm_refresh_duration_seconds_total`, `doris_fe_ivm_fallbacks_total`,
`doris_fe_ivm_refresh_failures_total` and
`doris_fe_ivm_schedule_rejected_total`. #69856, in review.
- [ ] **T6. TTL cleanup metrics on the BE.**
`doris_be_changelog_ttl_cleanup_total{result}`,
`doris_be_changelog_ttl_reclaimed_bytes_total` and
`doris_be_changelog_ttl_last_success_timestamp_seconds`. #69868, in review.
- [ ] **T2. Stream monitoring.** Offsets, resume, headroom, state, change
points and pending age, for user and internal streams;
`doris_fe_changelog_streams`,
`doris_fe_changelog_stream_max_pending_age_seconds`,
`doris_fe_changelog_stream_min_retention_headroom_seconds`, and the per-stream
state, pending age and headroom; the new columns of `table_streams` and
`table_stream_consumption` except the gap columns; the per-object quota with
`prom_output_table_stream_metrics_limit` and
`objects_not_exported{type="stream"}`; sampling in turns and the three sampler
limits. After T1, T8 and T9.
- [ ] **T4. IVM freshness.** The visible coverage, data delay, state,
lagging reason and excluded dependencies of an MV; `doris_fe_ivm_mvs`,
`doris_fe_ivm_mv_max_data_delay_seconds`, `doris_fe_ivm_mvs_by_data_delay`,
`doris_fe_ivm_mvs_excluded_dependency_outdated`,
`doris_fe_ivm_task_oldest_pending_age_seconds` and
`doris_fe_ivm_task_oldest_running_age_seconds`; the five per-MV series with
`prom_output_ivm_mv_metrics_limit` and `objects_not_exported{type="mv"}`; the
new tables `ivm_mv_status` and `ivm_dependencies`. After T1, T2 and T3.
- [ ] **T5. TTL events and gaps.**
`doris_fe_changelog_read_window_events_total`,
`doris_fe_changelog_stream_gap_commits_total` and the per-stream
`doris_fe_changelog_stream_gap_events_total`; the clipped partitions in
`OlapTableStreamUpdate` and the gap record of a stream; the gap columns; the
two optional fields of `TableStreamOffsetPB` and their update in MetaService.
After T2.
- [ ] **T7. Documentation and dashboard.** User documentation in English and
Chinese, the alert rules above, and a Grafana dashboard. After T0 to T6 and T8.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
