The 'bootstrap' command now emits table-copy instructions

The original plan for 'bootstrap' was to do live alterations, that they should
only lock what they needed, however InnoDB likes to lock everything. So instead
we need to always assume that "bootstrapping" will be a live table clone,
and at their conclusion the team should perform an atomic rename.

Author: jcjones

partitionmanager/bootstrap.py

@@ -7,17 +7,18 @@
 import logging
 import operator
 import yaml
-
 from partitionmanager.types import (
     ChangePlannedPartition,
     MaxValuePartition,
     NewPlannedPartition,
+    Table,
 )
 from partitionmanager.table_append_partition import (
-    table_is_compatible,
+    generate_sql_reorganize_partition_commands,
+    get_columns,
     get_current_positions,
     get_partition_map,
-    generate_sql_reorganize_partition_commands,
+    table_is_compatible,
 )
 from .tools import iter_show_end
 
@@ -89,7 +90,7 @@ def _plan_partitions_for_time_offsets(
     """
     changes = list()
     for (i, offset), is_final in iter_show_end(enumerate(time_offsets)):
-        increase = [x * offset / RATE_UNIT for x in rate_of_change]
+        increase = [x * (offset / RATE_UNIT) for x in rate_of_change]
         predicted_positions = [
             int(p + i) for p, i in zip(ordered_current_pos, increase)
         ]
@@ -115,6 +116,83 @@ def _plan_partitions_for_time_offsets(
     return changes
 
 
+def _suffix(lines, *, indent="", mid_suffix="", final_suffix=""):
+    """ Helper that suffixes each line with either mid- or final- suffix """
+    for line, is_final in iter_show_end(lines):
+        if is_final:
+            yield indent + line + final_suffix
+        else:
+            yield indent + line + mid_suffix
+
+
+def _trigger_column_copies(cols):
+    """ Helper that returns lines copying each column for a trigger. """
+    for c in cols:
+        yield f"`{c}` = NEW.`{c}`"
+
+
+def _generate_sql_copy_commands(
+    existing_table, map_data, columns, new_table, alter_commands_iter
+):
+    """ Generate a series of SQL commands to start a copy of the existing_table
+    to a new_table, applying the supplied alterations before starting the
+    triggers. """
+    log = logging.getLogger(
+        f"_generate_sql_copy_commands:{existing_table.name} to {new_table.name}"
+    )
+
+    max_val_part = map_data["partitions"][-1]
+    if not isinstance(max_val_part, MaxValuePartition):
+        log.error(f"Expected a MaxValue partition, got {max_val_part}")
+        raise Exception("Unexpected part?")
+
+    range_id_string = ", ".join(map_data["range_cols"])
+
+    yield f"DROP TABLE IF EXISTS {new_table.name};"
+    yield f"CREATE TABLE {new_table.name} LIKE {existing_table.name};"
+    yield f"ALTER TABLE {new_table.name} REMOVE PARTITIONING;"
+    yield f"ALTER TABLE {new_table.name} PARTITION BY RANGE({range_id_string}) ("
+    yield f"\tPARTITION {max_val_part.name} VALUES LESS THAN MAXVALUE"
+    yield ");"
+
+    for command in alter_commands_iter:
+        yield command
+
+    cols = set(columns)
+
+    yield f"CREATE OR REPLACE TRIGGER copy_inserts_from_{existing_table.name}_to_{new_table.name}"
+    yield f"\tAFTER INSERT ON {existing_table.name} FOR EACH ROW"
+    yield f"\t\tINSERT INTO {new_table.name} SET"
+
+    for line in _suffix(
+        _trigger_column_copies(sorted(cols)),
+        indent="\t\t\t",
+        mid_suffix=",",
+        final_suffix=";",
+    ):
+        yield line
+
+    update_columns = cols.difference(set(map_data["range_cols"]))
+    if not update_columns:
+        log.info("No columns to copy, so no UPDATE trigger being constructed.")
+        return
+
+    yield f"CREATE OR REPLACE TRIGGER copy_updates_from_{existing_table.name}_to_{new_table.name}"
+    yield f"\tAFTER UPDATE ON {existing_table.name} FOR EACH ROW"
+    yield f"\t\tUPDATE {new_table.name} SET"
+
+    for line in _suffix(
+        _trigger_column_copies(sorted(update_columns)), indent="\t\t\t", mid_suffix=","
+    ):
+        yield line
+
+    yield "\t\tWHERE " + " AND ".join(
+        _trigger_column_copies(map_data["range_cols"])
+    ) + ";"
+
+    return
+
+
 def calculate_sql_alters_from_state_info(conf, in_fp):
     """
     Using the config and the input yaml file-like object, return the SQL
@@ -154,6 +232,8 @@ def calculate_sql_alters_from_state_info(conf, in_fp):
         else:
             map_data = _override_config_to_map_data(conf)
 
+        columns = [r["Field"] for r in get_columns(conf.dbcmd, table)]
+
         current_positions = get_current_positions(
             conf.dbcmd, table, map_data["range_cols"]
         )
@@ -194,8 +274,16 @@ def calculate_sql_alters_from_state_info(conf, in_fp):
             max_val_part,
         )
 
+        table_new = Table(f"{table.name}_new_{conf.curtime:%Y%m%d}")
+
+        alter_commands_iter = generate_sql_reorganize_partition_commands(
+            table_new, changes
+        )
+
         commands[table.name] = list(
-            generate_sql_reorganize_partition_commands(table, changes)
+            _generate_sql_copy_commands(
+                table, map_data, columns, table_new, alter_commands_iter
+            )
         )
 
     return commands

partitionmanager/bootstrap_test.py

@@ -4,7 +4,11 @@
 from datetime import datetime, timedelta
 
 from .bootstrap import (
+    _generate_sql_copy_commands,
     _get_time_offsets,
+    _suffix,
+    _trigger_column_copies,
+    _override_config_to_map_data,
     calculate_sql_alters_from_state_info,
     write_state_info,
 )
@@ -14,7 +18,8 @@
 
 class MockDatabase(DatabaseCommand):
     def __init__(self):
-        self.response = []
+        self._response = list()
+        self._select_response = [[{"id": 150}]]
         self.num_queries = 0
 
     def run(self, cmd):
@@ -36,9 +41,9 @@ def run(self, cmd):
             ]
 
         if "SELECT" in cmd:
-            return [{"id": 150}]
+            return self._select_response.pop()
 
-        return self.response
+        return self._response.pop()
 
     def db_name(self):
         return SqlInput("the-database")
@@ -82,6 +87,7 @@ def test_get_time_offsets(self):
         )
 
     def test_read_state_info(self):
+        self.maxDiff = None
         conf_past = Config()
         conf_past.curtime = datetime(2021, 3, 1)
         conf_past.dbcmd = MockDatabase()
@@ -97,6 +103,12 @@ def test_read_state_info(self):
         conf_now = Config()
         conf_now.curtime = datetime(2021, 3, 3)
         conf_now.dbcmd = MockDatabase()
+        conf_now.dbcmd._response = [
+            [
+                {"Field": "id", "Type": "bigint UNSIGNED"},
+                {"Field": "serial", "Type": "varchar"},
+            ]
+        ]
         conf_now.tables = [Table("test").set_partition_period(timedelta(days=30))]
 
         state_fs.seek(0)
@@ -105,10 +117,137 @@ def test_read_state_info(self):
             x,
             {
                 "test": [
-                    "ALTER TABLE `test` REORGANIZE PARTITION `p_start` INTO "
-                    "(PARTITION `p_20210303` VALUES LESS THAN (156), "
-                    "PARTITION `p_20210402` VALUES LESS THAN (2406), "
-                    "PARTITION `p_20210502` VALUES LESS THAN MAXVALUE);"
+                    "DROP TABLE IF EXISTS test_new_20210303;",
+                    "CREATE TABLE test_new_20210303 LIKE test;",
+                    "ALTER TABLE test_new_20210303 REMOVE PARTITIONING;",
+                    "ALTER TABLE test_new_20210303 PARTITION BY RANGE(id) (",
+                    "\tPARTITION p_start VALUES LESS THAN MAXVALUE",
+                    ");",
+                    "ALTER TABLE `test_new_20210303` REORGANIZE PARTITION `p_start` "
+                    + "INTO (PARTITION `p_20210303` VALUES LESS THAN (156), "
+                    + "PARTITION `p_20210402` VALUES LESS THAN (2406), PARTITION "
+                    + "`p_20210502` VALUES LESS THAN MAXVALUE);",
+                    "CREATE OR REPLACE TRIGGER copy_inserts_from_test_to_test_new_20210303",
+                    "\tAFTER INSERT ON test FOR EACH ROW",
+                    "\t\tINSERT INTO test_new_20210303 SET",
+                    "\t\t\t`id` = NEW.`id`,",
+                    "\t\t\t`serial` = NEW.`serial`;",
+                    "CREATE OR REPLACE TRIGGER copy_updates_from_test_to_test_new_20210303",
+                    "\tAFTER UPDATE ON test FOR EACH ROW",
+                    "\t\tUPDATE test_new_20210303 SET",
+                    "\t\t\t`serial` = NEW.`serial`",
+                    "\t\tWHERE `id` = NEW.`id`;",
                 ]
             },
         )
+
+    def test_read_state_info_map_table(self):
+        self.maxDiff = None
+        conf = Config()
+        conf.assume_partitioned_on = ["order", "auth"]
+        conf.curtime = datetime(2021, 3, 3)
+        conf.dbcmd = MockDatabase()
+        conf.dbcmd._select_response = [[{"auth": 22}], [{"order": 11}]]
+        conf.dbcmd._response = [
+            [
+                {"Field": "order", "Type": "bigint UNSIGNED"},
+                {"Field": "auth", "Type": "bigint UNSIGNED"},
+            ]
+        ]
+        conf.tables = [Table("map_table").set_partition_period(timedelta(days=30))]
+
+        state_fs = io.StringIO()
+        yaml.dump(
+            {
+                "tables": {"map_table": {"order": 11, "auth": 22}},
+                "time": (conf.curtime - timedelta(days=1)),
+            },
+            state_fs,
+        )
+        state_fs.seek(0)
+
+        x = calculate_sql_alters_from_state_info(conf, state_fs)
+        print(x)
+        self.assertEqual(
+            x,
+            {
+                "map_table": [
+                    "DROP TABLE IF EXISTS map_table_new_20210303;",
+                    "CREATE TABLE map_table_new_20210303 LIKE map_table;",
+                    "ALTER TABLE map_table_new_20210303 REMOVE PARTITIONING;",
+                    "ALTER TABLE map_table_new_20210303 PARTITION BY RANGE(order, auth) (",
+                    "\tPARTITION p_assumed VALUES LESS THAN MAXVALUE",
+                    ");",
+                    "ALTER TABLE `map_table_new_20210303` REORGANIZE PARTITION "
+                    + "`p_assumed` INTO (PARTITION `p_20210303` VALUES LESS THAN "
+                    + "(11, 22), PARTITION `p_20210402` VALUES LESS THAN "
+                    + "(11, 22), PARTITION `p_20210502` VALUES LESS THAN "
+                    + "MAXVALUE, MAXVALUE);",
+                    "CREATE OR REPLACE TRIGGER copy_inserts_from_map_table_"
+                    + "to_map_table_new_20210303",
+                    "\tAFTER INSERT ON map_table FOR EACH ROW",
+                    "\t\tINSERT INTO map_table_new_20210303 SET",
+                    "\t\t\t`auth` = NEW.`auth`,",
+                    "\t\t\t`order` = NEW.`order`;",
+                ]
+            },
+        )
+
+    def test_trigger_column_copies(self):
+        self.assertEqual(list(_trigger_column_copies([])), [])
+        self.assertEqual(list(_trigger_column_copies(["a"])), ["`a` = NEW.`a`"])
+        self.assertEqual(
+            list(_trigger_column_copies(["b", "a", "c"])),
+            ["`b` = NEW.`b`", "`a` = NEW.`a`", "`c` = NEW.`c`"],
+        )
+
+    def test_suffix(self):
+        self.assertEqual(list(_suffix(["a"])), ["a"])
+        self.assertEqual(list(_suffix(["a", "b"])), ["a", "b"])
+        self.assertEqual(list(_suffix(["a", "b"], indent=" ")), [" a", " b"])
+        self.assertEqual(list(_suffix(["a", "b"], mid_suffix=",")), ["a,", "b"])
+        self.assertEqual(list(_suffix(["a", "b"], final_suffix=";")), ["a", "b;"])
+        self.assertEqual(
+            list(_suffix(["a", "b"], mid_suffix=",", final_suffix=";")), ["a,", "b;"]
+        )
+
+    def test_generate_sql_copy_commands(self):
+        conf = Config()
+        conf.assume_partitioned_on = ["id"]
+        conf.curtime = datetime(2021, 3, 3)
+        conf.dbcmd = MockDatabase()
+        map_data = _override_config_to_map_data(conf)
+        cmds = list(
+            _generate_sql_copy_commands(
+                Table("old"),
+                map_data,
+                ["id", "field"],
+                Table("new"),
+                ["STRAIGHT_UP_INSERTED", "STUFF GOES HERE"],
+            )
+        )
+
+        print(cmds)
+        self.assertEqual(
+            cmds,
+            [
+                "DROP TABLE IF EXISTS new;",
+                "CREATE TABLE new LIKE old;",
+                "ALTER TABLE new REMOVE PARTITIONING;",
+                "ALTER TABLE new PARTITION BY RANGE(id) (",
+                "\tPARTITION p_assumed VALUES LESS THAN MAXVALUE",
+                ");",
+                "STRAIGHT_UP_INSERTED",
+                "STUFF GOES HERE",
+                "CREATE OR REPLACE TRIGGER copy_inserts_from_old_to_new",
+                "\tAFTER INSERT ON old FOR EACH ROW",
+                "\t\tINSERT INTO new SET",
+                "\t\t\t`field` = NEW.`field`,",
+                "\t\t\t`id` = NEW.`id`;",
+                "CREATE OR REPLACE TRIGGER copy_updates_from_old_to_new",
+                "\tAFTER UPDATE ON old FOR EACH ROW",
+                "\t\tUPDATE new SET",
+                "\t\t\t`field` = NEW.`field`",
+                "\t\tWHERE `id` = NEW.`id`;",
+            ],
+        )