Bucket ID

picopyn.bucket

Client-side bucket_id computation from Picodata's distribution key metadata.

Based on the Picodata "bucket_id calculation algorithm".

Each distribution key parameter is encoded as MessagePack, the encoded bytes are concatenated (in query meta order) and hashed with MurmurHash -- the same hash Picodata's own vshard-based sharding uses -- then reduced mod the tier's bucket count.

Supported distribution-key types

type_oid -- Picodata type -- expected Python value:

OID	Picodata type	Python value
16	BOOLEAN	bool
20	INTEGER (incl. UNSIGNED)	int
25	TEXT/STRING/VARCHAR	str
701	DOUBLE	float
1184	DATETIME	datetime.datetime (must be timezone-aware)
1700	DECIMAL	decimal.Decimal (finite, at most 28 significant and 28 fractional digits)
2950	UUID	uuid.UUID

Note

DOUBLE accepts NaN/Infinity (hashed as their raw IEEE754 bytes, same as any other float); DECIMAL doesn't -- non-finite Decimal values have no such raw encoding, so they're rejected as an unsupported value. A DECIMAL is also rejected rather than rounded once it exceeds either of the two limits a bind parameter is subject to -- 28 significant digits or 28 fractional ones -- since we can't be sure the computed bucket_id would then match the server's.

Picodata currently calculates DECIMAL bucket IDs differently depending on the pgwire parameter format. Binary NUMERIC expands a positive exponent (1E+2 is hashed as 100), while text NUMERIC preserves it. As a result, the same numeric value may have different bucket IDs in the two formats. Pass the format used to send the parameter as decimal_format; asyncpg uses DecimalFormat.BINARY, while psycopg uses DecimalFormat.TEXT for regular %s placeholders. This behavior should be revisited when the Picodata issue is fixed.

type_oid is technically a Postgres OID, but a narrow one. Picodata has several user-facing SQL types and the scalar types can be used as table distribution. Picodata rejects table creation with a non-scalar sharding key column.

DecimalFormat

Bases: str, Enum

pgwire format used to send Decimal parameters.

Source code in picopyn/bucket.py

class DecimalFormat(str, Enum):
    """pgwire format used to send `Decimal` parameters."""

    BINARY = "binary"
    TEXT = "text"

calculate_bucket_id(dk_meta, params, bucket_count, *, decimal_format=DecimalFormat.BINARY)

Compute the bucket_id a distribution-key-filtered query would land on.

Parameters:

Name	Type	Description	Default
dk_meta	Sequence[tuple[int, int]]	(parameter_index, type_oid) pairs identifying the query's distribution key, as returned by Picodata (see PreparedStatementMetadata.dk_meta). parameter_index is 1-based ($1 is index 1).	required
params	Sequence[Any]	The query's bound parameters, in $1, $2, ... order -- i.e. exactly what's passed as *args to Pool.execute/etc.	required
bucket_count	int	Total number of buckets in the target tier.	required
decimal_format	DecimalFormat	pgwire format used to send Decimal parameters. asyncpg uses DecimalFormat.BINARY; psycopg uses DecimalFormat.TEXT for regular %s placeholders. See DecimalFormat.	BINARY

Returns:

Type	Description
int | None	The computed bucket_id, in range [1, bucket_count]. None if the bucket can't be calculated: dk_meta is empty bucket_count is not positive any parameter index is out of range for params any distribution key parameter has an unsupported type_oid (see the supported types table) any distribution key parameter's value doesn't fit its type_oid

Raises:

Type	Description
ValueError	If decimal_format is not a DecimalFormat.

Source code in picopyn/bucket.py

def calculate_bucket_id(
    dk_meta: Sequence[tuple[int, int]],
    params: Sequence[Any],
    bucket_count: int,
    *,
    decimal_format: DecimalFormat = DecimalFormat.BINARY,
) -> int | None:
    """Compute the `bucket_id` a distribution-key-filtered query would land on.

    Args:
        dk_meta: `(parameter_index, type_oid)` pairs identifying the query's
            distribution key, as returned by Picodata (see
            [PreparedStatementMetadata.dk_meta][picopyn.query_metadata.PreparedStatementMetadata]).
            `parameter_index` is 1-based (`$1` is index 1).
        params: The query's bound parameters, in `$1, $2, ...` order --
            i.e. exactly what's passed as `*args` to
            [Pool.execute][picopyn.asynchronous.pool.Pool.execute]/etc.
        bucket_count: Total [number of buckets][picopyn.topology.Topology] in the target tier.
        decimal_format: pgwire format used to send `Decimal` parameters. asyncpg uses
            `DecimalFormat.BINARY`; psycopg uses `DecimalFormat.TEXT` for regular `%s`
            placeholders. See [`DecimalFormat`][picopyn.bucket.DecimalFormat].

    Returns:
        The computed `bucket_id`, in range `[1, bucket_count]`. `None` if the bucket
            can't be calculated:

            - `dk_meta` is empty
            - `bucket_count` is not positive
            - any parameter index is out of range for `params`
            - any distribution key parameter has an unsupported `type_oid` (see the
              [supported types table][picopyn.bucket--supported-distribution-key-types])
            - any distribution key parameter's value doesn't fit its `type_oid`

    Raises:
        ValueError: If `decimal_format` is not a [`DecimalFormat`][picopyn.bucket.DecimalFormat].
    """
    if not isinstance(decimal_format, DecimalFormat):
        raise ValueError("decimal_format must be a DecimalFormat value")

    if not dk_meta or bucket_count <= 0:
        return None

    buf = bytearray()
    for idx, type_oid in dk_meta:
        if not 1 <= idx <= len(params):
            return None
        encoded = _encode_param(params[idx - 1], type_oid, decimal_format)
        if encoded is None:
            return None
        buf.extend(encoded)

    digest = mmh3.hash(bytes(buf), seed=_MURMUR_SEED, signed=False)
    return digest % bucket_count + 1