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

@@ -28,6 +28,10 @@ REGISTER_TUNABLE("abort_during_downgrade_if_scs_dont_stop", "Abort if scs don't
                  "after starting a downgrade (default OFF)", TUNABLE_BOOLEAN,
                  &gbl_abort_during_downgrade_if_scs_dont_stop, 0, NULL, NULL,
                  NULL, NULL);
+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 = 1;
 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

@@ -591,13 +591,14 @@ int do_alter_table(struct ireq *iq, struct schema_change_type *s,
 
     /* 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)) {
+    /*if (init_sc_genids(newdb, s)) {
         sc_errf(s, "failed initilizing sc_genids\n");
         delete_temp_table(iq, newdb);
         change_schemas_recover(s->tablename);
         decrement_sc_yet_to_resume_counter();
         return -1;
-    }
+    }*/
+    // todo: detect if this is alter or merge and init genids accordingly.
 
     Pthread_rwlock_wrlock(&db->sc_live_lk);
     db->sc_from = s->db = db;
@@ -654,7 +655,10 @@ int do_alter_table(struct ireq *iq, struct schema_change_type *s,
         changed == SC_CONSTRAINT_CHANGE) {
         if (!s->live)
             gbl_readonly_sc = 1;
-        rc = convert_all_records(db, newdb, newdb->sc_genids, s);
+
+        unsigned long long sc_genids[MAXDTASTRIPE];
+        memset(sc_genids, 0, MAXDTASTRIPE);
+        rc = convert_all_records(db, newdb, sc_genids, s);
         if (rc == 1) rc = 0;
     } else
         rc = 0;
@@ -775,13 +779,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 +791,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 +813,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 +827,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);
@@ -866,7 +871,7 @@ static int do_merge_table(struct ireq *iq, struct schema_change_type *s,
 
         for (i = 0; i < gbl_dtastripe; i++) {
             sc_errf(s, "  > [%s] stripe %2d was at 0x%016llx\n", s->tablename,
-                    i, newdb->sc_genids[i]);
+                    i, sc_genids[i]);
         }
 
         while (s->logical_livesc) {
@@ -885,6 +890,14 @@ static int do_merge_table(struct ireq *iq, struct schema_change_type *s,
     /* handle renaming sqlite_stat1 entries for idx */
     check_for_idx_rename(s->newdb, s->db);
 
+    // All shards point to the same newdb.
+    //
+    // By setting it to NULL here, we ensure that
+    // all shards except for the first shard (which does not run
+    // do_merge_table) have newdb set to NULL.
+    // This avoids double-freeing it.
+    if (rc == SC_MASTER_DOWNGRADE) { s->newdb = NULL; }
+
     return SC_OK;
 }
 

schemachange/sc_logic.c

@@ -47,7 +47,7 @@ void comdb2_cheapstack_sym(FILE *f, char *fmt, ...);
 
 extern int gbl_is_physical_replicant;
 
-int gbl_multitable_ddl = 0;
+int gbl_multitable_ddl = 1;
 
 /**** Utility functions */
 

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

@@ -1,60 +1,112 @@
 #!/usr/bin/env bash
 
+set -ex
+dbnm=$1
+tier="local"
+CDB2SQL_EXE="cdb2sql"
+DBNAME=$dbnm
+
 source ${TESTSROOTDIR}/tools/runit_common.sh
 source ${TESTSROOTDIR}/tools/cluster_utils.sh
+source util.sh
 
-set -ex
+declare -r CRASH=1
+declare -r DOWNGRADE=2
 
-[ -z "${CLUSTER}" ] && { echo "Test requires a cluster"; exit 0; }
+test_driver() {
+	# Given
+	(
+		local -r num_shards=$1 num_records_per_shard=$2 resume_trigger=$3
 
-dbnm=$1
+		local master
+		master=`cdb2sql --tabs ${CDB2_OPTIONS} $dbnm ${tier} 'SELECT host FROM comdb2_cluster WHERE is_master="Y"'`
 
-restart_cluster() {
-	set +e
-	for node in ${CLUSTER} ; do
-		kill_restart_node ${node} &
-	done
-	set -e
+		local -r tbl_name="t"
+		setup_table ${tbl_name} ${num_shards} ${num_records_per_shard}
+
+		trap "cdb2sql ${CDB2_OPTIONS} ${dbnm} ${tier} 'drop table ${tbl_name}'" EXIT
+
+	#	cdb2sql $dbnm --host $master "EXEC PROCEDURE sys.cmd.send('convert_record_sleep 1')"
+
+		# When
+		COMDB2_CONFIG_MIN_RETRIES=0 \
+		COMDB2_CONFIG_MAX_RETRIES=0 \
+		cdb2sql ${CDB2_OPTIONS} ${dbnm} ${tier} "alter table ${tbl_name} partitioned by none" &
+		local -r waitpid=$!
+
+		sleep 5
+
+		if (( resume_trigger == CRASH )); then
+			restart_cluster ${master}
+		elif (( resume_trigger == DOWNGRADE )); then
+			downgrade ${master}
+		else
+			echo "FAIL: expected resume_trigger to be one of: KILL, DOWNGRADE"
+			return 1
+		fi
+
+		# Then
+		if wait ${waitpid};
+		then
+			echo "FAIL: Merge succeeded before cluster bounced. Test is buggy."
+			return 1
+		fi
 
-	sleep 2
+		wait_for_outstanding_scs &> /dev/null
 
-	wait_for_cluster
+		local timepart
+		timepart=$(cdb2sql --tabs ${CDB2_OPTIONS} ${dbnm} ${tier} "select * from comdb2_timepartitions where name='${tbl_name}'")
+		if [[ -n ${timepart} ]];
+		then
+			echo "FAIL: Found time partition that should not exist"
+			return 1
+		fi
+
+		local num_records
+		num_records=$(cdb2sql --tabs ${CDB2_OPTIONS} ${dbnm} ${tier} "select count(*) from ${tbl_name}")
+		if (( num_records != num_shards*num_records_per_shard ));
+		then
+			echo "FAIL: Unexpected number of records after partition merge"
+			return 1
+		fi
+	)
 }
 
-test_partition_merge_resume() {
-	# Given
-	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)"
-
-	# When
-	COMDB2_CONFIG_MIN_RETRIES=1 cdb2sql ${CDB2_OPTIONS} ${dbnm} default "alter table t partitioned by none" &
-	waitpid=$!
-	sleep 1
-	restart_cluster &> /dev/null
-
-	# Then
-	if wait ${waitpid};
-	then
-		echo "FAIL: Merge succeeded before cluster bounced. Test is buggy."
-		return 1
-	fi
-
-	local timepart
-	timepart=$(cdb2sql --tabs ${CDB2_OPTIONS} ${dbnm} default "select * from comdb2_timepartitions where name='t'")
-	if [[ -z ${timepart} ]];
-	then
-		echo "FAIL: Could not find expected time partition"
-		return 1
-	fi
-
-	# Cleanup
-	cdb2sql ${CDB2_OPTIONS} ${dbnm} default "drop table t"
+test_resume_tiny_sc_after_master_downgrade() {
+	local -r num_shards=5 num_records_per_shard=1 resume_trigger=${DOWNGRADE}
+	test_driver ${num_shards} ${num_records_per_shard} ${resume_trigger}
+}
+
+test_resume_tiny_sc_after_master_crash() {
+	local -r num_shards=5 num_records_per_shard=1 resume_trigger=${CRASH}
+	test_driver ${num_shards} ${num_records_per_shard} ${resume_trigger}
+}
+
+test_resume_big_sc_after_master_downgrade() {
+	local -r num_shards=5 num_records_per_shard=10000 resume_trigger=${DOWNGRADE}
+	test_driver ${num_shards} ${num_records_per_shard} ${resume_trigger}
+}
+
+test_resume_big_sc_after_master_crash() {
+	local -r num_shards=5 num_records_per_shard=10000 resume_trigger=${CRASH}
+	test_driver ${num_shards} ${num_records_per_shard} ${resume_trigger}
+}
+
+
+main() {
+	# local -r tests=$(compgen -A function | grep -oh "test_\w*")
+	tests=test_resume_big_sc_after_master_downgrade
+
+	for testcase in ${tests};
+	do
+		echo "Running ${testcase} ----------"
+		if ${testcase}; then
+			echo "Passed ${testcase} ----------"
+		else
+			echo "Failed ${testcase} ----------"
+			return 1
+		fi
+	done
 }
 
-test_partition_merge_resume
+main