{180685083} feat(reqlog): post schema change activity

Statreqs tell us what the database is busy doing. If any schema changes are
running, that is significant and worth reporting in statreqs.

Schema change stats are logged in a dedicated section that looks like this:

```text
11/29 02:55:00:   SCHEMA CHANGE STATS
11/29 02:55:00:       table 't' records converted 269805 diff 169469 rate 84734 r/s adds 0 updates 0 deletes 77
```

Signed-off-by: Shoumyo Chakravorti <[email protected]>

Author: SChakravorti21

db/comdb2.c

@@ -141,6 +141,7 @@ void berk_memp_sync_alarm_ms(int);
 #include <net_appsock.h>
 #include "sc_csc2.h"
 #include "sc_util.h"
+#include "sc_global.h"
 #include "reverse_conn.h"
 #include "alias.h"
 #include "str_util.h" /* QUOTE */
@@ -4626,8 +4627,6 @@ int throttle_lim = 10000;
 int cpu_throttle_threshold = 100000;
 
 double gbl_cpupercent;
-#include <sc_util.h>
-
 
 static inline void log_tbl_item(int64_t curr, int64_t *prev, const char *(*type_to_str)(int), int type, char *string,
                                 int *hdr_p, struct reqlogger *statlogger, dbtable *tbl, int first) 
@@ -4729,6 +4728,8 @@ void *statthd(void *p)
     int have_scon_stats = 0;
     int64_t rw_evicts;
 
+    int last_num_sc = 0;
+    struct running_sc_info *last_schema_changes = NULL;
 
     thrman_register(THRTYPE_GENERIC);
     thread_started("statthd");
@@ -4902,7 +4903,9 @@ void *statthd(void *p)
             }
 
             int aa_include_updates = bdb_attr_get(thedb->bdb_attr, BDB_ATTR_AA_COUNT_UPD);
+
             rdlock_schema_lk();
+
             for (ii = 0; ii < dbenv->num_dbs; ++ii) {
                 dbtable *tbl = dbenv->dbs[ii];
                 int hdr = 0;
@@ -4949,6 +4952,50 @@ void *statthd(void *p)
                 log_tbl_item(tbl->deadlock_count, &tbl->saved_deadlock_count, NULL, 0, "deadlock count", &hdr,
                              statlogger, tbl, 0);
             }
+
+            int num_sc = 0;
+            struct running_sc_info *schema_changes = NULL;
+
+            if (list_running_schema_changes(&schema_changes, &num_sc) == 0) {
+                if (num_sc > 0) {
+                    /* log header */
+                    reqlog_logf(statlogger, REQL_INFO, "SCHEMA CHANGE STATS\n");
+                }
+
+                for (ii = 0; ii < num_sc; ii++) {
+                    struct running_sc_info sc = schema_changes[ii];
+                    struct running_sc_info *last_sc = NULL;
+
+                    for (jj = 0; jj < last_num_sc; jj++) {
+                        if (strcmp(last_schema_changes[jj].table_name, sc.table_name) == 0) {
+                            last_sc = &last_schema_changes[jj];
+                            break;
+                        }
+                    }
+
+                    uint64_t last_nrecs = last_sc ? last_sc->nrecs : 0;
+                    uint32_t last_adds = last_sc ? last_sc->adds : 0;
+                    uint32_t last_updates = last_sc ? last_sc->updates : 0;
+                    uint32_t last_deletes = last_sc ? last_sc->deletes : 0;
+
+                    uint64_t diff_nrecs = sc.nrecs - last_nrecs;
+
+                    reqlog_logf(
+                        statlogger, REQL_INFO,
+                        "    table '%s' records converted %ld diff %ld rate %ld r/s adds %d updates %d deletes %d\n",
+                        sc.table_name,                             //
+                        sc.nrecs, diff_nrecs, diff_nrecs / thresh, //
+                        sc.adds - last_adds,                       //
+                        sc.updates - last_updates,                 //
+                        sc.deletes - last_deletes);
+                }
+
+                free(last_schema_changes);
+
+                last_num_sc = num_sc;
+                last_schema_changes = schema_changes;
+            }
+
             unlock_schema_lk();
 
             pstats = bdb_get_process_stats();

schemachange/sc_global.c

@@ -697,3 +697,49 @@ void sc_alter_latency(int counter)
         }
     }
 }
+
+int list_running_schema_changes(struct running_sc_info **info, int *num_running_sc)
+{
+    unsigned int bucket;
+    void *entry;
+    sc_table_t *sc_table = NULL;
+    *num_running_sc = 0;
+
+    Pthread_mutex_lock(&schema_change_in_progress_mutex);
+
+    if (!sc_tables) {
+        Pthread_mutex_unlock(&schema_change_in_progress_mutex);
+        return 1;
+    }
+
+    /* preallocate outparam */
+    int unused;
+    int num_entries = 0;
+    hash_info(sc_tables, &num_entries, &unused, &unused, &unused, &unused, &unused, &unused);
+    *info = realloc(*info, sizeof(struct running_sc_info) * num_entries);
+
+    for (sc_table = hash_first(sc_tables, &entry, &bucket); sc_table;
+         sc_table = hash_next(sc_tables, &entry, &bucket)) {
+
+        /* get_dbtable_by_name looks at thedb->db_hash, so caller must
+         * hold the schema lock */
+        struct dbtable *db = get_dbtable_by_name(sc_table->tablename);
+
+        Pthread_rwlock_rdlock(&db->sc_live_lk);
+
+        if (db->doing_conversion || db->doing_upgrade) {
+            struct running_sc_info *sc_info = &((*info)[*num_running_sc]);
+            sc_info->table_name = strdup(sc_table->tablename);
+            sc_info->nrecs = db->sc_nrecs;
+            sc_info->adds = db->sc_adds;
+            sc_info->updates = db->sc_updates;
+            sc_info->deletes = db->sc_deletes;
+            ++(*num_running_sc);
+        }
+
+        Pthread_rwlock_unlock(&db->sc_live_lk);
+    }
+
+    Pthread_mutex_unlock(&schema_change_in_progress_mutex);
+    return 0;
+}

schemachange/sc_global.h

@@ -17,6 +17,10 @@
 #ifndef INCLUDE_SC_GLOBAL_H
 #define INCLUDE_SC_GLOBAL_H
 
+#include <stdint.h>
+
+#include "schemachange.h"
+
 extern pthread_mutex_t schema_change_in_progress_mutex;
 extern pthread_mutex_t fastinit_in_progress_mutex;
 extern pthread_mutex_t schema_change_sbuf2_lock;
@@ -60,6 +64,14 @@ extern int rep_sync_save;
 extern int log_sync_save;
 extern int log_sync_time_save;
 
+struct running_sc_info {
+    const char *table_name;
+    uint64_t nrecs;   /* num records converted (does not include `adds`) */
+    uint32_t adds;    /* num added in front of sc cursor */
+    uint32_t updates; /* num updated behind sc cursor */
+    uint32_t deletes; /* num deleted behind sc cursor */
+};
+
 int is_dta_being_rebuilt(struct scplan *plan);
 const char *get_sc_to_name(const char *);
 void wait_for_sc_to_stop(const char *operation, const char *func, int line);
@@ -87,5 +99,8 @@ struct schema_change_type *preempt_ongoing_alter(char *table, int action);
 void clear_ongoing_alter();
 int get_stopsc(const char *func, int line);
 void sc_alter_latency(int counter);
+/* List all running schema changes and their progress. Caller must acquire schema
+ * lock in read mode. */
+int list_running_schema_changes(struct running_sc_info **info, int *num_running_sc);
 
 #endif

schemachange/sc_records.c

@@ -140,16 +140,20 @@ static inline int print_aggregate_sc_stat(struct convert_record_data *data,
     /* totals across all threads */
     if (data->scanmode != SCAN_PARALLEL) return 1;
 
-    long long total_nrecs_diff =
-        data->from->sc_nrecs - data->from->sc_prev_nrecs;
+    uint64_t total_nrecs_diff = data->from->sc_nrecs - data->from->sc_prev_nrecs;
     data->from->sc_prev_nrecs = data->from->sc_nrecs;
+
+    /* actual progress made after accounting for updates and deletes that
+     * happened behind the schema change cursor (by definition, all adds happen
+     * in front of the schema change cursor) */
+    uint64_t actual_nrecs = data->from->sc_nrecs - (data->from->sc_updates + data->from->sc_deletes);
+
     sc_printf(data->s,
-              "[%s] progress TOTAL %lld +%lld actual "
-              "progress total %lld rate %lld r/s\n",
-              data->from->tablename, data->from->sc_nrecs, total_nrecs_diff,
-              data->from->sc_nrecs -
-                  (data->from->sc_adds + data->from->sc_updates),
-              total_nrecs_diff / sc_report_freq);
+              "[%s] progress TOTAL %lld +%lld rate %lld r/s "
+              "actual progress total %lld\n",
+              data->from->tablename, data->from->sc_nrecs, total_nrecs_diff, total_nrecs_diff / sc_report_freq,
+              actual_nrecs);
+
     return 1;
 }