Bug#121045 Server telemetry metrics backed by aggregated_stats_buffer always report 0

mysql-test/suite/perfschema/r/telemetry_metrics_shard_bug.result

@@ -0,0 +1,43 @@
+INSTALL COMPONENT 'file://component_test_server_telemetry_metrics';
+UPDATE performance_schema.setup_meters SET enabled = 'YES';
+SET SESSION tmp_table_size = 1024;
+SET SESSION max_heap_table_size = 16384;
+SET SESSION long_query_time = 0;
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT, c VARCHAR(60), KEY kb (b));
+CREATE TABLE t2 (x INT, y INT);
+FLUSH STATUS;
+# Drive each affected metric family:
+SELECT * FROM t1 WHERE c > '';
+SELECT COUNT(*) FROM t1 a1 JOIN t1 a2 ON a1.c = a2.c;
+SELECT * FROM t1 WHERE a BETWEEN 5 AND 400;
+SELECT * FROM t1 ORDER BY c;
+SELECT * FROM t1 ORDER BY a DESC LIMIT 5;
+SELECT MIN(a), MAX(a) FROM t1;
+SELECT a FROM t1 ORDER BY b LIMIT 400;
+SELECT b, COUNT(*) FROM t1 GROUP BY b;
+SELECT c, COUNT(*) FROM t1 GROUP BY c;
+SELECT * FROM t1 WHERE a = 42;
+INSERT INTO t1 VALUES (9001,1,'x'),(9002,2,'y');
+UPDATE t1 SET c = 'u' WHERE a = 9001;
+DELETE FROM t1 WHERE a = 9002;
+PASS: mysql.stats.select_scan populated
+PASS: mysql.stats.select_full_join populated
+PASS: mysql.stats.select_range populated
+PASS: mysql.stats.sort_scan populated
+PASS: mysql.stats.sort_rows populated
+PASS: mysql.stats.created.tmp_tables populated
+PASS: mysql.stats.created.tmp_disk_tables populated
+PASS: mysql.stats.slow_queries populated
+PASS: mysql.stats.handler.write populated
+PASS: mysql.stats.handler.update populated
+PASS: mysql.stats.handler.delete populated
+PASS: mysql.stats.handler.read_key populated
+PASS: mysql.stats.handler.read_next populated
+PASS: mysql.stats.handler.read_prev populated
+PASS: mysql.stats.handler.read_first populated
+PASS: mysql.stats.handler.read_last populated
+PASS: mysql.stats.handler.read_rnd populated
+PASS: mysql.stats.handler.read_rnd_next populated
+PASS: mysql.stats.handler.external_lock populated
+DROP TABLE t1, t2;
+UNINSTALL COMPONENT 'file://component_test_server_telemetry_metrics';

mysql-test/suite/perfschema/t/telemetry_metrics_shard_bug-master.opt

@@ -0,0 +1 @@
+$TEST_SERVER_TELEMETRY_METRICS_COMPONENT_OPT

mysql-test/suite/perfschema/t/telemetry_metrics_shard_bug.test

@@ -0,0 +1,116 @@
+# Regression test for the server telemetry metrics backed by
+# aggregated_stats_buffer (mysql.stats, mysql.stats.handler).
+#
+# These metrics are read by the OTEL metric callback from the sharded
+# aggregated_stats_buffer. The producers must write that shard next to the
+# per-THD status_var store; if they do not, the metric exports a constant 0
+# while SHOW GLOBAL STATUS shows real activity.
+#
+# This test drives activity exercising the Select, Sort, Created_tmp and
+# Handler (read and write) families, then asserts each covered metric is
+# populated (nonzero). It FAILS on the unfixed server (metrics stuck at 0) and
+# PASSES once the shard writes are in place.
+#
+# Coverage: 19 of the 25 fixed counters are asserted. The remaining 6 are
+# intentionally NOT asserted because they cannot be driven nonzero
+# deterministically and would make the test flaky:
+#   - max_execution_time_exceeded : timing dependent (query must exceed a wall
+#     clock limit)
+#   - mrr_init                     : depends on the optimizer choosing MRR
+#   - select_range_check           : depends on "range checked for each record"
+#   - select_full_range_join       : could not be driven nonzero reliably
+#   - sort_range                   : could not be driven nonzero reliably
+#   - sort_merge_passes            : depends on sort-buffer spill thresholds
+# All 25 flow through the same two mechanisms (THD::inc_status_* and
+# handler::ha_statistic_increment); the 19 asserted here exercise both in both
+# directions, so a broken shard write in either mechanism is caught.
+#
+# Values are not asserted to equal SHOW GLOBAL STATUS exactly: the metric
+# aggregates 64 shards lock-free and is an approximate, non-snapshot sum by
+# design. The regression property is "populated and tracking", i.e. nonzero.
+
+--source include/have_server_telemetry_metrics.inc
+
+INSTALL COMPONENT 'file://component_test_server_telemetry_metrics';
+
+# All meters on so the metrics are exported/readable.
+UPDATE performance_schema.setup_meters SET enabled = 'YES';
+
+# Force internal tmp tables to spill to disk so created.tmp_disk_tables fires,
+# and make every query "slow" so slow_queries (long_query_count) fires.
+SET SESSION tmp_table_size = 1024;
+SET SESSION max_heap_table_size = 16384;
+SET SESSION long_query_time = 0;
+
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT, c VARCHAR(60), KEY kb (b));
+CREATE TABLE t2 (x INT, y INT);
+--disable_query_log
+let $i = 1;
+while ($i <= 500)
+{
+  eval INSERT INTO t1 VALUES ($i, $i % 17, CONCAT('padding_padding_padding_row_', $i));
+  eval INSERT INTO t2 VALUES ($i % 13, $i);
+  inc $i;
+}
+--enable_query_log
+FLUSH STATUS;
+
+--echo # Drive each affected metric family:
+--disable_result_log
+# Full scan on an unindexed column: select_scan + per-row read_rnd_next.
+SELECT * FROM t1 WHERE c > '';
+# Self join without a usable index: select_full_join.
+SELECT COUNT(*) FROM t1 a1 JOIN t1 a2 ON a1.c = a2.c;
+# Primary key range scan: select_range + read_next.
+SELECT * FROM t1 WHERE a BETWEEN 5 AND 400;
+# ORDER BY on an unindexed column: filesort -> sort_scan / sort_rows.
+SELECT * FROM t1 ORDER BY c;
+# Index descending / min-max: read_prev / read_last / read_first.
+SELECT * FROM t1 ORDER BY a DESC LIMIT 5;
+SELECT MIN(a), MAX(a) FROM t1;
+# Filesort that fetches rows by position: read_rnd.
+SELECT a FROM t1 ORDER BY b LIMIT 400;
+# GROUP BY on unindexed columns: created.tmp_tables (+ disk spill on the wide
+# VARCHAR grouping).
+SELECT b, COUNT(*) FROM t1 GROUP BY b;
+SELECT c, COUNT(*) FROM t1 GROUP BY c;
+# Keyed lookup: read_key.
+SELECT * FROM t1 WHERE a = 42;
+# Write path: write / update / delete (and external_lock on every open).
+INSERT INTO t1 VALUES (9001,1,'x'),(9002,2,'y');
+UPDATE t1 SET c = 'u' WHERE a = 9001;
+DELETE FROM t1 WHERE a = 9002;
+--enable_result_log
+
+# Assert each covered metric is populated (> 0). Each was a constant 0 on the
+# unfixed server. The UDF returns -1 for an unknown meter/metric, so assert
+# strictly > 0 (a bare truthiness check would let both 0 and -1 through).
+#
+# The 19 covered metrics are checked by looping over a "meter:metric" list:
+#   - mysql.stats (inc_status_* producers): the Select/Sort/Created_tmp/slow
+#     families.
+#   - mysql.stats.handler (ha_statistic_increment): the read and write handler
+#     families.
+--let $metrics = mysql.stats:select_scan mysql.stats:select_full_join mysql.stats:select_range mysql.stats:sort_scan mysql.stats:sort_rows mysql.stats:created.tmp_tables mysql.stats:created.tmp_disk_tables mysql.stats:slow_queries mysql.stats.handler:write mysql.stats.handler:update mysql.stats.handler:delete mysql.stats.handler:read_key mysql.stats.handler:read_next mysql.stats.handler:read_prev mysql.stats.handler:read_first mysql.stats.handler:read_last mysql.stats.handler:read_rnd mysql.stats.handler:read_rnd_next mysql.stats.handler:external_lock
+
+while ($metrics != '')
+{
+  # Pop the first "meter:metric" token off the front of the list.
+  --let $pair = `SELECT SUBSTRING_INDEX('$metrics', ' ', 1)`
+  --let $metrics = `SELECT LTRIM(SUBSTRING('$metrics', LENGTH('$pair') + 1))`
+  --let $meter = `SELECT SUBSTRING_INDEX('$pair', ':', 1)`
+  --let $metric = `SELECT SUBSTRING_INDEX('$pair', ':', -1)`
+
+  --let $val = `SELECT test_report_single_metric('$meter','$metric')`
+  if (`SELECT $val <= 0`)
+  {
+    --die FAIL: $meter.$metric not populated (got $val)
+  }
+  --echo PASS: $meter.$metric populated
+}
+
+# Cleanup
+DROP TABLE t1, t2;
+UNINSTALL COMPONENT 'file://component_test_server_telemetry_metrics';
+--let $MYSQLD_DATADIR= `select @@datadir`
+--remove_file $MYSQLD_DATADIR/test_server_telemetry_metrics_component.log

sql/handler.cc

@@ -2720,8 +2720,14 @@ handler *handler::clone(const char *name, MEM_ROOT *mem_root) {
 }
 
 void handler::ha_statistic_increment(
-    ulonglong System_status_var::*offset) const {
-  if (table && table->in_use) (table->in_use->status_var.*offset)++;
+    ulonglong System_status_var::*offset,
+    std::atomic_uint64_t aggregated_stats_buffer::*shard_offset) const {
+  if (table && table->in_use) {
+    (table->in_use->status_var.*offset)++;
+    (global_aggregated_stats.get_shard(table->in_use->thread_id()).*
+     shard_offset)
+        .fetch_add(1, std::memory_order_relaxed);
+  }
 }
 
 THD *handler::ha_thd() const {
@@ -3572,7 +3578,8 @@ int handler::ha_read_first_row(uchar *buf, uint primary_key) {
   int error;
   DBUG_TRACE;
 
-  ha_statistic_increment(&System_status_var::ha_read_first_count);
+  ha_statistic_increment(&System_status_var::ha_read_first_count,
+                         &aggregated_stats_buffer::ha_read_first_count);
 
   /*
     If there is very few deleted rows in the table, find the first row by
@@ -6654,6 +6661,8 @@ int DsMrr_impl::dsmrr_init(RANGE_SEQ_IF *seq_funcs, void *seq_init_param,
   if (is_mrr_assoc) {
     assert(!thd->status_var_aggregated);
     table->in_use->status_var.ha_multi_range_read_init_count++;
+    global_aggregated_stats.get_shard(table->in_use->thread_id())
+        .ha_multi_range_read_init_count++;
   }
 
   rowids_buf_end = buf->buffer_end;
@@ -8023,7 +8032,8 @@ int handler::ha_external_lock(THD *thd, int lock_type) {
   /* SQL HANDLER call locks/unlock while scanning (RND/INDEX). */
   assert(inited == NONE || table->open_by_handler);
 
-  ha_statistic_increment(&System_status_var::ha_external_lock_count);
+  ha_statistic_increment(&System_status_var::ha_external_lock_count,
+                         &aggregated_stats_buffer::ha_external_lock_count);
 
   MYSQL_TABLE_LOCK_WAIT(PSI_TABLE_EXTERNAL_LOCK, lock_type,
                         { error = external_lock(thd, lock_type); })

sql/handler.h

@@ -34,6 +34,7 @@
 #include <sys/types.h>
 #include <time.h>
 #include <algorithm>
+#include <atomic>
 #include <bitset>
 #include <functional>
 #include <map>
@@ -91,6 +92,7 @@ class THD;
 class handler;
 class partition_info;
 struct System_status_var;
+struct aggregated_stats_buffer;
 
 namespace dd {
 class Properties;
@@ -6597,7 +6599,9 @@ class handler {
 
  protected:
   /* Service methods for use by storage engines. */
-  void ha_statistic_increment(ulonglong System_status_var::*offset) const;
+  void ha_statistic_increment(
+      ulonglong System_status_var::*offset,
+      std::atomic_uint64_t aggregated_stats_buffer::*shard_offset) const;
   THD *ha_thd() const;
 
   /**

sql/log.cc

@@ -1803,7 +1803,10 @@ bool log_slow_applicable(THD *thd) {
       (thd->get_examined_row_count() >= thd->variables.min_examined_row_limit);
 
   // The docs say slow queries must be counted even when the log is off.
-  if (log_this_query) thd->status_var.long_query_count++;
+  if (log_this_query) {
+    thd->status_var.long_query_count++;
+    global_aggregated_stats.get_shard(thd->thread_id()).long_query_count++;
+  }
 
   /*
     Do not log administrative statements unless the appropriate option is

sql/sql_class.cc

@@ -1609,6 +1609,8 @@ void THD::awake(THD::killed_state state_to_set) {
   if (state_to_set == THD::KILL_TIMEOUT) {
     assert(!status_var_aggregated);
     status_var.max_execution_time_exceeded++;
+    global_aggregated_stats.get_shard(thread_id())
+        .max_execution_time_exceeded++;
   }
 
   /* Broadcast a condition to kick the target if it is waiting on it. */
@@ -2428,6 +2430,7 @@ void THD::inc_examined_row_count(ha_rows count) {
 void THD::inc_status_created_tmp_disk_tables() {
   assert(!status_var_aggregated);
   status_var.created_tmp_disk_tables++;
+  global_aggregated_stats.get_shard(thread_id()).created_tmp_disk_tables++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_created_tmp_disk_tables)(m_statement_psi, 1);
 #endif
@@ -2436,6 +2439,7 @@ void THD::inc_status_created_tmp_disk_tables() {
 void THD::inc_status_created_tmp_tables() {
   assert(!status_var_aggregated);
   status_var.created_tmp_tables++;
+  global_aggregated_stats.get_shard(thread_id()).created_tmp_tables++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_created_tmp_tables)(m_statement_psi, 1);
 #endif
@@ -2444,6 +2448,7 @@ void THD::inc_status_created_tmp_tables() {
 void THD::inc_status_select_full_join() {
   assert(!status_var_aggregated);
   status_var.select_full_join_count++;
+  global_aggregated_stats.get_shard(thread_id()).select_full_join_count++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_select_full_join)(m_statement_psi, 1);
 #endif
@@ -2452,6 +2457,7 @@ void THD::inc_status_select_full_join() {
 void THD::inc_status_select_full_range_join() {
   assert(!status_var_aggregated);
   status_var.select_full_range_join_count++;
+  global_aggregated_stats.get_shard(thread_id()).select_full_range_join_count++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_select_full_range_join)(m_statement_psi, 1);
 #endif
@@ -2460,6 +2466,7 @@ void THD::inc_status_select_full_range_join() {
 void THD::inc_status_select_range() {
   assert(!status_var_aggregated);
   status_var.select_range_count++;
+  global_aggregated_stats.get_shard(thread_id()).select_range_count++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_select_range)(m_statement_psi, 1);
 #endif
@@ -2468,6 +2475,7 @@ void THD::inc_status_select_range() {
 void THD::inc_status_select_range_check() {
   assert(!status_var_aggregated);
   status_var.select_range_check_count++;
+  global_aggregated_stats.get_shard(thread_id()).select_range_check_count++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_select_range_check)(m_statement_psi, 1);
 #endif
@@ -2476,6 +2484,7 @@ void THD::inc_status_select_range_check() {
 void THD::inc_status_select_scan() {
   assert(!status_var_aggregated);
   status_var.select_scan_count++;
+  global_aggregated_stats.get_shard(thread_id()).select_scan_count++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_select_scan)(m_statement_psi, 1);
 #endif
@@ -2484,6 +2493,7 @@ void THD::inc_status_select_scan() {
 void THD::inc_status_sort_merge_passes() {
   assert(!status_var_aggregated);
   status_var.filesort_merge_passes++;
+  global_aggregated_stats.get_shard(thread_id()).filesort_merge_passes++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_sort_merge_passes)(m_statement_psi, 1);
 #endif
@@ -2492,6 +2502,7 @@ void THD::inc_status_sort_merge_passes() {
 void THD::inc_status_sort_range() {
   assert(!status_var_aggregated);
   status_var.filesort_range_count++;
+  global_aggregated_stats.get_shard(thread_id()).filesort_range_count++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_sort_range)(m_statement_psi, 1);
 #endif
@@ -2500,6 +2511,7 @@ void THD::inc_status_sort_range() {
 void THD::inc_status_sort_rows(ha_rows count) {
   assert(!status_var_aggregated);
   status_var.filesort_rows += count;
+  global_aggregated_stats.get_shard(thread_id()).filesort_rows += count;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_sort_rows)
   (m_statement_psi, static_cast<ulong>(count));
@@ -2509,6 +2521,7 @@ void THD::inc_status_sort_rows(ha_rows count) {
 void THD::inc_status_sort_scan() {
   assert(!status_var_aggregated);
   status_var.filesort_scan_count++;
+  global_aggregated_stats.get_shard(thread_id()).filesort_scan_count++;
 #ifdef HAVE_PSI_STATEMENT_INTERFACE
   PSI_STATEMENT_CALL(inc_statement_sort_scan)(m_statement_psi, 1);
 #endif

storage/archive/ha_archive.cc

@@ -42,6 +42,7 @@
 #include "mysql/psi/mysql_file.h"
 #include "mysql/psi/mysql_memory.h"
 #include "nulls.h"
+#include "sql/aggregated_stats_buffer.h"
 #include "sql/derror.h"
 #include "sql/field.h"
 #include "sql/sql_class.h"
@@ -792,7 +793,8 @@ int ha_archive::write_row(uchar *buf) {
 
   if (share->crashed) return HA_ERR_CRASHED_ON_USAGE;
 
-  ha_statistic_increment(&System_status_var::ha_write_count);
+  ha_statistic_increment(&System_status_var::ha_write_count,
+                         &aggregated_stats_buffer::ha_write_count);
   mysql_mutex_lock(&share->mutex);
 
   if (!share->archive_write_open && share->init_archive_writer()) {
@@ -1117,7 +1119,8 @@ int ha_archive::rnd_next(uchar *buf) {
   }
   scan_rows--;
 
-  ha_statistic_increment(&System_status_var::ha_read_rnd_next_count);
+  ha_statistic_increment(&System_status_var::ha_read_rnd_next_count,
+                         &aggregated_stats_buffer::ha_read_rnd_next_count);
   current_position = aztell(&archive);
   rc = get_row(&archive, buf);
 
@@ -1145,7 +1148,8 @@ void ha_archive::position(const uchar *) {
 int ha_archive::rnd_pos(uchar *buf, uchar *pos) {
   int rc;
   DBUG_TRACE;
-  ha_statistic_increment(&System_status_var::ha_read_rnd_next_count);
+  ha_statistic_increment(&System_status_var::ha_read_rnd_next_count,
+                         &aggregated_stats_buffer::ha_read_rnd_next_count);
   current_position = (my_off_t)my_get_ptr(pos, ref_length);
   if (azseek(&archive, current_position, SEEK_SET) == (my_off_t)(-1L)) {
     rc = HA_ERR_CRASHED_ON_USAGE;