Fix bugs in partition merge async resume

Signed-off-by: Morgan Douglas <[email protected]>

Author: morgando

cdb2api/cdb2api.c

@@ -492,6 +492,10 @@ static void do_init_once(void)
     if (min_retries) {
         MIN_RETRIES = atoi(min_retries);
     }
+    char *max_retries = getenv("COMDB2_CONFIG_MAX_RETRIES");
+    if (max_retries) {
+        MAX_RETRIES = atoi(max_retries);
+    }
 }
 
 /* if sqlstr is a read stmt will return 1 otherwise return 0

db/db_tunables.c

@@ -41,6 +41,7 @@
 /* Separator for composite tunable components. */
 #define COMPOSITE_TUNABLE_SEP '.'
 
+extern int gbl_enable_partitioned_table_merge_resume;
 extern int gbl_bulk_import_validation_werror;
 extern int gbl_debug_sleep_during_bulk_import;
 extern int gbl_waitalive_iterations;

db/db_tunables.h

@@ -24,6 +24,10 @@
   at multiple places.
 */
 
+REGISTER_TUNABLE("partitioned_table_merge_resume", "Toggle the ability to resume "
+                 "partitioned table collapse SCs (default OFF)", TUNABLE_BOOLEAN,
+                 &gbl_enable_partitioned_table_merge_resume, 0, NULL, NULL,
+                 NULL, NULL);
 REGISTER_TUNABLE("abort_on_in_use_rqid", NULL, TUNABLE_BOOLEAN,
                  &gbl_abort_on_clear_inuse_rqid, READONLY | NOARG, NULL, NULL,
                  NULL, NULL);

db/osqlcomm.c

@@ -73,6 +73,7 @@ extern int g_osql_ready;
 extern int gbl_goslow;
 extern int gbl_partial_indexes;
 
+int gbl_enable_partitioned_table_merge_resume = 0;
 int gbl_master_sends_query_effects = 1;
 int gbl_toblock_random_deadlock_trans;
 int gbl_toblock_random_verify_error;
@@ -6312,6 +6313,7 @@ static int start_schema_change_tran_wrapper_merge(const char *tblname,
     alter_sc->force_rebuild = 1; /* we are moving rows here */
     /* alter only in parallel mode for live */
     alter_sc->scanmode = SCAN_PARALLEL;
+    alter_sc->resume = sc->resume;
     /* link the sc */
     iq->sc = alter_sc;
 
@@ -6328,7 +6330,7 @@ static int start_schema_change_tran_wrapper_merge(const char *tblname,
     /* link the alter */
     iq->sc->sc_next = iq->sc_pending;
     iq->sc_pending = iq->sc;
-    iq->sc->newdb = NULL; /* lose ownership, otherwise double free */
+    if (alter_sc->nothrevent) { iq->sc->newdb = NULL; /* lose ownership, otherwise double free */ }
 
     if (arg->lockless) {
         *pview = timepart_reaquire_view(arg->part_name);
@@ -6407,15 +6409,16 @@ static int _process_partitioned_table_merge(struct ireq *iq)
     struct dbtable *first_shard = get_dbtable_by_name(first_shard_name);
     free(first_shard_name);
 
-    /* we need to move data */
-    sc->force_rebuild = 1;
+    const int latched_nothrevent = sc->nothrevent;
+
+    sc->force_rebuild = 1; /* we need to move data */
+    sc->nothrevent = 1; /* we need do_add_table / do_alter_table to run first */
+    sc->finalize = 0;
 
     if (!first_shard->sqlaliasname) {
         /*
          * create a table with the same name as the partition
          */
-        sc->nothrevent = 1; /* we need do_add_table to run first */
-        sc->finalize = 0;   /* make sure */
         sc->kind = SC_ADDTABLE;
 
         rc = start_schema_change_tran(iq, NULL);
@@ -6430,10 +6433,17 @@ static int _process_partitioned_table_merge(struct ireq *iq)
         iq->sc_pending = iq->sc;
     } else {
         /*
-         * use the fast shard as the destination, after first altering it
+         * use the first shard as the destination, after first altering it
          */
-        sc->nothrevent = 1; /* we need do_alter_table to run first */
-        sc->finalize = 0;
+        if (gbl_enable_partitioned_table_merge_resume) {
+            sc->partition.type = PARTITION_NONE;
+        } else {
+            assert(sc->partition.type == PARTITION_MERGE);
+            // If partitioned table merge resumes are disabled,
+            // then we keep the type equal to PARTITION_MERGE.
+            // There is code later on that blocks the resume if
+            // the type is PARTITION_MERGE.
+        }
 
         strncpy(sc->tablename, first_shard->tablename, sizeof(sc->tablename));
 
@@ -6456,6 +6466,7 @@ static int _process_partitioned_table_merge(struct ireq *iq)
     /* at this point we have created the future btree, launch an alter
      * for each of the shards of the partition
      */
+    sc->nothrevent = latched_nothrevent;
     arg.s = sc;
     arg.s->iq = iq;
     arg.part_name = strdup(sc->tablename);  /*sc->tablename gets rewritten*/

schemachange/sc_alter_table.c

@@ -775,13 +775,6 @@ static int do_merge_table(struct ireq *iq, struct schema_change_type *s,
         return -1;
     }
 
-    /* set sc_genids, 0 them if we are starting a new schema change, or
-     * restore them to their previous values if we are resuming */
-    if (init_sc_genids(newdb, s)) {
-        sc_client_error(s, "Failed to initialize sc_genids");
-        return -1;
-    }
-
     Pthread_rwlock_wrlock(&db->sc_live_lk);
     db->sc_from = s->db = db;
     db->sc_to = s->newdb = newdb;
@@ -794,6 +787,9 @@ static int do_merge_table(struct ireq *iq, struct schema_change_type *s,
     assert(db->sc_from == db && s->db == db);
     assert(db->sc_to == newdb && s->newdb == newdb);
     assert(db->doing_conversion == 1);
+    if (s->resume && IS_ALTERTABLE(s)) {
+        decrement_sc_yet_to_resume_counter();
+    }
     MEMORY_SYNC;
 
     if (get_stopsc(__func__, __LINE__)) {
@@ -813,8 +809,10 @@ static int do_merge_table(struct ireq *iq, struct schema_change_type *s,
     struct schema *tag = find_tag_schema(newdb, ".NEW..ONDISK");
     if (!tag) {
         struct errstat err = {0};
+        Pthread_mutex_lock(&csc2_subsystem_mtx);
         rc =
             populate_db_with_alt_schema(thedb, newdb, newdb->csc2_schema, &err);
+        Pthread_mutex_unlock(&csc2_subsystem_mtx);
         if (rc) {
             logmsg(LOGMSG_ERROR, "%s\ncsc2: \"%s\"\n", err.errstr,
                    newdb->csc2_schema);
@@ -825,9 +823,12 @@ static int do_merge_table(struct ireq *iq, struct schema_change_type *s,
 
     add_ongoing_alter(s);
 
+    unsigned long long sc_genids[MAXDTASTRIPE];
+    memset(sc_genids, 0, MAXDTASTRIPE);
+
     /* skip converting records for fastinit and planned schema change
      * that doesn't require rebuilding anything. */
-    rc = convert_all_records(db, newdb, newdb->sc_genids, s);
+    rc = convert_all_records(db, newdb, sc_genids, s);
     if (rc == 1) rc = 0;
 
     remove_ongoing_alter(s);

tests/TODO

@@ -83,6 +83,5 @@ rowlocks_blkseq.test
 sc_async_constraints.test
 sigstopcluster.test
 weighted_standing_queue.test -- failing in rhel8 + podman
-sc_resume_partition.test
 
 # vim: set sw=4 ts=4 et:

tests/sc_resume_partition.test/lrl.options

@@ -0,0 +1,2 @@
+multitable_ddl 1
+partitioned_table_merge_resume on

tests/sc_resume_partition.test/runit

@@ -5,38 +5,62 @@ source ${TESTSROOTDIR}/tools/cluster_utils.sh
 
 set -ex
 
-[ -z "${CLUSTER}" ] && { echo "Test requires a cluster"; exit 0; }
-
 dbnm=$1
 
 restart_cluster() {
+	local -r restart_delay_secs=10
+
 	set +e
-	for node in ${CLUSTER} ; do
-		kill_restart_node ${node} &
-	done
-	set -e
+	if [ -z $CLUSTER ];
+	then
+		kill_restart_node $1 &
+	else
+		for node in ${CLUSTER} ; do
+			kill_restart_node ${node} &
+		done
+	fi
 
 	sleep 2
-
 	wait_for_cluster
+	set -e
+}
+
+num_outstanding_scs() {
+	cdb2sql --tabs ${CDB2_OPTIONS} ${dbnm} default "select count(*) from comdb2_sc_status where status!='COMMITTED'"
+}
+
+wait_for_outstanding_scs() {
+	local num_scs
+
+	num_scs=$(num_outstanding_scs)
+	while (( num_scs > 0 )); do
+		sleep 1
+	done
 }
 
 test_partition_merge_resume() {
 	# Given
+	local -r num_shards=$1
 	local starttime
 	starttime=$(get_timestamp 120)
 	local master
 	master=`cdb2sql --tabs ${CDB2_OPTIONS} $dbnm default 'SELECT host FROM comdb2_cluster WHERE is_master="Y"'`
 
 	cdb2sql $dbnm --host $master "EXEC PROCEDURE sys.cmd.send('convert_record_sleep 1')"
-	cdb2sql ${CDB2_OPTIONS} ${dbnm} default "create table t(a int) partitioned by time period 'daily' retention 2 start '${starttime}'"
-	cdb2sql ${CDB2_OPTIONS} $dbnm default "insert into t values (1)"
+	cdb2sql ${CDB2_OPTIONS} ${dbnm} default "create table t(a int) partitioned by time period 'daily' retention ${num_shards} start '${starttime}'"
+
+	for i in $(seq 0 1 $((num_shards-1)));
+	do
+		local shard
+		shard=$(cdb2sql --tabs ${CDB2_OPTIONS} ${dbnm} default "select shardname from comdb2_timepartshards limit 1 offset ${i}")
+		cdb2sql ${CDB2_OPTIONS} ${dbnm} default "insert into '${shard}' values(1)"
+	done
 
 	# When
-	COMDB2_CONFIG_MIN_RETRIES=1 cdb2sql ${CDB2_OPTIONS} ${dbnm} default "alter table t partitioned by none" &
-	waitpid=$!
+	COMDB2_CONFIG_MIN_RETRIES=0 COMDB2_CONFIG_MAX_RETRIES=0 cdb2sql ${CDB2_OPTIONS} ${dbnm} default "alter table t partitioned by none" &
+	local -r waitpid=$!
 	sleep 1
-	restart_cluster &> /dev/null
+	restart_cluster $master &> /dev/null
 
 	# Then
 	if wait ${waitpid};
@@ -45,16 +69,31 @@ test_partition_merge_resume() {
 		return 1
 	fi
 
+	wait_for_outstanding_scs
+
 	local timepart
 	timepart=$(cdb2sql --tabs ${CDB2_OPTIONS} ${dbnm} default "select * from comdb2_timepartitions where name='t'")
-	if [[ -z ${timepart} ]];
+	if [[ -n ${timepart} ]];
 	then
-		echo "FAIL: Could not find expected time partition"
+		echo "FAIL: Found time partition that should not exist"
+		return 1
+	fi
+
+	local num_records
+	num_records=$(cdb2sql --tabs ${CDB2_OPTIONS} ${dbnm} default "select count(*) from t")
+	if (( num_records != num_shards ));
+	then
+		echo "FAIL: Unexpected number of records after partition merge"
 		return 1
 	fi
 
 	# Cleanup
 	cdb2sql ${CDB2_OPTIONS} ${dbnm} default "drop table t"
 }
 
-test_partition_merge_resume
+main() {
+	local -r num_shards=5
+	test_partition_merge_resume ${num_shards}
+}
+
+main