Handle mispredictions cleanly.

partitionmanager/table_append_partition.py

@@ -211,7 +211,7 @@ def _get_position_increase_per_day(p1, p2):
         # An empty list skips this pair in get_weighted_position_increase
         return list()
     if p1.timestamp() >= p2.timestamp():
-        raise ValueError(f"p1 {p1} must be before p2 {p2}")
+        raise ValueError(f"p1 {p1} must have a timestamp before p2 {p2}")
     if p1.num_columns != p2.num_columns:
         raise ValueError(f"p1 {p1} and p2 {p2} must have the same number of columns")
     delta_time = p2.timestamp() - p1.timestamp()
@@ -349,26 +349,37 @@ def _plan_partition_changes(
     if not active_partition:
         raise Exception("Active Partition can't be None")
 
-    if active_partition.timestamp() >= evaluation_time:
-        raise ValueError(
-            f"Evaluation time ({evaluation_time}) must be after "
-            f"the active partition {active_partition}."
+    rate_relevant_partitions = None
+
+    if active_partition.timestamp() < evaluation_time:
+        # This bit of weirdness is a fencepost issue: The partition list is strictly
+        # increasing until we get to "now" and the active partition. "Now" actually
+        # takes place _after_ active partition's start date (naturally), but
+        # contains a position that is before the top of active, by definition. For
+        # the rate processing to work, we need to swap the "now" and the active
+        # partition's dates and positions.
+        rate_relevant_partitions = filled_partitions + [
+            partitionmanager.types.InstantPartition(
+                active_partition.timestamp(), current_positions
+            ),
+            partitionmanager.types.InstantPartition(
+                evaluation_time, active_partition.positions
+            ),
+        ]
+    else:
+        # If the active partition's start date is later than today, then we
+        # previously mispredicted the rate of change. There's nothing we can
+        # do about that at this point, except limit our rate-of-change calculation
+        # to exclude the future-dated, irrelevant partition.
+        log.debug(
+            f"Misprediction: Evaluation time ({evaluation_time}) is "
+            f"before the active partition {active_partition}. Excluding from "
+            "rate calculations."
         )
+        rate_relevant_partitions = filled_partitions + [
+            partitionmanager.types.InstantPartition(evaluation_time, current_positions)
+        ]
 
-    # This bit of weirdness is a fencepost issue: The partition list is strictly
-    # increasing until we get to "now" and the active partition. "Now" actually
-    # takes place _after_ active partition's start date (naturally), but
-    # contains a position that is before the top of active, by definition. For
-    # the rate processing to work, we need to swap the "now" and the active
-    # partition's dates and positions.
-    rate_relevant_partitions = filled_partitions + [
-        partitionmanager.types.InstantPartition(
-            active_partition.timestamp(), current_positions
-        ),
-        partitionmanager.types.InstantPartition(
-            evaluation_time, active_partition.positions
-        ),
-    ]
     rates = _get_weighted_position_increase_per_day_for_partitions(
         rate_relevant_partitions
     )

partitionmanager/table_append_partition_test.py

@@ -725,6 +725,38 @@ def test_plan_partition_changes(self):
             ],
         )
 
+    def test_plan_partition_changes_misprediction(self):
+        """ We have to handle the case where the partition list doesn't cleanly
+        match reality. """
+        planned = _plan_partition_changes(
+            [
+                mkPPart("p_20210505", 9505010028),
+                mkPPart("p_20210604", 10152257517),
+                mkPPart("p_20210704", 10799505006),
+                mkTailPart("p_20210803"),
+            ],
+            [10264818175],
+            datetime(2021, 6, 8, tzinfo=timezone.utc),
+            timedelta(days=30),
+            3,
+        )
+
+        self.assertEqual(
+            planned,
+            [
+                ChangePlannedPartition(mkPPart("p_20210704", 10799505006)),
+                ChangePlannedPartition(mkTailPart("p_20210803")).set_position(
+                    [11578057459]
+                ),
+                NewPlannedPartition()
+                .set_position([12356609912])
+                .set_timestamp(datetime(2021, 9, 2, tzinfo=timezone.utc)),
+                NewPlannedPartition()
+                .set_columns(1)
+                .set_timestamp(datetime(2021, 10, 2, tzinfo=timezone.utc)),
+            ],
+        )
+
     def test_should_run_changes(self):
         self.assertFalse(
             _should_run_changes(