alter stress: automatically recover from bad state

Author: strtgbb

alter/stress/tests/actions.py

@@ -529,7 +529,7 @@ def delete_random_rows(self):
 def delete_random_rows_lightweight(self):
     """
     Lightweight delete a few rows at random.
-    Not supported with projections!
+    Not supported with projections.
     """
     table_name = get_random_table_name()
     node = get_random_node_for_table(table_name=table_name)
@@ -744,7 +744,9 @@ def drop_random_index(self):
                         node = get_random_node_for_table(table_name=table_name)
 
                     with And("waiting for any other mutations on that index to finish"):
-                        wait_for_mutations_to_finish(node=node, command_like=index_name, timeout=300)
+                        wait_for_mutations_to_finish(
+                            node=node, command_like=index_name, timeout=300
+                        )
 
                     with And("dropping the index"):
                         r = node.query(
@@ -793,84 +795,127 @@ def remove_random_ttl(self):
 
 
 @TestStep(Then)
-def check_tables_have_same_columns(self, tables):
+def check_tables_have_same_columns(self, tables, return_outliers=False):
     """
     Asserts that the given tables have the same columns.
     Smartly selects a node for each given replicated table.
     Does not check that all replicas of a table agree.
+
+    Args:
+        tables: List of table names to check.
+        return_outliers: If True, return the set of columns that are not present in all tables.
     """
-    with When("I get the columns for each table"):
-        table_columns = {}
-        for table_name in tables:
-            node = get_random_node_for_table(table_name=table_name)
-            table_columns[table_name] = set(
-                get_column_names(node=node, table_name=table_name)
-            )
+    try:
+        with When("I get the columns for each table"):
+            table_columns = {}
+            for table_name in tables:
+                node = get_random_node_for_table(table_name=table_name)
+                table_columns[table_name] = set(
+                    get_column_names(node=node, table_name=table_name)
+                )
 
-    with Then("all tables should have the same columns"):
-        for table1, table2 in combinations(tables, 2):
-            with By(f"checking {table1} and {table2}", flags=TE):
-                assert table_columns[table1] == table_columns[table2], error()
+        with Then("all tables should have the same columns"):
+            for table1, table2 in combinations(tables, 2):
+                with By(f"checking {table1} and {table2}", flags=TE):
+                    assert table_columns[table1] == table_columns[table2], error()
+    finally:
+        if return_outliers:
+            return set.union(*table_columns.values()) - set.intersection(
+                *table_columns.values()
+            )
+    return set()
 
 
 @TestStep(Then)
 def check_tables_have_same_projections(
-    self, tables, check_present: list = None, check_absent: list = None
+    self,
+    tables,
+    check_present: list = None,
+    check_absent: list = None,
+    return_outliers=False,
 ):
     """
     Asserts that the given tables have the same projections.
     Smartly selects a node for each given replicated table.
     Does not check that all replicas of a table agree.
-    """
-    with When("I get the projections for each table"):
-        table_projections = {}
-        for table_name in tables:
-            node = get_random_node_for_table(table_name=table_name)
-            wait_for_mutations_to_finish(node=node, command_like="PROJECTION")
-            table_projections[table_name] = set(
-                get_projections(node=node, table_name=table_name)
-            )
 
-    with Then("all tables should have the same projections"):
-        for table1, table2 in combinations(tables, 2):
-            with By(f"checking {table1} and {table2}", flags=TE):
-                assert table_projections[table1] == table_projections[table2], error()
-
-    if check_present is not None:
-        with And(f"I check that {check_present} exist"):
-            for projection_name in check_present:
-                assert projection_name in table_projections[tables[0]], error()
+    Args:
+        tables: List of table names to check.
+        check_present: List of projection names that should be present in all tables.
+        check_absent: List of projection names that should not be present in any tables.
+        return_outliers: If True, return the set of projections that are not present in all tables.
+    """
+    try:
+        with When("I get the projections for each table"):
+            table_projections = {}
+            for table_name in tables:
+                node = get_random_node_for_table(table_name=table_name)
+                wait_for_mutations_to_finish(node=node, command_like="PROJECTION")
+                table_projections[table_name] = set(
+                    get_projections(node=node, table_name=table_name)
+                )
 
-    if check_absent is not None:
-        with And(f"I check that {check_absent} do not exist"):
-            for projection_name in check_absent:
-                assert projection_name not in table_projections[tables[0]], error()
+        with Then("all tables should have the same projections"):
+            for table1, table2 in combinations(tables, 2):
+                with By(f"checking {table1} and {table2}", flags=TE):
+                    assert (
+                        table_projections[table1] == table_projections[table2]
+                    ), error()
+
+        if check_present is not None:
+            with And(f"I check that {check_present} exist"):
+                for projection_name in check_present:
+                    assert projection_name in table_projections[tables[0]], error()
+
+        if check_absent is not None:
+            with And(f"I check that {check_absent} do not exist"):
+                for projection_name in check_absent:
+                    assert projection_name not in table_projections[tables[0]], error()
+    finally:
+        if return_outliers:
+            return set.union(*table_projections.values()) - set.intersection(
+                *table_projections.values()
+            )
+    return set()
 
 
 @TestStep(Then)
-def check_tables_have_same_indexes(self, tables):
+def check_tables_have_same_indexes(self, tables, return_outliers=False):
     """
     Asserts that the given tables have the same indexes.
     Smartly selects a node for each given replicated table.
     Does not check that all replicas of a table agree.
+
+    Args:
+        tables: List of table names to check.
+        return_outliers: If True, return the set of indexes that are not present in all tables.
     """
-    with When("I get the indexes for each table"):
-        table_indexes = {}
-        for table_name in tables:
-            node = get_random_node_for_table(table_name=table_name)
-            table_indexes[table_name] = set(
-                get_indexes(node=node, table_name=table_name)
-            )
+    try:
+        with When("I get the indexes for each table"):
+            table_indexes = {}
+            for table_name in tables:
+                node = get_random_node_for_table(table_name=table_name)
+                table_indexes[table_name] = set(
+                    get_indexes(node=node, table_name=table_name)
+                )
 
-    with Then("all tables should have the same indexes"):
-        for table1, table2 in combinations(tables, 2):
-            with By(f"checking {table1} and {table2}", flags=TE):
-                assert table_indexes[table1] == table_indexes[table2], error()
+        with Then("all tables should have the same indexes"):
+            for table1, table2 in combinations(tables, 2):
+                with By(f"checking {table1} and {table2}", flags=TE):
+                    assert table_indexes[table1] == table_indexes[table2], error()
+    finally:
+        if return_outliers:
+            return set.union(*table_indexes.values()) - set.intersection(
+                *table_indexes.values()
+            )
+    return set()
 
 
 @TestStep(Then)
 @Retry(timeout=step_retry_timeout, delay=step_retry_delay)
-def check_consistency(self, tables=None, sync_timeout=None):
+def check_consistency(
+    self, tables=None, sync_timeout=None, restore_consistent_structure=False
+):
     """
     Check that the given tables hold the same amount of data on all nodes where they exist.
     Also check that column names match, subsequent part move tests require matching columns.
@@ -933,10 +978,37 @@ def check_consistency(self, tables=None, sync_timeout=None):
     # The below check also asserts that all tables have the same structure
     # Part move actions require matching structure
 
-    with Then("check that table structures are in sync"):
-        check_tables_have_same_columns(tables=tables)
-        check_tables_have_same_projections(tables=tables)
-        check_tables_have_same_indexes(tables=tables)
+    try:
+        with Then("check that table structures are in sync", flags=TE):
+            outlier_columns = check_tables_have_same_columns(
+                tables=tables, return_outliers=restore_consistent_structure
+            )
+            outlier_projections = check_tables_have_same_projections(
+                tables=tables, return_outliers=restore_consistent_structure
+            )
+            outlier_indexes = check_tables_have_same_indexes(
+                tables=tables, return_outliers=restore_consistent_structure
+            )
+    finally:
+        if restore_consistent_structure and any(
+            [outlier_columns, outlier_projections, outlier_indexes]
+        ):
+            with Finally("I clean up any inconsistencies between tables"):
+                for table_name in tables:
+                    node = get_random_node_for_table(table_name=table_name)
+
+                    for projection in outlier_projections:
+                        node.query(
+                            f"ALTER TABLE {table_name} DROP PROJECTION IF EXISTS {projection}"
+                        )
+                    for index in outlier_indexes:
+                        node.query(
+                            f"ALTER TABLE {table_name} DROP INDEX IF EXISTS {index}"
+                        )
+                    for column in outlier_columns:
+                        node.query(
+                            f"ALTER TABLE {table_name} DROP COLUMN IF EXISTS {column}"
+                        )
 
 
 @TestStep

alter/stress/tests/steps.py

@@ -221,7 +221,8 @@ def log_failing_mutations(self, nodes=None):
 
         with And("double checking the failed mutations"):
             r = node.query(
-                "SELECT latest_failed_part, table, latest_fail_reason FROM system.mutations WHERE is_done=0 FORMAT JSONCompactColumns"
+                "SELECT latest_failed_part, table, latest_fail_reason FROM system.mutations WHERE is_done=0 FORMAT JSONCompactColumns",
+                no_checks=True,
             )
             for part, table, fail_reason in json.loads(r.output):
                 if fail_reason == "":
@@ -232,6 +233,7 @@ def log_failing_mutations(self, nodes=None):
                 )
                 r = node.query(
                     f"SELECT * FROM system.parts WHERE name='{part}' and table='{table}' FORMAT Vertical",
+                    no_checks=True,
                 )
                 if r.output.strip():
                     note(f"State of {part}:\n{r.output.strip()}")
@@ -240,6 +242,7 @@ def log_failing_mutations(self, nodes=None):
                     column = re.search(r"column (.+):", fail_reason).group(1)
                     r = node.query(
                         f"SELECT * FROM system.parts_columns WHERE name='{part}' and table='{table}' and column='{column}' FORMAT Vertical",
+                        no_checks=True,
                     )
                     if r.output.strip():
                         note(f"State of {column}:\n{r.output.strip()}")

alter/stress/tests/stress_alter.py

@@ -129,6 +129,8 @@ def alter_combinations(
     insert_keeper_fault_injection_probability=0,
     network_impairment=False,
     limit_disk_space=False,
+    enforce_table_structure=None,
+    kill_stuck_mutations=None,
 ):
     """
     Perform combinations of alter actions, checking that all replicas agree.
@@ -150,6 +152,11 @@ def alter_combinations(
             limit_disk_space
         ), "enable limit_disk_space when using fill_disks to avoid unexpected behavior"
 
+    if enforce_table_structure is None:
+        enforce_table_structure = self.flags & TE
+    if kill_stuck_mutations is None:
+        kill_stuck_mutations = self.flags & TE
+
     action_groups = build_action_groups(
         actions=actions,
         combination_size=combination_size,
@@ -192,7 +199,9 @@ def alter_combinations(
                 if modify_random_ttl in actions
                 else None
             )
-            table_settings = "min_bytes_for_wide_part=0" if self.context.wide_parts_only else None
+            table_settings = (
+                "min_bytes_for_wide_part=0" if self.context.wide_parts_only else None
+            )
 
             for i in range(n_tables):
                 table_name = f"table{i}_{self.context.storage_policy}"
@@ -231,31 +240,50 @@ def alter_combinations(
                 title += "," + net_mode.name
 
             with Check(title):
-                if network_impairment:
-                    with Given("a network impairment"):
-                        impaired_network(network_mode=net_mode)
-
-                with When("I perform a group of actions"):
-                    for action in chain(background_actions, chosen_actions):
-                        By(
-                            f"I {action.name}",
-                            run=action,
-                            parallel=run_groups_in_parallel,
-                            flags=TE | ERROR_NOT_COUNTED,
-                        )
-
-                    for table in self.context.table_names:
-                        By(
-                            f"I OPTIMIZE {table}",
-                            test=optimize_random,
-                            parallel=run_optimize_in_parallel,
-                            flags=TE,
-                        )(table_name=table_name)
-
-                    join()
-
-                with Then("I check that the replicas are consistent", flags=TE):
-                    check_consistency()
+                try:
+                    if network_impairment:
+                        with Given("a network impairment"):
+                            impaired_network(network_mode=net_mode)
+
+                    with When("I perform a group of actions"):
+                        for action in chain(background_actions, chosen_actions):
+                            By(
+                                f"I {action.name}",
+                                run=action,
+                                parallel=run_groups_in_parallel,
+                                flags=TE | ERROR_NOT_COUNTED,
+                            )
+
+                        for table in self.context.table_names:
+                            By(
+                                f"I OPTIMIZE {table}",
+                                test=optimize_random,
+                                parallel=run_optimize_in_parallel,
+                                flags=TE,
+                            )(table_name=table_name)
+
+                        join()
+
+                finally:
+                    with Then("I make sure that the replicas are consistent", flags=TE):
+                        if kill_stuck_mutations:
+                            with By("killing any failing mutations"):
+                                for node in self.context.ch_nodes:
+                                    node.query(
+                                        "SELECT * FROM system.mutations WHERE is_done=0 AND latest_fail_reason != '' FORMAT Vertical",
+                                        no_checks=True,
+                                    )
+                                    r = node.query(
+                                        "KILL MUTATION WHERE latest_fail_reason != ''"
+                                    )
+                                    assert r.output == "", error(
+                                        "An erroring mutation was killed"
+                                    )
+
+                        with By("making sure that replicas agree"):
+                            check_consistency(
+                                restore_consistent_structure=enforce_table_structure
+                            )
 
             note(f"Average time per test combination {(time.time()-t)/(i+1):.1f}s")
 
@@ -429,7 +457,7 @@ def feature(self):
     # https://github.com/ClickHouse/ClickHouse/issues/62459
     self.context.disallow_move_partition_to_self = True
 
-    #https://github.com/ClickHouse/ClickHouse/issues/63545#issuecomment-2105013462
+    # https://github.com/ClickHouse/ClickHouse/issues/63545#issuecomment-2105013462
     self.context.wide_parts_only = True
 
     with Given("I have S3 disks configured"):