Sharding Keys

picopyn.sharding

Build reusable sharding keys from Picodata table metadata.

A ShardingKeyFactory is created once for a table and can then calculate keys for different rows without additional database requests. A ShardingKey contains the tier and bucket_id needed to acquire a connection to the current replicaset master.

The factory uses the table's distribution and format from Picodata's _pico_table system table. Use async Pool.create_sharding_key_factory() or sync Pool.create_sharding_key_factory() instead of constructing a factory directly.

Examples:

Async:

factory = await pool.create_sharding_key_factory("orders")
key = factory.create_key({"tenant_id": 42, "region": "eu"})

async with pool.acquire_by_sharding_key(key) as conn:
    await conn.execute(
        'SELECT * FROM "orders" WHERE tenant_id = $1 AND region = $2',
        42,
        "eu",
    )

Sync:

factory = pool.create_sharding_key_factory("orders")
key = factory.create_key({"tenant_id": 42, "region": "eu"})

with pool.connection_by_sharding_key(key) as conn:
    cursor = conn.cursor()
    cursor.execute(
        'SELECT * FROM "orders" WHERE tenant_id = %s AND region = %s',
        (42, "eu"),
    )

The explicit routing methods do not fall back to an arbitrary connection when the target master cannot be resolved or is unavailable.

ShardingKey dataclass

Routing information for a single shard.

Parameters:

Name	Type	Description	Default
bucket_id	int	Bucket containing the target data.	required
tier	str	Tier containing the bucket.	'default'

Source code in picopyn/sharding.py

@dataclass(frozen=True, slots=True)
class ShardingKey:
    """Routing information for a single shard.

    Args:
        bucket_id: Bucket containing the target data.
        tier: Tier containing the bucket.
    """

    bucket_id: int
    tier: str = "default"

    def __post_init__(self) -> None:
        if not isinstance(self.bucket_id, int) or isinstance(self.bucket_id, bool):
            raise TypeError("bucket_id must be an integer")
        if self.bucket_id < 1:
            raise ValueError("bucket_id must be positive")
        if not isinstance(self.tier, str):
            raise TypeError("tier must be a string")
        if not self.tier:
            raise ValueError("tier must be a non-empty string")

ShardingKeyFactory

Create sharding keys using one table's distribution schema.

The factory is a schema snapshot. Recreate it after dropping/recreating the table or changing its distribution schema.

Source code in picopyn/sharding.py

class ShardingKeyFactory:
    """Create sharding keys using one table's distribution schema.

    The factory is a schema snapshot. Recreate it after dropping/recreating the
    table or changing its distribution schema.
    """

    def __init__(
        self,
        table_info: _TableShardingInfo,
        bucket_count: int,
        decimal_format: DecimalFormat,
    ) -> None:
        if bucket_count < 1:
            raise ValueError("bucket_count must be positive")
        self._table_info = table_info
        self._bucket_count = bucket_count
        self._decimal_format = decimal_format

    @property
    def table_name(self) -> str:
        """Name of the table whose distribution schema the factory uses."""
        return self._table_info.table_name

    @property
    def tier(self) -> str:
        """Tier containing the table."""
        return self._table_info.tier

    @property
    def bucket_count(self) -> int:
        """Number of buckets in the table's tier."""
        return self._bucket_count

    @property
    def sharding_fields(self) -> tuple[str, ...]:
        """Table fields forming the sharding key, in distribution order."""
        return tuple(field.name for field in self._table_info.fields)

    def create_key(self, values: Mapping[str, Any]) -> ShardingKey:
        """Create a sharding key from table field values.

        `values` may contain the complete row: fields outside the table's
        sharding key are ignored.

        The field values are hashed to calculate `bucket_id`.

        Args:
            values: Mapping from table field names to their values.

        Returns:
            The calculated sharding key.

        Raises:
            TypeError: If `values` is not a mapping.
            ValueError: If a sharding field is missing, has an unsupported
                value, or the sharding key cannot be calculated.
        """
        if not isinstance(values, Mapping):
            raise TypeError("values must be a mapping")

        params: list[Any] = []
        for field in self._table_info.fields:
            try:
                params.append(values[field.name])
            except KeyError as e:
                raise ValueError(f"Missing value for sharding key field: {field.name}") from e

        dk_meta = [
            (index, field.type_oid) for index, field in enumerate(self._table_info.fields, start=1)
        ]
        bucket_id = calculate_bucket_id(
            dk_meta,
            params,
            self._bucket_count,
            decimal_format=self._decimal_format,
        )
        if bucket_id is None:
            raise ValueError(
                f"Cannot calculate sharding key for table {self.table_name!r} from given values"
            )
        return ShardingKey(bucket_id=bucket_id, tier=self.tier)

bucket_count property

Number of buckets in the table's tier.

sharding_fields property

Table fields forming the sharding key, in distribution order.

table_name property

Name of the table whose distribution schema the factory uses.

tier property

Tier containing the table.

create_key(values)

Create a sharding key from table field values.

values may contain the complete row: fields outside the table's sharding key are ignored.

The field values are hashed to calculate bucket_id.

Parameters:

Name	Type	Description	Default
values	Mapping[str, Any]	Mapping from table field names to their values.	required

Returns:

Type	Description
ShardingKey	The calculated sharding key.

Raises:

Type	Description
TypeError	If values is not a mapping.
ValueError	If a sharding field is missing, has an unsupported value, or the sharding key cannot be calculated.

Source code in picopyn/sharding.py

def create_key(self, values: Mapping[str, Any]) -> ShardingKey:
    """Create a sharding key from table field values.

    `values` may contain the complete row: fields outside the table's
    sharding key are ignored.

    The field values are hashed to calculate `bucket_id`.

    Args:
        values: Mapping from table field names to their values.

    Returns:
        The calculated sharding key.

    Raises:
        TypeError: If `values` is not a mapping.
        ValueError: If a sharding field is missing, has an unsupported
            value, or the sharding key cannot be calculated.
    """
    if not isinstance(values, Mapping):
        raise TypeError("values must be a mapping")

    params: list[Any] = []
    for field in self._table_info.fields:
        try:
            params.append(values[field.name])
        except KeyError as e:
            raise ValueError(f"Missing value for sharding key field: {field.name}") from e

    dk_meta = [
        (index, field.type_oid) for index, field in enumerate(self._table_info.fields, start=1)
    ]
    bucket_id = calculate_bucket_id(
        dk_meta,
        params,
        self._bucket_count,
        decimal_format=self._decimal_format,
    )
    if bucket_id is None:
        raise ValueError(
            f"Cannot calculate sharding key for table {self.table_name!r} from given values"
        )
    return ShardingKey(bucket_id=bucket_id, tier=self.tier)