generate a create view query and populate sqlite engine with a view representing the generic shard

Signed-off-by: Aakash Arayambeth <[email protected]>

Author: Aakash Arayambeth

bdb/bdb_schemachange.c

@@ -190,7 +190,7 @@ int handle_scdone(DB_ENV *dbenv, u_int32_t rectype, llog_scdone_args *scdoneop,
     }
 
     if (sctype == rename_table || sctype == alias_table ||
-        (sctype == add && (strlen(table) + 1 < scdoneop->table.size))) {
+        ((sctype == add || sctype == gen_shard) && (strlen(table) + 1 < scdoneop->table.size))) {
         strarg = &table[strlen(table) + 1];
     }
 

db/comdb2.c

@@ -208,7 +208,7 @@ static void *purge_old_files_thread(void *arg);
 static int lrllinecmp(char *lrlline, char *cmpto);
 static void ttrap(struct timer_parm *parm);
 int clear_temp_tables(void);
-int gen_shard_deserialize_shard(uint32_t *numdbs, char ***dbnames, uint32_t *numcols, char ***columns, char ***shardnames, char *serializedStr);
+int gen_shard_deserialize_shard(char **genshard_name, uint32_t *numdbs, char ***dbnames, uint32_t *numcols, char ***columns, char ***shardnames, char *serializedStr);
 pthread_key_t comdb2_open_key;
 
 /*---GLOBAL SETTINGS---*/
@@ -250,7 +250,7 @@ int gbl_watchdog_watch_threshold = 60;
 int gbl_watchdog_disable_at_start = 0; /* don't enable watchdog on start */
 int gbl_nonames = 1;
 int gbl_reject_osql_mismatch = 1;
-int gbl_abort_on_clear_inuse_rqid = 1;
+int gbl_abort_on_clear_inuse_rqid = 0;
 int gbl_archive_on_init = 1;
 
 pthread_t gbl_invalid_tid; /* set this to our main threads tid */
@@ -2299,7 +2299,7 @@ int llmeta_load_genshards(struct dbenv *dbenv, void *tran) {
     int bdberr = 0;
     /* allow as many generic sharded tables as regular tables (?) */
     char *tablenames[MAX_NUM_TABLES];
-    char *shard_info = NULL;
+    char *shard_info = NULL, *genshard_name = NULL;
     char **dbnames = NULL, **columns = NULL, **shardnames = NULL;
     int table_count = 0, size = 0;
     uint32_t numdbs=0, numcols=0;
@@ -2322,14 +2322,15 @@ int llmeta_load_genshards(struct dbenv *dbenv, void *tran) {
             goto err;
         }
 
-        if (gen_shard_deserialize_shard(&numdbs, &dbnames, &numcols, &columns, &shardnames, shard_info)) {
+        if (gen_shard_deserialize_shard(&genshard_name, &numdbs, &dbnames, &numcols, &columns, &shardnames, shard_info)) {
             logmsg(LOGMSG_ERROR, "Failed to deserialize llmeta str for generic shard %s\n", tablenames[i]);
         }
 
         for(int i=0;i<numdbs;i++){
             db = get_dbtable_by_name(shardnames[i]);
             if (db) {
                 /*update the table object*/
+                db->genshard_name = genshard_name;
                 db->numdbs = numdbs;
                 db->dbnames = dbnames;
                 db->numcols = numcols;

db/comdb2.h

@@ -791,6 +791,7 @@ typedef struct dbtable {
     const char *timepartition_name;
 
     /* generic sharding metadata */
+    char *genshard_name;
     uint32_t numdbs;
     char **dbnames;
     uint32_t numcols;

db/gen_shard.c

@@ -15,6 +15,13 @@ int cson_extract_int(cson_object *cson_obj, const char *param,
                              struct errstat *err);
 cson_array *cson_extract_array(cson_object *cson_obj, const char *param,
                                        struct errstat *err);
+int views_sqlite_del_view(const char *view_name, sqlite3 *db,
+                           struct errstat *err);
+char *describe_row(const char *tblname, const char *prefix,
+                           enum views_trigger_op op_type, struct errstat *err);
+
+
+
 int gen_shard_serialize_shard(const char *tablename, uint32_t numdbs, char **dbnames, uint32_t numcols,
 									char **columns, char **shardnames, uint32_t *outLen, char **out)
 {
@@ -165,11 +172,11 @@ int gen_shard_llmeta_read(void *tran, const char *name, char **pstr)
     return rc;
 }
 
-int gen_shard_deserialize_shard(uint32_t *numdbs, char ***dbnames, uint32_t *numcols, char ***columns, char ***shardnames, char *serializedStr) {
+int gen_shard_deserialize_shard(char **genshard_name, uint32_t *numdbs, char ***dbnames, uint32_t *numcols, char ***columns, char ***shardnames, char *serializedStr) {
     cson_object *rootObj = NULL;
     cson_value *rootVal = NULL, *arrVal = NULL;
     cson_array *dbs_arr = NULL, *cols_arr = NULL, *shards_arr = NULL;
-    const char *tablename = NULL;
+    char *tablename = NULL;
     char **dbs = NULL, **cols = NULL, **shards = NULL;
     char *err_str = NULL;
     int num_dbs = 0, num_cols = 0;
@@ -186,7 +193,7 @@ int gen_shard_deserialize_shard(uint32_t *numdbs, char ***dbnames, uint32_t *num
     rc = cson_value_is_object(rootVal);
     rootObj = cson_value_get_object(rootVal);
 
-    tablename = cson_extract_str(rootObj, "TABLENAME", &err);
+    tablename = (char *)cson_extract_str(rootObj, "TABLENAME", &err);
     if (!tablename) {
         err_str = "INVALID CSON. Couldn't find 'TABLENAME' key";
         goto error;
@@ -262,7 +269,7 @@ int gen_shard_deserialize_shard(uint32_t *numdbs, char ***dbnames, uint32_t *num
         }
         shards[i] = strdup(cson_value_get_cstr(arrVal));
     }
-
+    *genshard_name = tablename;
     *numdbs = num_dbs;
     *dbnames = dbs;
     *numcols = num_cols;
@@ -308,21 +315,22 @@ int gen_shard_deserialize_shard(uint32_t *numdbs, char ***dbnames, uint32_t *num
 int gen_shard_update_inmem_db(void *tran, struct dbtable *db, const char *name) {
     char *serializedStr = NULL;
     uint32_t numdbs = 0, numcols = 0;
-    char **dbnames = NULL, **columns = NULL, **shardnames = NULL;
+    char **dbnames = NULL, **columns = NULL, **shardnames = NULL, *genshard_name = NULL;
     int rc = 0;
     rc = gen_shard_llmeta_read(tran, name, &serializedStr); 
     if (rc) {
         logmsg(LOGMSG_ERROR, "Failed to read from llmeta for table %s\n", name);
         goto done;
     }
 
-    rc = gen_shard_deserialize_shard(&numdbs, &dbnames, &numcols, &columns, &shardnames, serializedStr);
+    rc = gen_shard_deserialize_shard(&genshard_name,&numdbs, &dbnames, &numcols, &columns, &shardnames, serializedStr);
     if (rc) {
         logmsg(LOGMSG_ERROR, "Failed to deserialized llmeta str for table %s\n", name);
         goto done;
     }
 
     /*update the table object*/
+    db->genshard_name = genshard_name;
     db->numdbs = numdbs;
     db->dbnames = dbnames;
     db->numcols = numcols;
@@ -335,4 +343,159 @@ int gen_shard_update_inmem_db(void *tran, struct dbtable *db, const char *name)
     return rc;
 }
 
+char *gen_shard_create_view_query(struct dbtable *tbl, sqlite3 *db, struct errstat *err)
+{
+    char *select_str = NULL;
+    char *cols_str = NULL;
+    char *tmp_str = NULL;
+    char *ret_str = NULL;
+    int numshards = tbl->numdbs;
+    const char *viewname = tbl->genshard_name;
+    char **dbnames = tbl->dbnames;
+    char **shardnames = tbl->shardnames;
+    int i;
+    cols_str = sqlite3_mprintf("rowid as __hidden__rowid, ");
+    if (!cols_str) {
+        goto malloc;
+    }
+
+    cols_str = describe_row(tbl->tablename, cols_str, VIEWS_TRIGGER_QUERY, err);
+    if (!cols_str) {
+        /* preserve error, if any */
+        if (err->errval != VIEW_NOERR)
+            return NULL;
+        goto malloc;
+    } else {
+        logmsg(LOGMSG_USER, "GOT cols_str as %s\n", cols_str);
+    }
+
+    select_str = sqlite3_mprintf("");
+    i = 0;
+    logmsg(LOGMSG_USER, "num shards is : %d\n", numshards);
+    for(;i<numshards;i++){
+        tmp_str = sqlite3_mprintf("%s%sSELECT %s FROM %s.'\%s'", select_str, (i > 0) ? " UNION ALL " : "", cols_str,
+                                  dbnames[i], shardnames[i]);
+        sqlite3_free(select_str);
+        if (!tmp_str) {
+            sqlite3_free(cols_str);
+            goto malloc;
+        }
+        select_str = tmp_str;
+    }
+
+    ret_str = sqlite3_mprintf("CREATE VIEW %w AS %s", viewname, select_str);
+    if (!ret_str) {
+        sqlite3_free(select_str);
+        sqlite3_free(cols_str);
+        goto malloc;
+    }
+
+    sqlite3_free(select_str);
+    sqlite3_free(cols_str);
+
+    logmsg(LOGMSG_USER, "THE GENERATED VIEW QUERY IS %s\n", ret_str);
+    //dbg_verbose_sqlite("Generated:\n\"%s\"\n", ret_str);
+
+    return ret_str;
+
+malloc:
+    err->errval = VIEW_ERR_MALLOC;
+    snprintf(err->errstr, sizeof(err->errstr), "View %s out of memory\n", viewname);
+    return NULL;
+}
+
+int gen_shard_run_sql(sqlite3 *db, char *stmt, struct errstat *err)
+{
+    char *errstr = NULL;
+    int rc;
+
+    /* create the view */
+    rc = sqlite3_exec(db, stmt, NULL, NULL, &errstr);
+    if (rc != SQLITE_OK) {
+        err->errval = VIEW_ERR_BUG;
+        snprintf(err->errstr, sizeof(err->errstr), "Sqlite error \"%s\"", errstr);
+        /* can't control sqlite errors */
+        err->errstr[sizeof(err->errstr) - 1] = '\0';
+
+        logmsg(LOGMSG_USER, "%s: sqlite error \"%s\" sql \"%s\"\n", __func__, errstr, stmt);
+
+        if (errstr)
+            sqlite3_free(errstr);
+        return err->errval;
+    }
+
+    return VIEW_NOERR;
+}
+
+int gen_shard_add_view(struct dbtable *tbl, sqlite3 *db, struct errstat *err) 
+{
+    char *stmt_str;
+    int rc;
+
+    /* create the statement */
+    stmt_str = gen_shard_create_view_query(tbl, db, err);
+    if (!stmt_str) {
+        return err->errval;
+    }
+
+    rc = gen_shard_run_sql(db, stmt_str, err);
+
+    logmsg(LOGMSG_USER, "+++++++++++sql: %s, rc: %d\n", stmt_str, rc);
+    /* free the statement */
+    sqlite3_free(stmt_str);
+
+    if (rc != VIEW_NOERR) {
+        return err->errval;
+    }
+    return rc;
+}
+
+int gen_shard_delete_view(struct dbtable *tbl, sqlite3 *db, struct errstat *err) 
+{
+    return views_sqlite_del_view(tbl->genshard_name, db, err);
+}
+
+int gen_shard_update_sqlite(sqlite3 *db, struct errstat *err)
+{
+    Table *tab;
+    int rc;
+    for (int tbl_idx = 0; tbl_idx < thedb->num_dbs; ++tbl_idx) {
+        struct dbtable *tbl = thedb->dbs[tbl_idx];
+        if (tbl->genshard_name) {
+            /* this table is a component shard of a genshard table*/
+            tab = sqlite3FindTableCheckOnly(db, tbl->genshard_name, NULL);
+            if (tab) {
+                /* found view, is it the same version ? */
+                if (tbl->tableversion != tab->version) {
+                    /* older version, destroy current view */
+                    rc = gen_shard_delete_view(tbl, db, err);
+                    if (rc != VIEW_NOERR) {
+                        logmsg(LOGMSG_ERROR, "%s: failed to remove old view\n", __func__);
+                        goto done;
+                    }
+                } else {
+                    /* up to date, nothing to do */
+                    continue;
+                }
+            }
+            rc = gen_shard_add_view(tbl, db, err);
+            if (rc != VIEW_NOERR) {
+                goto done;
+            }
+        }
+    }
+    rc = VIEW_NOERR;
+done:
+    return rc;
+}
+
+int is_gen_shard(const char *tablename) {
+    for(int i=0;i<thedb->num_dbs;i++) {
+        struct dbtable *tbl = thedb->dbs[i];
+        if (tbl->genshard_name && strcmp(tbl->genshard_name, tablename)==0) {
+            return 1;
+        }
+    }
+    return 0;
+}
 

db/gen_shard.h

@@ -4,4 +4,8 @@ int gen_shard_llmeta_add(tran_type *tran, char *tablename, uint32_t numdbs, char
 		uint32_t numcols, char **columns, char **shardnames, struct errstat *);
 int gen_shard_update_inmem_db(void *tran, struct dbtable *db, const char *name);
 int gen_shard_llmeta_remove(tran_type *tran, char *tablename, struct errstat *err);
+int gen_shard_update_sqlite(sqlite3 *db, struct errstat *err);
+int gen_shard_delete_view(struct dbtable *tbl, sqlite3 *db, struct errstat *err);
+int gen_shard_add_view(struct dbtable *tbl, sqlite3 *db, struct errstat *err); 
+int is_gen_shard(const char *tablename);
 #endif

db/sqlinterfaces.c

@@ -113,7 +113,7 @@
 #include <net_appsock.h>
 #include <typessql.h>
 #include <sqlwriter.h>
-
+#include "gen_shard.h"
 /*
 ** WARNING: These enumeration values are not arbitrary.  They represent
 **          indexes into the array of meta-command names contained in
@@ -4300,6 +4300,12 @@ static int prepare_engine(struct sqlthdstate *thd, struct sqlclntstate *clnt,
                 }
             }
 
+            rc = gen_shard_update_sqlite(thd->sqldb, &xerr);
+            if (rc) {
+                logmsg(LOGMSG_FATAL, "Failed to create views for generic shards."
+                        "rc=%d errstr=%s\n", xerr.errval, xerr.errstr);
+                abort();
+            }
             /* save the views generation number */
             thd->views_gen = gbl_views_gen;
         }

db/tag.c

@@ -6154,6 +6154,7 @@ void freedb_int(dbtable *db, dbtable *replace)
     int i;
     int dbs_idx;
     char *sqlaliasname = db->sqlaliasname;
+    char *genshard_name = db->genshard_name;
     const char *timepartition_name = db->timepartition_name;
     char **dbnames = alloca(sizeof(char*) * db->numdbs);
     for (i = 0; i < db->numdbs; i++)
@@ -6167,6 +6168,7 @@ void freedb_int(dbtable *db, dbtable *replace)
 
     if (!replace) {
         free(sqlaliasname);
+        free(genshard_name);
         for (i = 0; i < db->numdbs; i++) {
             free(dbnames[i]);
         }
@@ -6242,6 +6244,7 @@ void freedb_int(dbtable *db, dbtable *replace)
         db->dbs_idx = dbs_idx;
         db->sqlaliasname = sqlaliasname;
         db->timepartition_name = timepartition_name;
+        db->genshard_name = genshard_name;
         for (i = 0; i < db->numdbs; i++)
             db->dbnames[i] = dbnames[i];
         for (i = 0; i < db->numcols; i++)

db/views.c

@@ -97,7 +97,7 @@ static int _views_do_op(timepart_views_t *views, const char *name,
                         int (*op)(timepart_views_t *, timepart_view_t *,
                                   struct errstat *),
                         struct errstat *err);
-static char *_describe_row(const char *tblname, const char *prefix,
+char *describe_row(const char *tblname, const char *prefix,
                            enum views_trigger_op op_type, struct errstat *err);
 static void *timepart_cron_kickoff(struct cron_event *_, struct errstat *err);
 static int _next_shard_exists(timepart_view_t *view, char *newShardName,
@@ -1439,7 +1439,7 @@ int timepart_dump_timepartitions(FILE *dest)
     return has_tp;
 }
 
-static char *_describe_row(const char *tblname, const char *prefix,
+char *describe_row(const char *tblname, const char *prefix,
                            enum views_trigger_op op_type, struct errstat *err)
 {
     struct dbtable *gdb;

db/views.h

@@ -217,7 +217,7 @@ int views_sqlite_add_view(timepart_view_t *view, sqlite3 *db,
  *
  *
  */
-int views_sqlite_del_view(timepart_view_t *view, sqlite3 *db,
+int views_sqlite_del_timepart_view(timepart_view_t *view, sqlite3 *db,
                           struct errstat *err);
 
 /**