Predictive partitioning (#3)

* First pass algorithm

* Add ability to compare partition positions

* Add a split method for dividing partition lists

* More tests

* Add a position rate function

* Add methods to determine a weighted rate of increase

* Add docs to the new table_append_partition methods

* Use the Partition timestamp() method

* plan_partition_changes algorithm

* More partition planning tests

* Predictive partitiong algorithm functioning in tests

* Rework the CLI to use the new partition planning algorithm

* Passing integration tests

* Handle short and bespoke partition names.

* Improve logging

* Remove spurious strip

* Moving to 0.2.0

* Logging cleanups

* Fix a host of pylint issues

$ pylint --ignore-patterns=.*_test.py partitionmanager/ --disable W1203 --disable invalid-name --disable bad-continuation

************* Module partitionmanager.tools
partitionmanager/tools.py:22:11: R1708: Do not raise StopIteration in generator, use return statement instead (stop-iteration-return)
************* Module partitionmanager.sql
partitionmanager/sql.py:36:0: R0903: Too few public methods (1/2) (too-few-public-methods)
************* Module partitionmanager.stats
partitionmanager/stats.py:12:0: R0903: Too few public methods (0/2) (too-few-public-methods)
partitionmanager/stats.py:65:0: R0912: Too many branches (14/12) (too-many-branches)
************* Module partitionmanager.table_append_partition
partitionmanager/table_append_partition.py:98:0: R0914: Too many local variables (16/15) (too-many-locals)
partitionmanager/table_append_partition.py:306:0: R0914: Too many local variables (23/15) (too-many-locals)

------------------------------------------------------------------
Your code has been rated at 9.92/10 (previous run: 9.91/10, +0.01)

* Better logging on partition

* Never adjust the active_partition

MariaDB has a limitation on editing the active partition, particularly:

`ERROR 1520 (HY000): Reorganize of range partitions cannot change total ranges
except for last partition where it can extend the range`

so we can't edit the active partition, either.

* Never edit positions on empty partitions

Like the previous commit, MariaDB has a limitation on editing any partition's
offset:

    `ERROR 1520 (HY000): Reorganize of range partitions cannot change total ranges
    except for last partition where it can extend the range`

So the positions field should never be edited for existing partitions, only
their names.

* Consolidate logic to use partition names as start-of-fill dates

* stderr is not so useful from the Subprocess Database Command, let's dump it

* Bugfix: get_current_positions needs to query the latest of each column

Before, get_current_positions returned each column for the entry with the
largest ID from the first column, while for partitioning purposes we
actually want to always be strictly increasing.

This does make such tables less space-efficient, but that's a matter for
partition design.

* Add "bootstrap" methods to prepare partitioned tables

Tables whose partitions don't contain datestamps of the p_YYYYMMDD form don't
provide partman enough info to derive rates of change, so these bootstrap
routines will save a YAML file somewhere with point-in-time data that can be
reloaded to derive a rate-of-change. This is only intended to be used for the
initial partitioning of a table, or when a table has no empty partitions.

In a subsequent commit I'll tie this into cli.py, ensuring to add alerts that
these ALTERs cannot be expected to complete quickly, that likely the database
will hold locks for substantial amounts of time for each of the ALTER commands,
and the tool will simply be printing potential ALTER commands to console for
an operator to analyze and run in the manner they find best.

* Wire up Bootstrap to the CLI

* Rework CLI to print yaml-like but stringified output

Author: jcjones

README.md

@@ -34,6 +34,7 @@ partitionmanager:
   # mariadb: /usr/local/bin/mariadb
   partition_period:
     days: 7
+  num_empty: 2
 
   tables:
     table1:
@@ -48,6 +49,46 @@ partitionmanager:
 ```
 
 
+# Algorithm
+
+For a given table and that table's intended partition period, desired end-state is to have:
+- All the existing partitions containing data,
+- A configurable number of trailing partitions which contain no data, and
+- An "active" partition currently being filled with data
+
+To make it easier to manage, we give all the filled partitions a name to indicate the approximate date that partition began being filled with data. This date is approximate because once a partition contains data, it is no longer an instant `ALTER` operation to rename the partition, rather every contained row gets copied, so this tool predicts the date at which the new partition will become the "active" one.
+
+Inputs:
+- The table name
+- The intended partition period
+- The number of trailing partitions to keep
+- The table's current partition list
+- The table's partition id's current value(s)
+
+Outputs:
+- An intended partition list, changing only the empty partitions, or
+- If no partitions can be reorganized, an error.
+
+Procedure:
+- Using the current values, split the partition list into two sub-lists: empty partitions, and non-empty partitions.
+- If there are no empty partitions:
+  - Raise an error and halt the algorithm.
+
+- Perform a statistical regression using each non-empty partition to determine each partition's fill rate.
+- Using each partition's fill rate and their age, predict the future partition fill rate.
+- Create a new list of intended empty partitions.
+- For each empty partition:
+  - Predict the start-of-fill date using the partition's position relative to the current active partition, the current active partition's date, the partition period, and the future partition fill rate.
+  - Predict the end-of-fill value using the start-of-fill date and the future partition fill rate.
+  - If the start-of-fill date is different than the partition's name, rename the partition.
+  - If the end-of-fill value is different than the partition's current value, change that value.
+  - Append the changed partition to the intended empty partition list.
+- While the number of empty partitions is less than the intended number of trailing partitions to keep:
+  - Predict the start-of-fill date for a new partition using the previous partition's date and the partition period.
+  - Predict the end-of-fill value using the start-of-fill date and the future partition fill rate.
+  - Append the new partition to the intended empty partition list.
+- Return the lists of non-empty partitions, the current empty partitions, and the post-algorithm intended empty partitions.
+
 # TODOs
 
 Lots. A drop mechanism, for one. Yet more tests, particularly live integration tests with a test DB, for another.

partitionmanager/bootstrap.py

@@ -0,0 +1,182 @@
+"""
+Bootstrap a table that does not have sufficient partitions to determine rates
+of change.
+"""
+
+from datetime import timedelta
+import logging
+import operator
+import yaml
+
+from partitionmanager.types import (
+    ChangePlannedPartition,
+    MaxValuePartition,
+    NewPlannedPartition,
+)
+from partitionmanager.table_append_partition import (
+    table_is_compatible,
+    get_current_positions,
+    get_partition_map,
+    generate_sql_reorganize_partition_commands,
+)
+from .tools import iter_show_end
+
+RATE_UNIT = timedelta(hours=1)
+MINIMUM_FUTURE_DELTA = timedelta(hours=2)
+
+
+def write_state_info(conf, out_fp):
+    """
+    Write the state info for tables defined in conf to the provided file-like
+    object.
+    """
+    log = logging.getLogger("write_state_info")
+
+    log.info("Writing current state information")
+    state_info = {"time": conf.curtime, "tables": dict()}
+    for table in conf.tables:
+        problem = table_is_compatible(conf.dbcmd, table)
+        if problem:
+            raise Exception(problem)
+
+        map_data = get_partition_map(conf.dbcmd, table)
+        positions = get_current_positions(conf.dbcmd, table, map_data["range_cols"])
+
+        log.info(f'(Table("{table.name}"): {positions}),')
+        state_info["tables"][str(table.name)] = positions
+
+    yaml.dump(state_info, out_fp)
+
+
+def _get_time_offsets(num_entries, first_delta, subseq_delta):
+    """
+    Construct a list of timedeltas of size num_entries of the form
+    [ first_delta, subseq_delta, [subseq_delta...] ]
+    """
+    if num_entries < 1:
+        raise ValueError("Must request at least one entry")
+
+    time_units = [first_delta]
+    while len(time_units) < num_entries:
+        prev = time_units[-1]
+        time_units.append(prev + subseq_delta)
+
+    return time_units
+
+
+def _plan_partitions_for_time_offsets(
+    now_time, time_offsets, rate_of_change, ordered_current_pos, max_val_part
+):
+    """
+    Return a list of PlannedPartitions, starting from now, corresponding to
+    each supplied offset that will represent the positions then from the
+    supplied current positions and the rate of change. The first planned
+    partition will be altered out of the supplied MaxValue partition.
+    """
+    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]
+        predicted_positions = [
+            int(p + i) for p, i in zip(ordered_current_pos, increase)
+        ]
+        predicted_time = now_time + offset
+
+        part = None
+        if i == 0:
+            part = (
+                ChangePlannedPartition(max_val_part)
+                .set_position(predicted_positions)
+                .set_timestamp(predicted_time)
+            )
+
+        else:
+            part = NewPlannedPartition().set_timestamp(predicted_time)
+
+            if is_final:
+                part.set_columns(len(predicted_positions))
+            else:
+                part.set_position(predicted_positions)
+
+        changes.append(part)
+    return changes
+
+
+def calculate_sql_alters_from_state_info(conf, in_fp):
+    """
+    Using the config and the input yaml file-like object, return the SQL
+    statements to bootstrap the tables in config that also have data in
+    the input yaml as a dictionary of { Table -> list(SQL ALTER statements) }
+    """
+    log = logging.getLogger("calculate_sql_alters")
+
+    log.info("Reading prior state information")
+    prior_data = yaml.safe_load(in_fp)
+
+    time_delta = (conf.curtime - prior_data["time"]) / RATE_UNIT
+    if time_delta <= 0:
+        raise ValueError(
+            f"Time delta is too small: {conf.curtime} - "
+            f"{prior_data['time']} = {time_delta}"
+        )
+
+    commands = dict()
+
+    for table_name, prior_pos in prior_data["tables"].items():
+        table = None
+        for t in conf.tables:
+            if t.name == table_name:
+                table = t
+        if not table:
+            log.info(f"Skipping {table_name} as it is not in the current config")
+            continue
+
+        problem = table_is_compatible(conf.dbcmd, table)
+        if problem:
+            raise Exception(problem)
+
+        map_data = get_partition_map(conf.dbcmd, table)
+        current_positions = get_current_positions(
+            conf.dbcmd, table, map_data["range_cols"]
+        )
+
+        ordered_current_pos = [
+            current_positions[name] for name in map_data["range_cols"]
+        ]
+        ordered_prior_pos = [prior_pos[name] for name in map_data["range_cols"]]
+
+        delta_positions = list(
+            map(operator.sub, ordered_current_pos, ordered_prior_pos)
+        )
+        rate_of_change = list(map(lambda pos: pos / time_delta, delta_positions))
+
+        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?")
+
+        log.info(
+            f"{table}, {time_delta:0.1f} hours, {ordered_prior_pos} - {ordered_current_pos}, "
+            f"{delta_positions} pos_change, {rate_of_change}/hour"
+        )
+
+        part_duration = conf.partition_period
+        if table.partition_period:
+            part_duration = table.partition_period
+
+        time_offsets = _get_time_offsets(
+            1 + conf.num_empty, MINIMUM_FUTURE_DELTA, part_duration
+        )
+
+        changes = _plan_partitions_for_time_offsets(
+            conf.curtime,
+            time_offsets,
+            rate_of_change,
+            ordered_current_pos,
+            max_val_part,
+        )
+
+        commands[table.name] = list(
+            generate_sql_reorganize_partition_commands(table, changes)
+        )
+
+    return commands

partitionmanager/bootstrap_test.py

@@ -0,0 +1,114 @@
+import io
+import unittest
+import yaml
+from datetime import datetime, timedelta
+
+from .bootstrap import (
+    _get_time_offsets,
+    calculate_sql_alters_from_state_info,
+    write_state_info,
+)
+from .cli import Config
+from .types import DatabaseCommand, Table, SqlInput
+
+
+class MockDatabase(DatabaseCommand):
+    def __init__(self):
+        self.response = []
+        self.num_queries = 0
+
+    def run(self, cmd):
+        self.num_queries += 1
+
+        if "CREATE_OPTIONS" in cmd:
+            return [{"CREATE_OPTIONS": "partitioned"}]
+
+        if "SHOW CREATE TABLE" in cmd:
+            return [
+                {
+                    "Create Table": """CREATE TABLE `burgers` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  PRIMARY KEY (`id`),
+) ENGINE=InnoDB AUTO_INCREMENT=150 DEFAULT CHARSET=utf8
+ PARTITION BY RANGE (`id`)
+(PARTITION `p_start` VALUES LESS THAN MAXVALUE ENGINE = InnoDB)"""
+                }
+            ]
+
+        if "SELECT" in cmd:
+            return [{"id": 150}]
+
+        return self.response
+
+    def db_name(self):
+        return SqlInput("the-database")
+
+
+class TestBootstrapTool(unittest.TestCase):
+    def test_writing_state_info(self):
+        conf = Config()
+        conf.curtime = datetime(2021, 3, 1)
+        conf.dbcmd = MockDatabase()
+        conf.tables = [Table("test")]
+
+        out = io.StringIO()
+
+        write_state_info(conf, out)
+
+        written_yaml = yaml.safe_load(out.getvalue())
+
+        self.assertEqual(
+            written_yaml, {"tables": {"test": {"id": 150}}, "time": conf.curtime}
+        )
+
+    def test_get_time_offsets(self):
+        self.assertEqual(
+            _get_time_offsets(1, timedelta(hours=4), timedelta(days=30)),
+            [timedelta(hours=4)],
+        )
+
+        self.assertEqual(
+            _get_time_offsets(2, timedelta(hours=4), timedelta(days=30)),
+            [timedelta(hours=4), timedelta(days=30, hours=4)],
+        )
+
+        self.assertEqual(
+            _get_time_offsets(3, timedelta(hours=4), timedelta(days=30)),
+            [
+                timedelta(hours=4),
+                timedelta(days=30, hours=4),
+                timedelta(days=60, hours=4),
+            ],
+        )
+
+    def test_read_state_info(self):
+        conf_past = Config()
+        conf_past.curtime = datetime(2021, 3, 1)
+        conf_past.dbcmd = MockDatabase()
+        conf_past.tables = [Table("test").set_partition_period(timedelta(days=30))]
+
+        state_fs = io.StringIO()
+        yaml.dump({"tables": {"test": {"id": 0}}, "time": conf_past.curtime}, state_fs)
+        state_fs.seek(0)
+
+        with self.assertRaises(ValueError):
+            calculate_sql_alters_from_state_info(conf_past, state_fs)
+
+        conf_now = Config()
+        conf_now.curtime = datetime(2021, 3, 3)
+        conf_now.dbcmd = MockDatabase()
+        conf_now.tables = [Table("test").set_partition_period(timedelta(days=30))]
+
+        state_fs.seek(0)
+        x = calculate_sql_alters_from_state_info(conf_now, state_fs)
+        self.assertEqual(
+            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);"
+                ]
+            },
+        )