{172878933}: Converting to ez-systable

This patch converts comdb2_tables, comdb2_columns, comdb2_keys and comdb2_keycomponents
to use the ez-systable wrapper, so that they do not hold views_lk for too long in
a stored procedure, which can block a time partition rollout (Thanks, Dorin!)

Signed-off-by: Rivers Zhang <hzhang320@bloomberg.net>

db/sql.h

@@ -264,7 +264,6 @@ typedef struct {
     int maxchunksize;     /* multi-transaction bulk mode */
     int crtchunksize;     /* how many rows are processed already */
     int nchunks;          /* number of chunks. 0 for a non-chunked transaction. */
-    int views_lk_held;    /* Am I holding views_lk? */
 } dbtran_type;
 typedef dbtran_type trans_t;
 
@@ -590,8 +589,7 @@ enum prepare_flags {
     PREPARE_NO_NORMALIZE = 32,
     PREPARE_ONLY = 64,
     PREPARE_ALLOW_TEMP_DDL = 128,
-    PREPARE_ACQUIRE_SPLOCK = 256,
-    PREPARE_ACQUIRE_VIEWSLK = 512
+    PREPARE_ACQUIRE_SPLOCK = 256
 };
 
 /* This structure is designed to hold several pieces of data related to

db/sqlglue.c

@@ -7642,11 +7642,6 @@ static int sqlite3LockStmtTables_int(sqlite3_stmt *pStmt, int after_recovery)
         return 0;
     }
 
-    if ((p->vTableFlags & PREPARE_ACQUIRE_VIEWSLK) && !clnt->dbtran.views_lk_held) {
-        Pthread_rwlock_rdlock(&views_lk);
-        clnt->dbtran.views_lk_held = 1;
-    }
-
     for (int i = 0; i < p->numVTableLocks; i++) {
         if ((rc = bdb_lock_tablename_read_fromlid(thedb->bdb_env, p->vTableLocks[i],
                                                   bdb_get_lid_from_cursortran(clnt->dbtran.cursor_tran))) != 0) {
@@ -9692,11 +9687,6 @@ int put_curtran_flags(bdb_state_type *bdb_state, struct sqlclntstate *clnt,
     rc = bdb_put_cursortran(bdb_state, clnt->dbtran.cursor_tran, curtran_flags,
                             &bdberr);
 
-    if (clnt->dbtran.views_lk_held) {
-        Pthread_rwlock_unlock(&views_lk);
-        clnt->dbtran.views_lk_held = 0;
-    }
-
     if (rc) {
         logmsg(LOGMSG_DEBUG, "%s: %p rc %d bdberror %d\n", __func__, (void *)pthread_self(), rc, bdberr);
         ctrace("%s: rc %d bdberror %d\n", __func__, rc, bdberr);

db/sqlinterfaces.c

@@ -771,18 +771,6 @@ static void record_locked_vtable(struct sql_authorizer_state *pAuthState, const
     if (table != NULL && (strcmp(table, "comdb2_triggers") == 0)) {
         pAuthState->flags |= PREPARE_ACQUIRE_SPLOCK;
     }
-    if (table != NULL && (strncasecmp(table, "comdb2_", 7) == 0)) {
-        table += 7;
-        if (strcasecmp(table, "tables") == 0 ||
-            strcasecmp(table, "columns") == 0 ||
-            strcasecmp(table, "keys") == 0 ||
-            strcasecmp(table, "keycomponents") == 0 ||
-            strcasecmp(table, "timepartitions") == 0 ||
-            strcasecmp(table, "timepartshards") == 0 ||
-            strcasecmp(table, "timepartevents") == 0) {
-            pAuthState->flags |= PREPARE_ACQUIRE_VIEWSLK;
-        }
-    }
     if (vtable_lock && !vtable_search(pAuthState->vTableLocks, pAuthState->numVTableLocks, vtable_lock)) {
         pAuthState->vTableLocks =
             (char **)realloc(pAuthState->vTableLocks, sizeof(char *) * (pAuthState->numVTableLocks + 1));

db/views_systable.c

@@ -25,6 +25,7 @@ int timepart_systable_timepartitions_collect(void **data, int *nrecords)
     int rc = 0;
     uuidstr_t us;
 
+    Pthread_rwlock_rdlock(&views_lk);
     arr = calloc(views->nviews, sizeof(systable_timepartition_t));
     if (!arr) {
         logmsg(LOGMSG_ERROR, "%s OOM %zu!\n", __func__,
@@ -55,6 +56,7 @@ int timepart_systable_timepartitions_collect(void **data, int *nrecords)
         }
     }
 done:
+    Pthread_rwlock_unlock(&views_lk);
     *data = arr;
     *nrecords = narr;
     return rc;
@@ -88,6 +90,8 @@ int timepart_systable_timepartshards_collect(void **data, int *nrecords)
     int nview;
     int rc = 0;
 
+    Pthread_rwlock_rdlock(&views_lk);
+
     narr = 0;
     for (nview = 0; nview < views->nviews; nview++) {
         narr += views->views[nview]->nshards;
@@ -120,6 +124,7 @@ int timepart_systable_timepartshards_collect(void **data, int *nrecords)
         }
     }
 done:
+    Pthread_rwlock_unlock(&views_lk);
     *data = arr;
     *nrecords = narr;
     return rc;