Use actual data for date predictions if available, not partition names. (#37)

* Use actual data for date predictions if available, not partition names.

Fixes #30

This commit limits the functionality to tables with single-column primary keys.

* More text to the README about insertion_date_query

* Set to v0.3.0 unstable

* Rename insertion_date_query to earliest_utc_timestamp_query

* Use SQL quoting in the examples

* Fix spelling error

Author: jcjones

README.md

@@ -28,10 +28,17 @@ Similar tools:
       days: 90
   dburl: "sql://user:password@localhost3306:/test_db"
   tables:
-    cats: {}
+    cats:
+      earliest_utc_timestamp_query: >
+        SELECT UNIX_TIMESTAMP(`created`) FROM `cats` WHERE `id` > '?' ORDER BY `id` ASC LIMIT 1;
     dogs:
       partition_period:
         days: 30
+      earliest_utc_timestamp_query: >
+        SELECT UNIX_TIMESTAMP(`c`.`created`) FROM `dogs` AS `d`
+          JOIN `cats` AS `c` ON `c`.`house_id` = `d`.`house_id`
+          WHERE `d`.`id` > '?'
+          ORDER BY `d`.`id` ASC LIMIT 1;
   prometheus_stats: "/tmp/prometheus-textcollect-partition-manager.prom"
 EOF
  → partition-manager --config /tmp/partman.conf.yml maintain --noop
@@ -80,15 +87,23 @@ partitionmanager:
     table1:
       retention:
         days: 60
+      earliest_utc_timestamp_query: >
+        SELECT UNIX_TIMESTAMP(created) FROM table1 WHERE id > ? ORDER BY id ASC LIMIT 1;
     table2:
       partition_period:
         days: 30
+      earliest_utc_timestamp_query: >
+        SELECT UNIX_TIMESTAMP(created) FROM table2 WHERE id > ? ORDER BY id ASC LIMIT 1;
     table3:
       retention:
         days: 14
+      earliest_utc_timestamp_query: >
+        SELECT UNIX_TIMESTAMP(created) FROM table3 WHERE id > ? ORDER BY id ASC LIMIT 1;
     table4: {}
 ```
 
+The `earliest_utc_timestamp_query` entries are optional SQL queries that are run during partition map analysis to determine the eact timestamp of the earliest entry in each partition. If you configure such a query for a table, it must return a single row and column, specifically the epoch timestamp in UTC of the earliest entry the partition. There is expcected a single `?` entry which will be replaced with the partition value of that partition.
+
 For tables which are either partitioned but not yet using this tool's schema, or which have no empty partitions, the `migrate` command can be useful for proposing alterations to run manually. Note that `migrate` proposes commands that are likely to require partial copies of each table, so likely they will require a maintenance period.
 
 ```sh

partitionmanager/cli.py

@@ -133,6 +133,15 @@ def from_yaml_file(self, file):
                             tabledata["partition_period"]
                         )
                     )
+                if (
+                    isinstance(tabledata, dict)
+                    and "earliest_utc_timestamp_query" in tabledata
+                ):
+                    tab.set_earliest_utc_timestamp_query(
+                        partitionmanager.types.SqlQuery(
+                            tabledata["earliest_utc_timestamp_query"]
+                        )
+                    )
 
                 self.tables.add(tab)
         if "prometheus_stats" in data:
@@ -327,6 +336,7 @@ def do_partition(conf):
             cur_pos.set_position([positions[col] for col in map_data["range_cols"]])
 
             sql_cmds = pm_tap.get_pending_sql_reorganize_partition_commands(
+                database=conf.dbcmd,
                 table=table,
                 partition_list=map_data["partitions"],
                 current_position=cur_pos,

partitionmanager/table_append_partition.py

@@ -2,7 +2,7 @@
 Design and perform partition management.
 """
 
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 import logging
 import operator
 import re
@@ -360,34 +360,16 @@ def _calculate_start_time(last_changed_time, evaluation_time, allowed_lifespan):
     return partition_start_time.replace(minute=0, second=0, microsecond=0)
 
 
-def _plan_partition_changes(
-    table,
-    partition_list,
-    current_position,
-    evaluation_time,
-    allowed_lifespan,
-    num_empty_partitions,
+def _get_rate_partitions_with_implicit_timestamps(
+    table, filled_partitions, current_position, evaluation_time, active_partition
 ):
-    """Return a list of partitions to modify or create.
+    """ Return a list of PositionPartitions for use in rate calculations.
 
-    This method makes recommendations in order to meet the supplied table
-    requirements, using an estimate as to the rate of fill from the supplied
-    partition_list, current_position, and evaluation_time.
+    The partitions are set with implicit timestamps.
     """
-    log = logging.getLogger(f"plan_partition_changes:{table.name}")
-
-    filled_partitions, active_partition, empty_partitions = _split_partitions_around_position(
-        partition_list, current_position
+    log = logging.getLogger(
+        f"_get_rate_partitions_with_implicit_timestamps:{table.name}"
     )
-    if not empty_partitions:
-        log.error(
-            f"Partition {active_partition.name} requires manual ALTER "
-            "as without an empty partition to manipulate, you'll need to "
-            "perform an expensive copy operation. See the bootstrap mode."
-        )
-        raise partitionmanager.types.NoEmptyPartitionsAvailableException()
-    if not active_partition:
-        raise Exception("Active Partition can't be None")
 
     rate_relevant_partitions = None
 
@@ -400,10 +382,10 @@ def _plan_partition_changes(
         # partition's dates and positions.
         rate_relevant_partitions = filled_partitions + [
             partitionmanager.types.InstantPartition(
-                active_partition.timestamp(), current_position
+                "p_current_pos", active_partition.timestamp(), current_position
             ),
             partitionmanager.types.InstantPartition(
-                evaluation_time, active_partition.position
+                "p_prev_pos", evaluation_time, active_partition.position
             ),
         ]
     else:
@@ -420,12 +402,132 @@ def _plan_partition_changes(
             lambda f: f.timestamp() < evaluation_time, filled_partitions
         )
         rate_relevant_partitions = list(filled_partitions) + [
-            partitionmanager.types.InstantPartition(evaluation_time, current_position)
+            partitionmanager.types.InstantPartition(
+                "p_current_pos", evaluation_time, current_position
+            )
         ]
 
+    return rate_relevant_partitions
+
+
+def _get_rate_partitions_with_queried_timestamps(
+    database, table, partition_list, current_position, evaluation_time, active_partition
+):
+    """ Return a list of PositionPartitions for use in rate calculations.
+
+    The partitions' timestamps are explicitly queried.
+    """
+    log = logging.getLogger(
+        f"_get_rate_partitions_with_queried_timestamps:{table.name}"
+    )
+
+    if not table.has_date_query:
+        raise ValueError("Table has no defined date query")
+
+    instant_partitions = list()
+
+    for partition in partition_list:
+        if len(partition.position) != 1:
+            raise ValueError(
+                "This method is only valid for single-column partitions right now"
+            )
+        arg = partition.position.as_sql_input()[0]
+
+        sql_select_cmd = table.earliest_utc_timestamp_query.get_statement_with_argument(
+            arg
+        )
+        log.debug(
+            "Executing %s to derive partition %s at position %s",
+            sql_select_cmd,
+            partition.name,
+            partition.position,
+        )
+
+        start = datetime.now()
+        exact_time_result = database.run(sql_select_cmd)
+        end = datetime.now()
+
+        assert len(exact_time_result) == 1
+        assert len(exact_time_result[0]) == 1
+        for key, value in exact_time_result[0].items():
+            exact_time = datetime.fromtimestamp(value, tz=timezone.utc)
+            break
+
+        log.debug(
+            "Exact time of %s returned for %s at position %s, query took %s",
+            exact_time,
+            partition.name,
+            partition.position,
+            (end - start),
+        )
+
+        instant_partitions.append(
+            partitionmanager.types.InstantPartition(
+                partition.name, exact_time, partition.position
+            )
+        )
+
+    instant_partitions.append(
+        partitionmanager.types.InstantPartition(
+            active_partition.name, evaluation_time, current_position
+        )
+    )
+
+    return instant_partitions
+
+
+def _plan_partition_changes(
+    database,
+    table,
+    partition_list,
+    current_position,
+    evaluation_time,
+    allowed_lifespan,
+    num_empty_partitions,
+):
+    """Return a list of partitions to modify or create.
+
+    This method makes recommendations in order to meet the supplied table
+    requirements, using an estimate as to the rate of fill from the supplied
+    partition_list, current_position, and evaluation_time.
+    """
+    log = logging.getLogger(f"plan_partition_changes:{table.name}")
+
+    filled_partitions, active_partition, empty_partitions = _split_partitions_around_position(
+        partition_list, current_position
+    )
+    if not empty_partitions:
+        log.error(
+            f"Partition {active_partition.name} requires manual ALTER "
+            "as without an empty partition to manipulate, you'll need to "
+            "perform an expensive copy operation. See the bootstrap mode."
+        )
+        raise partitionmanager.types.NoEmptyPartitionsAvailableException()
+    if not active_partition:
+        raise Exception("Active Partition can't be None")
+
+    if table.has_date_query:
+        rate_relevant_partitions = _get_rate_partitions_with_queried_timestamps(
+            database,
+            table,
+            filled_partitions,
+            current_position,
+            evaluation_time,
+            active_partition,
+        )
+    else:
+        rate_relevant_partitions = _get_rate_partitions_with_implicit_timestamps(
+            table,
+            filled_partitions,
+            current_position,
+            evaluation_time,
+            active_partition,
+        )
+
     rates = _get_weighted_position_increase_per_day_for_partitions(
         rate_relevant_partitions
     )
+
     log.debug(
         f"Rates of change calculated as {rates} per day from "
         f"{len(rate_relevant_partitions)} partitions"
@@ -620,6 +722,7 @@ def generate_sql_reorganize_partition_commands(table, changes):
 
 def get_pending_sql_reorganize_partition_commands(
     *,
+    database,
     table,
     partition_list,
     current_position,
@@ -656,6 +759,7 @@ def get_pending_sql_reorganize_partition_commands(
     )
 
     partition_changes = _plan_partition_changes(
+        database,
         table,
         partition_list,
         current_position,