Update materialized view monitoring and system views documentation for introspection state log changes (#596)

Author: yokofly

docs/materialized-view-monitoring.md

@@ -36,7 +36,7 @@ WHERE type = 'MaterializedView'
 
 ## States
 
-The `system.stream_state_log` stream contains detailed state information that can be used for monitoring and alerting.
+In Timeplus Enterprise 3.0 and later, the `system.introspection_state_log` stream contains detailed state information for monitoring and alerting. In earlier releases, use `system.stream_state_log` instead.
 
 Supported states include:
 
@@ -57,15 +57,21 @@ Supported states include:
 SELECT
     *
 FROM
-    table(system.stream_state_log)
+    table(system.introspection_state_log)
 WHERE
-    dimension = 'materialized_view'
+    starts_with(dimension, 'materialized_view')
     AND state_name IN ('end_sn', 'processed_sn');
 ```
 
 ## System Built-in Views
 
-Timeplus provides built-in system views to help monitor Materialized Views:
+Timeplus provides built-in system views to help monitor and debug Materialized Views:
 
 - `system.v_failed_mat_views`: Tracks Materialized Views that have failed.
-- `system.v_mat_view_lags`: Shows Materialized Views with processing lag.
+- `system.v_mat_view_lags`: Shows Materialized Views with processing lag.
+- `system.v_storages`: Shows stream and checkpoint storage usage, which is useful when a Materialized View is accumulating checkpoint data.
+- `system.v_stream_applied_lags`: Shows per-node applied lag for replicated streams when a Materialized View is blocked by upstream replication or storage apply delay.
+
+For the full list of built-in troubleshooting views, see [Views in system namespace](/system-views).
+
+If you are running a 2.x release, replace `system.introspection_state_log` in the examples above with `system.stream_state_log`.

docs/system-views.md

@@ -1,6 +1,8 @@
 # Views in system namespace
 Timeplus provides system views that enable effective troubleshooting and monitoring of your streaming data operations.
 
+In Timeplus Enterprise 3.0 and later, these built-in views are defined on `system.introspection_state_log`. If you are running an older 2.x release, the equivalent state stream is `system.stream_state_log`.
+
 ## v_failed_mat_views
 ```sql
 CREATE OR REPLACE VIEW system.v_failed_mat_views
@@ -9,16 +11,16 @@ WITH running_mvs_in_last_5m AS
 (
     SELECT
       database, name
-    FROM system.stream_state_log
-    WHERE (_tp_time > (now() - 5m)) AND (dimension = 'materialized_view') AND (state_name = 'status') AND (state_string_value = 'ExecutingPipeline')
+    FROM system.introspection_state_log
+    WHERE (_tp_time > (now() - 5m)) AND starts_with(dimension, 'materialized_view') AND (state_name = 'status') AND (state_string_value = 'Executing')
     ORDER BY _tp_time DESC -- order here to make sure we have the latest state
     SETTINGS query_mode = 'table'
 )
 SELECT database, name, state_string_value AS state, _tp_time
-FROM system.stream_state_log
-WHERE (_tp_time > (now() - 5m)) AND (dimension = 'materialized_view') AND (state_name = 'status') AND NOT ((database, name) IN running_mvs_in_last_5m)
+FROM system.introspection_state_log
+WHERE (_tp_time > (now() - 5m)) AND starts_with(dimension, 'materialized_view') AND (state_name = 'status') AND NOT ((database, name) IN running_mvs_in_last_5m)
 SETTINGS query_mode = 'table'
-COMMENT 'version 2';
+COMMENT 'version 3';
 ```
 
 ## v_mat_view_lags
@@ -28,7 +30,7 @@ AS
 WITH last_5m_progressing_status AS
 (
   SELECT database, name, state_name, dimension, state_value, _tp_time AS ts
-  FROM system.stream_state_log
+  FROM system.introspection_state_log
   WHERE (_tp_time > (now() - 5m)) AND (state_name IN ('processed_sn', 'ckpt_sn', 'end_sn'))
   ORDER BY _tp_time DESC -- order here to make sure we have latest state
   SETTINGS query_mode = 'table'
@@ -67,7 +69,7 @@ SELECT
     if (ckpt_sn != 0 AND processed_sn != 0, processed_sn - ckpt_sn, 0) AS ckpt_lag,
     ts
 FROM mv_lagging_aggr_per_mv
-COMMENT 'version 2';
+COMMENT 'version 3';
 ```
 
 ## v_no_leader_shards
@@ -77,23 +79,23 @@ AS
 WITH last_5m_stream_shards AS
 (
   SELECT database, name, dimension AS shard
-  FROM system.stream_state_log
+  FROM system.introspection_state_log
   WHERE _tp_time > (now() - 5m) AND state_name = 'committed_sn'
   GROUP BY database, name, dimension
   SETTINGS query_mode = 'table'
 ),
 last_5m_quorum_status AS
 (
   SELECT database, name, dimension AS shard
-  FROM system.stream_state_log
+  FROM system.introspection_state_log
   WHERE _tp_time > (now() - 5m) AND state_name = 'quorum_replication_status'
   GROUP BY database, name, dimension
   SETTINGS query_mode = 'table'
 )
 SELECT database, name, shard
 FROM last_5m_stream_shards
 WHERE (database, name, shard) NOT IN last_5m_quorum_status
-COMMENT 'version 1';
+COMMENT 'version 2';
 ```
 
 ## v_replication_lags
@@ -111,7 +113,7 @@ WITH recent_replication_statuses AS
     array_map(x -> to_uint64(x:node), replica_statuses) AS replica_nodes,
     map_cast(array_map(x -> to_uint64(x:node), replica_statuses), replica_statuses) AS replicas_map,
     _tp_time AS ts
-  FROM system.stream_state_log
+  FROM system.introspection_state_log
   WHERE (_tp_time > (now() - 5m)) AND (state_name = 'quorum_replication_status')
   ORDER BY _tp_time DESC
   SETTINGS query_mode = 'table'
@@ -138,7 +140,7 @@ SELECT
     to_int64((replicas_map[leader_node]):next_sn) - to_int64((replicas_map[replica_node]):next_sn) AS lagging,
     ts
 FROM latest_replication_statuses
-COMMENT 'version 2';
+COMMENT 'version 3';
 ```
 
 ## v_shard_leaders
@@ -148,15 +150,15 @@ AS
 WITH last_5m_quorum_status AS
 (
   SELECT database, name, dimension AS shard, node_id AS leader, _tp_time AS ts
-  FROM system.stream_state_log
+  FROM system.introspection_state_log
   WHERE _tp_time > (now() - 5m) AND state_name = 'quorum_replication_status'
   ORDER BY _tp_time DESC -- order here to make sure we have latest state
   SETTINGS query_mode = 'table'
 )
 SELECT database, name, shard, earliest(leader) AS leader, earliest(ts) AS  ts
 FROM last_5m_quorum_status
 GROUP BY database, name, shard
-COMMENT 'version 1';
+COMMENT 'version 2';
 ```
 
 ## v_under_replication_replicas
@@ -174,7 +176,7 @@ WITH recent_replication_statuses AS
      array_map(x -> to_uint64(x:node), replica_statuses) AS replica_nodes,
      map_from_arrays(array_map(x -> to_uint64(x:node), replica_statuses), replica_statuses) AS replicas_map,
      _tp_time AS ts
-  FROM system.stream_state_log
+  FROM system.introspection_state_log
   WHERE (_tp_time > (now() - 5m)) AND (state_name = 'quorum_replication_status')
   ORDER BY _tp_time DESC
   SETTINGS query_mode = 'table'
@@ -195,5 +197,114 @@ latest_replication_statuses AS
 SELECT database, name, shard, leader_node, replica_node, (replicas_map[replica_node]):state AS state, ts
 FROM latest_replication_statuses
 WHERE state != 'Replicate'
+COMMENT 'version 3';
+```
+
+## v_stream_applied_lags
+```sql
+CREATE OR REPLACE VIEW system.v_stream_applied_lags
+AS
+WITH applied_sn_data AS
+(
+  SELECT database, name, node_id, state_value AS applied_sn, _tp_time AS ts
+  FROM system.introspection_state_log
+  WHERE (_tp_time > (now() - 5m)) AND (state_name = 'applied_sn')
+  ORDER BY _tp_time DESC
+  SETTINGS query_mode = 'table'
+),
+latest_applied_sn_per_node AS
+(
+  SELECT database, name, node_id,
+    arg_max(applied_sn, ts) AS applied_sn,
+    arg_max(ts, ts) AS latest_ts
+  FROM applied_sn_data
+  GROUP BY database, name, node_id
+),
+quorum_replication_data AS
+(
+  SELECT
+    database,
+    name,
+    state_string_value:shard AS shard,
+    node_id AS leader_node,
+    state_string_value:shard_replication_statuses[*] AS replica_statuses,
+    array_map(x -> to_uint64(x:node), replica_statuses) AS replica_nodes,
+    map_cast(array_map(x -> to_uint64(x:node), replica_statuses), replica_statuses) AS replicas_map,
+    _tp_time AS ts
+  FROM system.introspection_state_log
+  WHERE (_tp_time > (now() - 5m)) AND (state_name = 'quorum_replication_status')
+  ORDER BY _tp_time DESC
+  SETTINGS query_mode = 'table'
+),
+latest_replication_statuses AS
+(
+  SELECT
+    database,
+    name,
+    to_int(shard) AS shard,
+    arg_max(leader_node, ts) AS leader_node,
+    arg_max(replica_nodes, ts) AS replica_nodes,
+    arg_max(replicas_map, ts) AS replicas_map,
+    arg_max(ts, ts) AS latest_ts
+  FROM quorum_replication_data
+  GROUP BY database, name, shard
+),
+combined_sn_data AS
+(
+  SELECT
+    a.database,
+    a.name,
+    a.node_id,
+    a.applied_sn,
+    r.shard,
+    to_int64((r.replicas_map[a.node_id]):next_sn) - 1 AS log_replicated_sn,
+    log_replicated_sn - a.applied_sn AS lags
+  FROM latest_applied_sn_per_node a
+  JOIN latest_replication_statuses r ON a.database = r.database AND a.name = r.name
+)
+SELECT
+  database,
+  name,
+  shard,
+  node_id,
+  log_replicated_sn,
+  applied_sn AS storage_applied_sn,
+  lags
+FROM combined_sn_data
+ORDER BY lags DESC
+COMMENT 'version 1';
+```
+
+## v_storages
+```sql
+CREATE OR REPLACE VIEW system.v_storages
+AS
+WITH storage_sizes AS
+  (
+    SELECT
+      database, name, uuid, dimension AS store_type, any(state_string_value) AS stream_type, sum(state_value) AS total_bytes
+    FROM
+      system.introspection_state_log
+    WHERE
+      (_tp_time > (now() - 5m)) AND (state_name = 'disk_size') AND (database != 'neutron')
+    GROUP BY
+      database, name, uuid, dimension
+    UNION ALL
+    SELECT
+      database, name, uuid, 'ckpt_size' AS store_type, dimension AS stream_type, sum(state_value) AS total_bytes
+    FROM
+      system.introspection_state_log
+    WHERE
+      (_tp_time > (now() - 5m)) AND state_name = 'checkpoint_storage_size'
+    GROUP BY
+      database, name, uuid, dimension
+  )
+SELECT
+    database, name, uuid, store_type, stream_type, total_bytes
+FROM
+    storage_sizes
+ORDER BY
+    total_bytes DESC
+SETTINGS query_mode = 'table'
 COMMENT 'version 2';
 ```