Let bootstrap work on tables that aren't partitioned yet.

Add a --assume-partitioned-on flag to bootstrap, to facilitate operation on
tables that do not yet have a partition map. One needs to supply
`--assume-partitioned-on COLUMN_NAME` as many times as needed to identify all
the columns which will be part of the partition expression.

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.

Fix unit test to not depend on time of day (oops)

Catch an arithmetic error

Emit output lines with MySQL comment characters

Author: jcjones

partitionmanager/bootstrap.py

@@ -16,6 +16,18 @@
 MINIMUM_FUTURE_DELTA = timedelta(hours=2)
 
 
+def _override_config_to_map_data(conf):
+    """Return an analog to get_partition_map from override data in conf"""
+    return {
+        "range_cols": [str(x) for x in conf.assume_partitioned_on],
+        "partitions": [
+            partitionmanager.types.MaxValuePartition(
+                "p_assumed", count=len(conf.assume_partitioned_on)
+            )
+        ],
+    }
+
+
 def write_state_info(conf, out_fp):
     """
     Write the state info for tables defined in conf to the provided file-like
@@ -26,11 +38,15 @@ def write_state_info(conf, out_fp):
     log.info("Writing current state information")
     state_info = {"time": conf.curtime, "tables": dict()}
     for table in conf.tables:
-        problems = pm_tap.get_table_compatibility_problems(conf.dbcmd, table)
-        if problems:
-            raise Exception("; ".join(problems))
+        map_data = None
+        if not conf.assume_partitioned_on:
+            problems = pm_tap.get_table_compatibility_problems(conf.dbcmd, table)
+            if problems:
+                raise Exception("; ".join(problems))
+            map_data = pm_tap.get_partition_map(conf.dbcmd, table)
+        else:
+            map_data = _override_config_to_map_data(conf)
 
-        map_data = pm_tap.get_partition_map(conf.dbcmd, table)
         positions = pm_tap.get_current_positions(
             conf.dbcmd, table, map_data["range_cols"]
         )
@@ -73,7 +89,7 @@ def _plan_partitions_for_time_offsets(
     for (i, offset), is_final in partitionmanager.tools.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)
         ]
@@ -101,6 +117,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 partitionmanager.tools.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, partitionmanager.types.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
@@ -130,15 +223,22 @@ def calculate_sql_alters_from_state_info(conf, in_fp):
             log.info(f"Skipping {table_name} as it is not in the current config")
             continue
 
-        problem = pm_tap.get_table_compatibility_problems(conf.dbcmd, table)
-        if problem:
-            raise Exception(problem)
+        map_data = None
+
+        if not conf.assume_partitioned_on:
+            problem = pm_tap.get_table_compatibility_problems(conf.dbcmd, table)
+            if problem:
+                raise Exception(problem)
+            map_data = pm_tap.get_partition_map(conf.dbcmd, table)
+        else:
+            map_data = _override_config_to_map_data(conf)
 
-        map_data = pm_tap.get_partition_map(conf.dbcmd, table)
         current_positions = pm_tap.get_current_positions(
             conf.dbcmd, table, map_data["range_cols"]
         )
 
+        columns = [r["Field"] for r in pm_tap.get_columns(conf.dbcmd, table)]
+
         ordered_current_pos = [
             current_positions[name] for name in map_data["range_cols"]
         ]
@@ -178,7 +278,17 @@ def calculate_sql_alters_from_state_info(conf, in_fp):
             max_val_part,
         )
 
+        table_new = partitionmanager.types.Table(
+            f"{table.name}_new_{conf.curtime:%Y%m%d}"
+        )
+
+        alter_commands_iter = pm_tap.generate_sql_reorganize_partition_commands(
+            table_new, changes
+        )
+
         commands[table.name] = list(
-            pm_tap.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,17 +4,30 @@
 from datetime import datetime, timedelta
 
 from .bootstrap import (
+    _generate_sql_copy_commands,
     _get_time_offsets,
+    _suffix,
+    _trigger_column_copies,
+    _override_config_to_map_data,
+    _plan_partitions_for_time_offsets,
     calculate_sql_alters_from_state_info,
     write_state_info,
 )
 from .cli import Config
-from .types import DatabaseCommand, Table, SqlInput
+from .types import (
+    DatabaseCommand,
+    Table,
+    SqlInput,
+    MaxValuePartition,
+    ChangePlannedPartition,
+    NewPlannedPartition,
+)
 
 
 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,8 +49,9 @@ def run(self, cmd):
             ]
 
         if "SELECT" in cmd:
-            return [{"id": 150}]
-        return self.response
+            return self._select_response.pop()
+
+        return self._response.pop()
 
     def db_name(self):
         return SqlInput("the-database")
@@ -81,6 +95,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()
@@ -96,6 +111,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)
@@ -104,10 +125,148 @@ 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`;",
+            ],
+        )
+
+    def test_plan_partitions_for_time_offsets(self):
+        parts = _plan_partitions_for_time_offsets(
+            datetime(2021, 3, 3),
+            [timedelta(days=60), timedelta(days=360)],
+            [11943234],
+            [16753227640],
+            MaxValuePartition("p_assumed", count=1),
+        )
+        self.assertIsInstance(parts[0], ChangePlannedPartition)
+        self.assertIsInstance(parts[1], NewPlannedPartition)