Query Metadata

picopyn.query_metadata

Picodata's query metadata protocol extension: distribution key metadata for a prepared statement, and stale-plan invalidation.

Based on the Picodata pgproto metadata mechanism.

Requires Picodata 26.1.1 or newer -- older Picodata silently ignores special startup options (only logs a warning server-side), so on an older cluster this whole mechanism just quietly never produces any query-meta or stale-plan recovery, with no error on the client side.

The async and sync drivers receive a query's meta from a Picodata Notice, caches it, and detects the stale-plan signal that means it needs re-fetching (see async and sync usage). When cached metadata is available, async Pool.execute can use it to compute bucket_id and route the query to the bucket master's pooled connection.

QUERY_METADATA_OPTION = 'pico_query_metadata=true' module-attribute

Makes Picodata send a Notice with query distribution meta. Only worth requesting on a connection that actually forwards it -- the meta doesn't depend on which connection asked.

STMT_INVALIDATED_SQLSTATE = '42999' module-attribute

The stale-plan signal: the SQLSTATE Picodata raises (instead of silently using a stale plan) when a prepared statement's schema has changed since it was Parsed, requiring the caller to close it and re-Parse. Requires special connection option. See the DDL invalidation.

STMT_INVALIDATION_OPTION = 'pico_stmt_invalidation=true' module-attribute

Makes Picodata return the stale-plan signal instead of silently running a stale plan, once a prepared statement's schema has changed. Requested on every connection, not just the one doing metadata discovery, so ordinary queries can detect and recover from stale plans too.

PreparedStatementMetadata dataclass

Distribution key metadata for a prepared statement.

Sent by Picodata as a Notice right before ParseComplete, when the connection enables Notice receiving via the special connection option.

Note

Currently meta is only populated for 1-row INSERT statements: anything else (SELECT/UPDATE/...) always gets an empty dk_meta, even a simple equality filter on the distribution key like WHERE id = $1. Also empty when Picodata couldn't resolve a single bucket_id even for a supported INSERT (see Unbounded queries).

Parameters:

Name	Type	Description	Default
query	str	The exact SQL text Picodata echoes back, matching what the client sent in the Parse message.	required
tier	str | None	The tier the query's tables belong to, or None if it cannot be estimated.	required
dk_meta	list[tuple[int, int]]	(parameter_index, type_oid) pairs (see the supported types table) identifying which bind parameters make up the query's distribution key. parameter_index is 1-based, matching the query's own $1/$2/... placeholder numbering. Can be empty if Picodata cannot locate a single bucket_id for the query.	required

Source code in picopyn/query_metadata.py

@dataclass(frozen=True)
class PreparedStatementMetadata:
    """Distribution key metadata for a prepared statement.

    Sent by Picodata as a Notice right before `ParseComplete`, when the
    connection enables Notice receiving via the
    [special connection option][picopyn.query_metadata.QUERY_METADATA_OPTION].

    Note:
        [Currently](https://git.picodata.io/core/picodata/-/issues/2226) meta
        is only populated for 1-row `INSERT` statements: anything else (SELECT/UPDATE/...)
        always gets an empty `dk_meta`, even a simple equality filter on
        the distribution key like `WHERE id = $1`. Also empty when Picodata
        couldn't resolve a single `bucket_id` even for a supported INSERT (see
        [Unbounded queries](https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-query-metadata.md#unbounded-queries)).

    Args:
        query: The exact SQL text Picodata echoes back, matching what the
            client sent in the `Parse` message.
        tier: The tier the query's tables belong to, or None if it cannot be estimated.
        dk_meta: `(parameter_index, type_oid)` pairs (see the
            [supported types table][picopyn.bucket--supported-distribution-key-types])
            identifying which bind parameters make up the query's distribution key.
            `parameter_index` is 1-based, matching the query's own
            `$1`/`$2`/... placeholder numbering.

            Can be empty if Picodata cannot locate a single `bucket_id` for the query.
    """

    query: str
    tier: str | None
    dk_meta: list[tuple[int, int]]

QueryMetadataCache

Bounded, least-recently-used cache of PreparedStatementMetadata.

async Pool and sync Pool use this cache, controlled by their QueryMetadataService.

It also remembers the queries Picodata parsed but sent no metadata Notice for. Both kinds of entry share the same bound and are dropped together.

Source code in picopyn/query_metadata.py

class QueryMetadataCache:
    """Bounded, least-recently-used cache of [PreparedStatementMetadata][picopyn.query_metadata.PreparedStatementMetadata].

    [async Pool][picopyn.asynchronous.pool.Pool] and
    [sync Pool][picopyn.synchronous.pool.Pool] use this cache, controlled by their
    `QueryMetadataService`.

    It also remembers the queries Picodata parsed but sent no metadata Notice for. Both
    kinds of entry share the same bound and are dropped together.
    """

    def __init__(self, max_size: int) -> None:
        """
        Args:
            max_size: Maximum number of distinct queries to cache.
                The least recently used entry is evicted once this is exceeded.

        Raises:
            ValueError: If `max_size` is None or less than 1.
        """
        if max_size is None:
            raise ValueError("max_size can not be None")

        if max_size < 1:
            raise ValueError("max_size must be at least 1")
        self._max_size = max_size
        self._entries: OrderedDict[str, PreparedStatementMetadata | None] = OrderedDict()

    def resize(self, max_size: int) -> None:
        """Change the cache limit, immediately evicting excess LRU entries.

        Entries for queries without metadata count towards the same limit.

        Args:
            max_size: Maximum number of distinct queries to cache.

        Raises:
            ValueError: If `max_size` is None or less than 1.
        """
        if max_size is None:
            raise ValueError("max_size can not be None")
        if max_size < 1:
            raise ValueError("max_size must be at least 1")
        self._max_size = max_size
        while len(self._entries) > max_size:
            self._entries.popitem(last=False)

    def _set_entry(self, query: str, metadata: PreparedStatementMetadata | None) -> None:
        """Store an entry as the most recently used one, evicting the least
        recently used one if the cache is now over its max size.
        """
        self._entries[query] = metadata
        self._entries.move_to_end(query)
        if len(self._entries) > self._max_size:
            self._entries.popitem(last=False)

    def get(self, query: str) -> PreparedStatementMetadata | None:
        """Look up cached metadata for `query`, marking it as most recently used.

        Args:
            query: The SQL query text to look up.

        Returns:
            The cached metadata, or None on a cache miss and for a query known
            to have no metadata.
        """
        if query not in self._entries:
            return None
        self._entries.move_to_end(query)
        return self._entries[query]

    def set(self, metadata: PreparedStatementMetadata) -> None:
        """Cache query metadata, keyed by its own query text.

        Evicts the least recently used entry if the cache is now over
        its max size.

        Args:
            metadata: The metadata to cache.
        """
        self._set_entry(metadata.query, metadata)

    def set_no_metadata(self, query: str) -> None:
        """Remember that Picodata sent no metadata Notice for `query`, so
        asking again is pointless until the schema changes.

        Picodata answers such a Parse as usual (the statement is prepared, no
        error -- ParseComplete), it just does not answer with the Notice. The
        reason can be:

        - Picodata does not support the mechanism (<26.1.1), so it never sends
          the Notice for any query
        - Picodata parses the query without answering anything about it (DDL)

        The entry lives by the same rules as a cached metadata, so it can be
        evict/cleared.

        Args:
            query: The SQL query text Picodata sent no metadata Notice for.
        """
        self._set_entry(query, None)

    def is_known_without_metadata(self, query: str) -> bool:
        """True if `query` is the one Picodata already parsed (sent ParseComplete)
        without sending any metadata Notice.

        Args:
            query: The SQL query text to look up.
        """
        return query in self._entries and self._entries[query] is None

    def evict(self, query: str) -> None:
        """Drop a single cached entry, e.g. after a stale-plan error (see
        [`is_stmt_invalidated_error`][picopyn.query_metadata.is_stmt_invalidated_error]).

        Args:
            query: The SQL query text whose cached entry should be dropped.
                A no-op if nothing is cached for it.
        """
        self._entries.pop(query, None)

    def clear(self) -> None:
        """Remove all cached entries."""
        self._entries.clear()

    def __len__(self) -> int:
        """Return the number of currently cached entries."""
        return len(self._entries)

__init__(max_size)

Parameters:

Name	Type	Description	Default
max_size	int	Maximum number of distinct queries to cache. The least recently used entry is evicted once this is exceeded.	required

Raises:

Type	Description
ValueError	If max_size is None or less than 1.

Source code in picopyn/query_metadata.py

def __init__(self, max_size: int) -> None:
    """
    Args:
        max_size: Maximum number of distinct queries to cache.
            The least recently used entry is evicted once this is exceeded.

    Raises:
        ValueError: If `max_size` is None or less than 1.
    """
    if max_size is None:
        raise ValueError("max_size can not be None")

    if max_size < 1:
        raise ValueError("max_size must be at least 1")
    self._max_size = max_size
    self._entries: OrderedDict[str, PreparedStatementMetadata | None] = OrderedDict()

clear()

Remove all cached entries.

Source code in picopyn/query_metadata.py

def clear(self) -> None:
    """Remove all cached entries."""
    self._entries.clear()

evict(query)

Drop a single cached entry, e.g. after a stale-plan error (see is_stmt_invalidated_error).

Parameters:

Name	Type	Description	Default
query	str	The SQL query text whose cached entry should be dropped. A no-op if nothing is cached for it.	required

Source code in picopyn/query_metadata.py

def evict(self, query: str) -> None:
    """Drop a single cached entry, e.g. after a stale-plan error (see
    [`is_stmt_invalidated_error`][picopyn.query_metadata.is_stmt_invalidated_error]).

    Args:
        query: The SQL query text whose cached entry should be dropped.
            A no-op if nothing is cached for it.
    """
    self._entries.pop(query, None)

get(query)

Look up cached metadata for query, marking it as most recently used.

Parameters:

Name	Type	Description	Default
query	str	The SQL query text to look up.	required

Returns:

Type	Description
PreparedStatementMetadata | None	The cached metadata, or None on a cache miss and for a query known
PreparedStatementMetadata | None	to have no metadata.

Source code in picopyn/query_metadata.py

def get(self, query: str) -> PreparedStatementMetadata | None:
    """Look up cached metadata for `query`, marking it as most recently used.

    Args:
        query: The SQL query text to look up.

    Returns:
        The cached metadata, or None on a cache miss and for a query known
        to have no metadata.
    """
    if query not in self._entries:
        return None
    self._entries.move_to_end(query)
    return self._entries[query]

is_known_without_metadata(query)

True if query is the one Picodata already parsed (sent ParseComplete) without sending any metadata Notice.

Parameters:

Name	Type	Description	Default
query	str	The SQL query text to look up.	required

Source code in picopyn/query_metadata.py

def is_known_without_metadata(self, query: str) -> bool:
    """True if `query` is the one Picodata already parsed (sent ParseComplete)
    without sending any metadata Notice.

    Args:
        query: The SQL query text to look up.
    """
    return query in self._entries and self._entries[query] is None

resize(max_size)

Change the cache limit, immediately evicting excess LRU entries.

Entries for queries without metadata count towards the same limit.

Parameters:

Name	Type	Description	Default
max_size	int	Maximum number of distinct queries to cache.	required

Raises:

Type	Description
ValueError	If max_size is None or less than 1.

Source code in picopyn/query_metadata.py

def resize(self, max_size: int) -> None:
    """Change the cache limit, immediately evicting excess LRU entries.

    Entries for queries without metadata count towards the same limit.

    Args:
        max_size: Maximum number of distinct queries to cache.

    Raises:
        ValueError: If `max_size` is None or less than 1.
    """
    if max_size is None:
        raise ValueError("max_size can not be None")
    if max_size < 1:
        raise ValueError("max_size must be at least 1")
    self._max_size = max_size
    while len(self._entries) > max_size:
        self._entries.popitem(last=False)

set(metadata)

Cache query metadata, keyed by its own query text.

Evicts the least recently used entry if the cache is now over its max size.

Parameters:

Name	Type	Description	Default
metadata	PreparedStatementMetadata	The metadata to cache.	required

Source code in picopyn/query_metadata.py

def set(self, metadata: PreparedStatementMetadata) -> None:
    """Cache query metadata, keyed by its own query text.

    Evicts the least recently used entry if the cache is now over
    its max size.

    Args:
        metadata: The metadata to cache.
    """
    self._set_entry(metadata.query, metadata)

set_no_metadata(query)

Remember that Picodata sent no metadata Notice for query, so asking again is pointless until the schema changes.

Picodata answers such a Parse as usual (the statement is prepared, no error -- ParseComplete), it just does not answer with the Notice. The reason can be:

• Picodata does not support the mechanism (<26.1.1), so it never sends the Notice for any query
• Picodata parses the query without answering anything about it (DDL)

The entry lives by the same rules as a cached metadata, so it can be evict/cleared.

Parameters:

Name	Type	Description	Default
query	str	The SQL query text Picodata sent no metadata Notice for.	required

Source code in picopyn/query_metadata.py

def set_no_metadata(self, query: str) -> None:
    """Remember that Picodata sent no metadata Notice for `query`, so
    asking again is pointless until the schema changes.

    Picodata answers such a Parse as usual (the statement is prepared, no
    error -- ParseComplete), it just does not answer with the Notice. The
    reason can be:

    - Picodata does not support the mechanism (<26.1.1), so it never sends
      the Notice for any query
    - Picodata parses the query without answering anything about it (DDL)

    The entry lives by the same rules as a cached metadata, so it can be
    evict/cleared.

    Args:
        query: The SQL query text Picodata sent no metadata Notice for.
    """
    self._set_entry(query, None)

is_stmt_invalidated_error(exc)

True if the exception is a Postgres error with Picodata's stale-plan SQLSTATE.

Connection wrappers (async Connection.execute/fetch/fetchrow/prepare) wrap the original driver error in a RuntimeError, which drops the sqlstate attribute but keeps the original exception (as __cause__). So we simply use the same check to see whether it's the invalidation signal.

Parameters:

Name	Type	Description	Default
exc	BaseException	The exception to check -- either the original driver error, or a RuntimeError wrapping it via raise ... from e.	required

Returns:

Type	Description
bool	True if the exception carries the stale-plan SQLSTATE.

Source code in picopyn/query_metadata.py

def is_stmt_invalidated_error(exc: BaseException) -> bool:
    """True if the exception is a Postgres error with Picodata's stale-plan SQLSTATE.

    Connection wrappers (async `Connection.execute`/`fetch`/`fetchrow`/`prepare`)
    wrap the original driver error in a `RuntimeError`, which drops the `sqlstate`
    attribute but keeps the original exception (as `__cause__`). So we simply use
    the same check to see whether it's the invalidation signal.

    Args:
        exc: The exception to check -- either the original driver error, or a
            `RuntimeError` wrapping it via `raise ... from e`.

    Returns:
        True if the exception carries the stale-plan SQLSTATE.
    """
    if getattr(exc, "sqlstate", None) == STMT_INVALIDATED_SQLSTATE:
        return True
    cause = exc.__cause__
    return cause is not None and getattr(cause, "sqlstate", None) == STMT_INVALIDATED_SQLSTATE

parse_query_metadata_notice(message)

Parse a Notice's text as Picodata query metadata. See the Picodata message contract.

Parameters:

Name	Type	Description	Default
message	str	The raw text of a Postgres Notice message.	required

Returns:

Type	Description
PreparedStatementMetadata | None	The parsed metadata, or None if message isn't a query metadata Notice.

Source code in picopyn/query_metadata.py

def parse_query_metadata_notice(message: str) -> PreparedStatementMetadata | None:
    """Parse a Notice's text as Picodata query metadata.
    See the [Picodata message contract](https://git.picodata.io/core/picodata/-/blob/master/doc/adr/2025-09-12-pgproto-query-metadata.md#структура-preparedstatementmetadata).

    Args:
        message: The raw text of a Postgres Notice message.

    Returns:
        The parsed metadata, or None if `message` isn't a query metadata Notice.
    """
    # TODO: takes a plain str, not a driver-specific message object, since
    # asyncpg's PostgresLogMessage.message and psycopg's
    # Diagnostic.message_primary don't agree on the attribute name -- extracting
    # it is left to whichever async/sync connection code calls this function.
    #
    # Not checking the Notice's sqlstate: Picodata's pgproto has exactly one
    # call site that ever builds a NoticeResponse at all (always "00000"), so
    # it can't discriminate anything the JSON/dk_meta checks below don't already.
    try:
        payload = json.loads(message)
    except (json.JSONDecodeError, TypeError):
        # TODO: maybe use log.debug instead, since this can spam errors on older Picodata
        logger.warning("Failed to parse query metadata Notice: %r", message)
        return None

    if not isinstance(payload, dict) or _DK_META_KEY not in payload:
        return None

    return PreparedStatementMetadata(
        query=payload["query"],
        tier=payload.get("tier"),
        dk_meta=[(pair[0], pair[1]) for pair in payload[_DK_META_KEY]],
    )