# patroni.postgresql.mpp.citus module

class patroni.postgresql.mpp.citus.Citus(config: Dict[str, str | int])

  Bases: `AbstractMPP`

  _abc_impl = <_abc._abc_data object>
  property coordinator_group_id: int

    The group id of the Citus coordinator PostgreSQL cluster.

  property group: int

    The group of this Citus node.

  group_re: Any = re.compile('^(0|[1-9][0-9]*)$')
  static validate_config(config: Any) → bool

    Check whether provided config is good for a given MPP.

    Parameters:

      config – configuration of `citus` MPP section.

    Returns:

      `True` is config passes validation, otherwise `False`.

class patroni.postgresql.mpp.citus.CitusHandler(postgresql: Postgresql, config: Dict[str, str | int])

  Bases: `Citus`, `AbstractMPPHandler`, `Thread`

  Define the interfaces for handling an underlying Citus cluster.

  __init__(postgresql: Postgresql, config: Dict[str, str | int]) → None

    “Initialize a new instance of `CitusHandler`.

    Parameters:

      - postgresql – the Postgres node.

      - config – the `citus` MPP config section.

  _abc_impl = <_abc._abc_data object>
  _add_task(task: PgDistTask) → bool
  _in_flight: PgDistTask | None
  _pg_dist_group: Dict[int, PgDistTask]
  static _pg_dist_node(role: str, conn_url: str) → PgDistNode | None
  _tasks: List[PgDistTask]
  add_task(event: str, groupid: int, cluster: Cluster, leader_name: str, leader_url: str, timeout: float | None = None, cooldown: float | None = None) → PgDistTask | None
  adjust_postgres_gucs(parameters: Dict[str, Any]) → None

    Adjust GUCs in the current PostgreSQL configuration.

    Parameters:

      parameters – dictionary of GUCs, with key as GUC name and the corresponding value as current GUC value.

  bootstrap() → None

    Bootstrap handler.

    Is called when the new cluster is initialized (through `initdb` or a custom bootstrap method).

  find_task_by_groupid(groupid: int) → int | None
  handle_event(cluster: Cluster, event: Dict[str, Any]) → None

    Handle an event sent from a worker node.

    Parameters:

      - cluster – the currently known cluster state from DCS.

      - event – the event to be handled.

  ignore_replication_slot(slot: Dict[str, str]) → bool

    Check whether provided replication slot existing in the database should not be removed.

    Note

    MPP database may create replication slots for its own use, for example to migrate data between workers using logical replication, and we don’t want to suddenly drop them.

    Parameters:

      slot – dictionary containing the replication slot settings, like `name`, `database`, `type`, and `plugin`.

    Returns:

      `True` if the replication slots should not be removed, otherwise `False`.

  load_pg_dist_group() → bool

    Read from the pg_dist_node table and put it into the local cache

  on_demote() → None

    On demote handler.

    Is called when the primary was demoted.

  pick_task() → Tuple[int | None, PgDistTask | None]

    Returns the tuple(i, task), where i - is the task index in the self._tasks list

    Tasks are picked by following priorities:

    1. If there is already a transaction in progress, pick a task that that will change already affected worker primary.

    2. If the coordinator address should be changed - pick a task with groupid=0 (coordinators are always in groupid 0).

    3. Pick a task that is the oldest (first from the self._tasks)

  process_task(task: PgDistTask) → bool

    Updates a single row in pg_dist_group table, optionally in a transaction.

    The transaction is started if we do a demote of the worker node or before promoting the other worker if there is no transaction in progress. And, the transaction is committed when the switchover/failover completed.

    Parameters:

      task – reference to a `PgDistTask` object that represents a row to be updated/created.

    Returns:

      `True` if the row was successfully created/updated or transaction in progress was committed as an indicator that the self._pg_dist_group cache should be updated, or, if the new transaction was opened, this method returns False.

  process_tasks() → None
  query(sql: str, *params: Any) → List[Tuple[Any, ...]]
  run() → None

    Method representing the thread’s activity.

    You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

  schedule_cache_rebuild() → None

    Cache rebuild handler.

    Is called to notify handler that it has to refresh its metadata cache from the database.

  sync_meta_data(cluster: Cluster) → None

    Maintain the `pg_dist_node` from the coordinator leader every heartbeat loop.

    We can’t always rely on REST API calls from worker nodes in order to maintain pg_dist_node, therefore at least once per heartbeat loop we make sure that works registered in self._pg_dist_group cache are matching the cluster view from DCS by creating tasks the same way as it is done from the REST API.

  update_group(task: PgDistTask, transaction: bool) → None
  update_node(groupid: int, node: PgDistNode, cooldown: float = 10000) → None

class patroni.postgresql.mpp.citus.PgDistGroup(groupid: int, nodes: Collection[PgDistNode] | None = None)

  Bases: `Set`[`PgDistNode`]

  A `set`-like object that represents a Citus group in “pg_dist_node” table.

  This class implements a set of methods to compare topology and if it is necessary to transition from the old to the new topology in a “safe” manner:

  > - register new primary/secondaries

  > - replace gone secondaries with added secondaries

  > - failover and switchover

  Typically there will be at least one `PgDistNode` object registered (`primary`). In addition to that there could be one or more `secondary` nodes.

  Variables:

    - failover – whether the `primary` row should be updated as a result of `transition()` method call.

    - groupid – the “groupid” from “pg_dist_node”.

  __init__(groupid: int, nodes: Collection[PgDistNode] | None = None) → None

    Creates a `PgDistGroup` object based on given arguments.

    Parameters:

      - groupid – the groupid from “pg_dist_node”.

      - nodes – a collection of `PgDistNode` objects that belong to a groupid.

  equals(other: PgDistGroup, check_nodeid: bool = False) → bool

    Compares two `PgDistGroup` objects.

    Parameters:

      - other – what we want to compare with.

      - check_nodeid – whether `PgDistNode.nodeid` should be compared in addition to `PgDistNode.host`, `PgDistNode.port`, and `PgDistNode.role`.

    Returns:

      `True` if two `PgDistGroup` objects are fully identical.

  get(value: PgDistNode) → PgDistNode | None

    Performs a lookup of the actual value in a set.

    Note

    It is necessary because `__hash__()` and `__eq__()` methods in `PgDistNode` are redefined and effectively they check only `PgDistNode.host` and `PgDistNode.port` attributes.

    Parameters:

      value – the key we search for.

    Returns:

      the actual `PgDistNode` value from this `PgDistGroup` object or `None` if not found.

  primary() → PgDistNode | None

    Finds and returns `PgDistNode` object that represents the “primary”.

    Returns:

      `PgDistNode` object which represents the “primary” or `None` if not found.

  transition(old: PgDistGroup) → Iterator[PgDistNode]

    Compares this topology with the old one and yields transitions that transform the old to the new one.

    Note

    The actual yielded object is `PgDistNode` that will be passed to the `CitusHandler.update_node()` to execute all transitions in a transaction.

    In addition to the yielding transactions this method fills up `PgDistNode.nodeid` attribute for nodes that are presented in the old and in the new topology.

    There are a few simple rules/constraints that are imposed by Citus and must be followed: - adding/removing nodes is only possible when metadata is synced to all registered “priorities”.

    - the “primary” row in “pg_dist_node” always keeps the nodeid (unless it is removed, but it is not supported by Patroni).

    - “nodename”, “nodeport” must be unique across all rows in the “pg_dist_node”. This means that every time we want to change the nodeid of an existing node (i.e. to change it from secondary to primary), we should first write some other “nodename”/”nodeport” to the row it’s currently in.

    - updating “broken” nodes always works and metadata is synced asynchnonously after the commit.

    Following these rules below is an example of the switchover between node1 (primary, nodeid=4) and node2 (secondary, nodeid=5).

        BEGIN;
            SELECT citus_update_node(4, 'node1-demoted', 5432);
            SELECT citus_update_node(5, 'node1', 5432);
            SELECT citus_update_node(4, 'node2', 5432);
        COMMIT;

    Parameters:

      old – the last known topology registered in “pg_dist_node” for a given `groupid`.

    Yields:

      `PgDistNode` objects that must be updated/added/removed in “pg_dist_node”.

class patroni.postgresql.mpp.citus.PgDistNode(host: str, port: int, role: str, nodeid: int | None = None)

  Bases: `object`

  Represents a single row in “pg_dist_node” table.

  Note

  Unlike “noderole” possible values of `role` are `primary`, `secondary`, and `demoted`. The last one is used to pause client connections on the coordinator to the worker by appending `-demoted` suffix to the “nodename”. The actual “noderole” in DB remains `primary`.

  Variables:

    - host – “nodename” value

    - port – “nodeport” value

    - role – “noderole” value

    - nodeid – “nodeid” value

  __init__(host: str, port: int, role: str, nodeid: int | None = None) → None

    Create a `PgDistNode` object based on given arguments.

    Parameters:

      - host – “nodename” of the Citus coordinator or worker.

      - port – “nodeport” of the Citus coordinator or worker.

      - role – “noderole” value.

      - nodeid – id of the row in the “pg_dist_node”.

  as_tuple(include_nodeid: bool = False) → Tuple[str, int, str, int | None]

    Helper method to compare two `PgDistGroup` objects.

    Note

    include_nodeid is set to `True` only in unit-tests.

    Parameters:

      include_nodeid – whether `nodeid` should be taken into account when comparison is performed.

    Returns:

      `tuple` object with `host`, `port`, `role`, and optionally `nodeid`.

  is_primary() → bool

    Checks whether this object represents “primary” in a corresponding group.

    Returns:

      `True` if this object represents the `primary`.

class patroni.postgresql.mpp.citus.PgDistTask(groupid: int, nodes: Collection[PgDistNode] | None, event: str, timeout: float | None = None, cooldown: float | None = None)

  Bases: `PgDistGroup`

  A “task” that represents the current or desired state of “pg_dist_node” for a provided groupid.

  Variables:

    - group – the “groupid” in “pg_dist_node”.

    - event – an “event” that resulted in creating this task. possible values: “before_demote”, “before_promote”, “after_promote”.

    - timeout – a transaction timeout if the task resulted in starting a transaction.

    - cooldown – the cooldown value for `citus_update_node()` UDF call.

    - deadline – the time in unix seconds when the transaction is allowed to be rolled back.

  __init__(groupid: int, nodes: Collection[PgDistNode] | None, event: str, timeout: float | None = None, cooldown: float | None = None) → None

    Create a `PgDistTask` object based on given arguments.

    Parameters:

      - groupid – the groupid from “pg_dist_node”.

      - nodes – a collection of `PgDistNode` objects that belong to a groupid.

      - event – an “event” that resulted in creating this task.

      - timeout – a transaction timeout if the task resulted in starting a transaction.

      - cooldown – the cooldown value for `citus_update_node()` UDF call.

  deadline: float
  wait() → None

    Wait until this task is processed by a dedicated thread.

  wakeup() → None

    Notify a thread that created a task that it was processed.