Move partition definitions from tuples to explicit classes

Author: jcjones

partitionmanager/table_append_partition.py

@@ -1,6 +1,9 @@
 from partitionmanager.types import (
  DuplicatePartitionException,
+ MaxValuePartition,
  MismatchedIdException,
+ Partition,
+ PositionPartition,
  SqlInput,
  TableInformationException,
  UnexpectedPartitionException,
@@ -51,8 +54,12 @@ def get_current_positions(database, table_name, columns):
  columns_str = ", ".join([f"`{x}`" for x in columns])
  sql = f"SELECT {columns_str} FROM `{table_name}` ORDER BY {order_col} DESC LIMIT 1;"
  rows = database.run(sql)
- if len(rows) != 1:
- raise TableInformationException("Expected one result")
+ if len(rows) > 1:
+ raise TableInformationException(f"Expected one result from {table_name}")
+ if len(rows) == 0:
+ raise TableInformationException(
+ f"Table {table_name} appears to be empty. (No results)"
+ )
  ordered_positions = list()
  for c in columns:
  ordered_positions.append(rows[0][c])
@@ -71,9 +78,8 @@ def get_partition_map(database, table_name):
 
 def parse_partition_map(rows):
  """
- Read a partition statement from a table creation string and produce tuples
- for each partition with a max, and a single string for the partition using
- "maxvalue".
+ Read a partition statement from a table creation string and produce Partition
+ objets for each partition.
  """
  partition_range = re.compile(r"[ ]*PARTITION BY RANGE \(([\w,` ]+)\)")
  partition_member = re.compile(
@@ -110,13 +116,17 @@ def parse_partition_map(rows):
  )
  raise MismatchedIdException("Partition columns mismatch")
 
- partitions.append((part_name, part_vals))
+ pos_part = PositionPartition(part_name)
+ for v in part_vals:
+ pos_part.add_position(v)
+
+ partitions.append(pos_part)
 
  member_tail = partition_tail.match(l)
  if member_tail:
  part_name = member_tail.group(1)
  logging.debug(f"Found tail partition named {part_name}")
- partitions.append(part_name)
+ partitions.append(MaxValuePartition(part_name, len(range_cols)))
 
  return {"range_cols": range_cols, "partitions": partitions}
 
@@ -130,31 +140,43 @@ def parition_name_now():
 
 def reorganize_partition(partition_list, new_partition_name, partition_positions):
  """
- From a partial partitions list (ending with a single value that indicates MAX VALUE),
- add a new partition at the partition_positions, which must be a list.
+ From a partial partitions list of Partition types add a new partition at the
+ partition_positions, which must be a list.
  """
  if type(partition_positions) is not list:
  raise ValueError()
 
- last_value = partition_list.pop()
- if type(last_value) is not str:
- raise UnexpectedPartitionException(last_value)
- if last_value == new_partition_name:
- raise DuplicatePartitionException(last_value)
- if type(partition_list[0][1]) is list:
- if len(partition_list[0][1]) != len(partition_positions):
- raise MismatchedIdException("Didn't get the same number of partition IDs")
- else:
- if len(partition_positions) != 1:
- raise MismatchedIdException("Expected only a single partition ID")
+ num_partition_ids = partition_list[0].num_columns
+
+ tail_part = partition_list.pop()
+ if not isinstance(tail_part, MaxValuePartition):
+ raise UnexpectedPartitionException(tail_part)
+ if tail_part.name == new_partition_name:
+ raise DuplicatePartitionException(tail_part)
+
+ # Check any remaining partitions in the list after popping off the tail
+ # to make sure each entry has the same number of partition IDs as the first
+ # entry.
+ for p in partition_list:
+ if len(p.positions) != num_partition_ids:
+ raise MismatchedIdException(
+ "Didn't get the same number of partition IDs: "
+ + f"{p} has {len(p)} while expected {num_partition_ids}"
+ )
+ if len(partition_positions) != num_partition_ids:
+ raise MismatchedIdException(
+ f"Provided {len(partition_positions)} partition IDs,"
+ + f" but expected {num_partition_ids}"
+ )
+
+ altered_partition = PositionPartition(tail_part.name)
+ for p in partition_positions:
+ altered_partition.add_position(p)
 
- positions_str = ", ".join([str(x) for x in partition_positions])
- maxvalue_str = ", ".join(["MAXVALUE"] * len(partition_positions))
+ new_partition = MaxValuePartition(new_partition_name, num_partition_ids)
 
- reorganized_list = list()
- reorganized_list.append((last_value, f"({positions_str})"))
- reorganized_list.append((new_partition_name, maxvalue_str))
- return last_value, reorganized_list
+ reorganized_list = [altered_partition, new_partition]
+ return altered_partition.name, reorganized_list
 
 
 def format_sql_reorganize_partition_command(
@@ -166,9 +188,9 @@ def format_sql_reorganize_partition_command(
  """
  partition_strings = list()
  for p in partition_list:
- if type(p) is not tuple:
+ if not isinstance(p, Partition):
  raise UnexpectedPartitionException(p)
- partition_strings.append(f"PARTITION `{p[0]}` VALUES LESS THAN {p[1]}")
+ partition_strings.append(f"PARTITION `{p.name}` VALUES LESS THAN {p.values()}")
  partition_update = ", ".join(partition_strings)
 
  return (

partitionmanager/table_append_partition_test.py

@@ -5,9 +5,12 @@
 from partitionmanager.types import (
  DatabaseCommand,
  DuplicatePartitionException,
- TableInformationException,
+ MaxValuePartition,
  MismatchedIdException,
+ Partition,
+ PositionPartition,
  SqlInput,
+ TableInformationException,
  UnexpectedPartitionException,
 )
 from partitionmanager.table_append_partition import (
@@ -77,7 +80,7 @@ def test_single_partition(self):
  ]
  results = parse_partition_map(create_stmt)
  self.assertEqual(len(results["partitions"]), 1)
- self.assertEqual(results["partitions"][0], "p_20201204")
+ self.assertEqual(results["partitions"][0], mkTailPart("p_20201204"))
  self.assertEqual(results["range_cols"], ["id"])
 
  def test_two_partitions(self):
@@ -96,8 +99,8 @@ def test_two_partitions(self):
  ]
  results = parse_partition_map(create_stmt)
  self.assertEqual(len(results["partitions"]), 2)
- self.assertEqual(results["partitions"][0], ("before", [100]))
- self.assertEqual(results["partitions"][1], "p_20201204")
+ self.assertEqual(results["partitions"][0], mkPPart("before", 100))
+ self.assertEqual(results["partitions"][1], mkTailPart("p_20201204"))
  self.assertEqual(results["range_cols"], ["id"])
 
  def test_dual_keys_single_partition(self):
@@ -115,7 +118,7 @@ def test_dual_keys_single_partition(self):
  ]
  results = parse_partition_map(create_stmt)
  self.assertEqual(len(results["partitions"]), 1)
- self.assertEqual(results["partitions"][0], "p_start")
+ self.assertEqual(results["partitions"][0], mkTailPart("p_start", count=2))
  self.assertEqual(results["range_cols"], ["firstID", "secondID"])
 
  def test_dual_keys_multiple_partitions(self):
@@ -134,8 +137,8 @@ def test_dual_keys_multiple_partitions(self):
  ]
  results = parse_partition_map(create_stmt)
  self.assertEqual(len(results["partitions"]), 2)
- self.assertEqual(results["partitions"][0], ("p_start", [255, 1234567890]))
- self.assertEqual(results["partitions"][1], "p_next")
+ self.assertEqual(results["partitions"][0], mkPPart("p_start", 255, 1234567890))
+ self.assertEqual(results["partitions"][1], mkTailPart("p_next", count=2))
  self.assertEqual(results["range_cols"], ["firstID", "secondID"])
 
 
@@ -153,35 +156,55 @@ def test_okay(self):
  SqlInput("zz-table")
 
 
+def mkPPart(name, *pos):
+ p = PositionPartition(name)
+ for x in pos:
+ p.add_position(x)
+ return p
+
+
+def mkTailPart(name, count=1):
+ return MaxValuePartition(name, count)
+
+
 class TestReorganizePartitions(unittest.TestCase):
  def test_list_without_final_entry(self):
  with self.assertRaises(UnexpectedPartitionException):
- reorganize_partition([("a", 1), ("b", 2)], "new", [3])
+ reorganize_partition([mkPPart("a", 1), mkPPart("b", 2)], "new", [3])
 
  def test_reorganize_with_duplicate(self):
  with self.assertRaises(DuplicatePartitionException):
- reorganize_partition([("a", 1), "b"], "b", [3])
+ reorganize_partition([mkPPart("a", 1), mkTailPart("b")], "b", [3])
+
+ def test_reorganize_single_partition(self):
+ last_value, reorg_list = reorganize_partition([mkTailPart("a")], "b", [1])
+ self.assertEqual(last_value, "a")
+ self.assertEqual(reorg_list, [mkPPart("a", 1), mkTailPart("b")])
 
  def test_reorganize(self):
- last_value, reorg_list = reorganize_partition([("a", 1), "b"], "c", [2])
+ last_value, reorg_list = reorganize_partition(
+ [mkPPart("a", 1), mkTailPart("b")], "c", [2]
+ )
  self.assertEqual(last_value, "b")
- self.assertEqual(reorg_list, [("b", "(2)"), ("c", "MAXVALUE")])
+ self.assertEqual(reorg_list, [mkPPart("b", 2), mkTailPart("c")])
 
  def test_reorganize_too_many_partition_ids(self):
  with self.assertRaises(MismatchedIdException):
- reorganize_partition([("a", 1), "b"], "c", [2, 3, 4])
+ reorganize_partition([mkPPart("a", 1), mkTailPart("b")], "c", [2, 3, 4])
 
  def test_reorganize_too_few_partition_ids(self):
  with self.assertRaises(MismatchedIdException):
- reorganize_partition([("a", [1, 1, 1]), "b"], "c", [2, 3])
+ reorganize_partition([mkPPart("a", 1, 1, 1), mkTailPart("b")], "c", [2, 3])
 
  def test_reorganize_with_dual_keys(self):
  last_value, reorg_list = reorganize_partition(
- [("p_start", [255, 1234567890]), "p_next"], "new", [512, 2345678901]
+ [mkPPart("p_start", 255, 1234567890), mkTailPart("p_next", count=2)],
+ "new",
+ [512, 2345678901],
  )
  self.assertEqual(last_value, "p_next")
  self.assertEqual(
- reorg_list, [("p_next", "(512, 2345678901)"), ("new", "MAXVALUE, MAXVALUE")]
+ reorg_list, [mkPPart("p_next", 512, 2345678901), mkTailPart("new", count=2)]
  )
 
 

partitionmanager/types.py

@@ -38,6 +38,94 @@ def run(self, sql):
  """
 
 
+class Partition(abc.ABC):
+ """
+ Represents a single SQL table partition.
+ """
+
+ @abc.abstractmethod
+ def values(self):
+ """
+ Return a SQL partition value string.
+ """
+
+ @property
+ @abc.abstractmethod
+ def name(self):
+ """
+ Return the partition's name.
+ """
+
+ @property
+ @abc.abstractmethod
+ def num_columns(self):
+ """
+ Return the number of columns this partition represents
+ """
+
+ def __repr__(self):
+ return f"{type(self).__name__}<{str(self)}>"
+
+ def __str__(self):
+ return f"{self.name}: {self.values()}"
+
+
+class PositionPartition(Partition):
+ """
+ A partition that may have positions assocated with it.
+ """
+
+ def __init__(self, name):
+ self._name = name
+ self.positions = list()
+
+ @property
+ def name(self):
+ return self._name
+
+ def add_position(self, position):
+ self.positions.append(int(position))
+
+ @property
+ def num_columns(self):
+ return len(self.positions)
+
+ def values(self):
+ return "(" + ", ".join([str(x) for x in self.positions]) + ")"
+
+ def __eq__(self, other):
+ if isinstance(other, PositionPartition):
+ return self._name == other._name and self.positions == other.positions
+ return False
+
+
+class MaxValuePartition(Partition):
+ """
+ A partition that lives at the tail of a partition list, saying
+ all remaining values belong in this partition.
+ """
+
+ def __init__(self, name, count):
+ self._name = name
+ self.count = count
+
+ @property
+ def name(self):
+ return self._name
+
+ @property
+ def num_columns(self):
+ return self.count
+
+ def values(self):
+ return ", ".join(["MAXVALUE"] * self.count)
+
+ def __eq__(self, other):
+ if isinstance(other, MaxValuePartition):
+ return self._name == other._name and self.count == other.count
+ return False
+
+
 class MismatchedIdException(Exception):
  """
  Raised if the partition map doesn't use the primary key as its range id.