Wrap wirelog's typed row API so float columns work in a Session

docs/api-stability.md

@@ -66,6 +66,8 @@ The v1 stable public API includes these exported names:
 - `ParseError`
 - `InvalidIRError`
 - `ExecError`
+- `TypedRowError`
+- `TypedErrorCode`
 - `WirelogMemoryError`
 - `WirelogIOError`
 - `CompoundSaturatedError`

src/pyrewire/__init__.py

@@ -11,6 +11,7 @@
     ExecError,
     InvalidIRError,
     ParseError,
+    TypedRowError,
     WirelogError,
     WirelogInternError,
     WirelogIOError,
@@ -28,6 +29,7 @@
     ErrorCode,
     IRNodeType,
     StrFn,
+    TypedErrorCode,
 )
 from pyrewire._ffi._util import (
     agg_fn_name,
@@ -93,6 +95,8 @@
     "WirelogError",
     "ParseError",
     "InvalidIRError",
+    "TypedRowError",
+    "TypedErrorCode",
     "ExecError",
     "WirelogMemoryError",
     "WirelogIOError",

src/pyrewire/_core/callbacks.py

@@ -17,8 +17,8 @@
    the registry slot. `CallbackHandle.drain()` then re-raises after
    wirelog has returned control to Python.
 
-The two trampolines (`_delta_trampoline`, `_tuple_trampoline`) are
-module-level singletons. Per-session instances would multiply ctypes
+The three trampolines (`_delta_trampoline`, `_tuple_trampoline`,
+`_typed_delta_trampoline`) are module-level singletons. Per-session instances would multiply ctypes
 overhead and complicate lifetime management.
 """
 
@@ -31,12 +31,15 @@
 from dataclasses import dataclass, field
 from typing import Any
 
-from .._ffi._types import OnDeltaFn, OnTupleFn
+from .._ffi._types import OnDeltaFn, OnTupleFn, OnTypedTupleFn
+from .lanes import decode_typed_row
 
-# Event payloads buffered by the trampolines. The first element ("delta"
-# or "tuple") is the trampoline kind; the rest is the decoded payload.
+# Event payloads buffered by the trampolines. The first element is the
+# trampoline kind; the rest is the decoded payload.
 DeltaEvent = tuple[str, str, tuple[int, ...], int]  # ("delta", rel, row_ids, diff)
 TupleEvent = tuple[str, str, tuple[int, ...]]  # ("tuple", rel, row_ids)
+# ("typed_delta", rel, decoded_values, diff) - values are typed, not ids
+TypedDeltaEvent = tuple[str, str, tuple[object, ...], int]
 Event = Any  # union of the above
 
 
@@ -62,7 +65,7 @@ def _next_token() -> int:
 
 # --- module-level trampolines ----------------------------------------------
 #
-# Both trampolines defensively swallow every exception. They store the
+# Every trampoline defensively swallows all exceptions. They store the
 # exception on the registry slot (best effort) so `drain()` can re-raise
 # it. Never let one propagate into wirelog's C call.
 
@@ -114,6 +117,34 @@ def _tuple_trampoline(
             state.last_error = exc
 
 
+@OnTypedTupleFn  # type: ignore[untyped-decorator]
+def _typed_delta_trampoline(
+    relation: bytes,
+    row: Any,
+    diff: int,
+    user_data: Any,
+) -> None:
+    """Typed counterpart of `_delta_trampoline`.
+
+    wirelog refuses to install the UNTYPED delta callback on a program
+    whose schema carries a FLOAT column, so this is the only delta path
+    available to such a session. The descriptor is borrowed for this call
+    only; `decode_typed_row` copies every value out before returning.
+    """
+    state: _TrampolineState | None = None
+    try:
+        raw = ctypes.cast(user_data, ctypes.c_void_p).value
+        token = int(raw) if raw else 0
+        state = _REGISTRY.get(token)
+        if state is None or not row:
+            return
+        rel = relation.decode() if relation else ""
+        state.queue.append(("typed_delta", rel, decode_typed_row(row[0]), int(diff)))
+    except BaseException as exc:  # never propagate to C
+        if state is not None:
+            state.last_error = exc
+
+
 # --- public API ------------------------------------------------------------
 
 
@@ -125,7 +156,7 @@ class CallbackHandle:
     __slots__ = ("kind", "token", "_state", "__weakref__")
 
     def __init__(self, kind: str, user_fn: Callable[..., Any] | None = None) -> None:
-        if kind not in ("delta", "tuple"):
+        if kind not in ("delta", "tuple", "typed_delta"):
             raise ValueError(f"unknown callback kind: {kind!r}")
         self.kind: str = kind
         self.token: int = _next_token()
@@ -140,7 +171,11 @@ def user_data(self) -> ctypes.c_void_p:
     @property
     def fn(self) -> Any:
         """The module-level CFUNCTYPE instance matching `kind`."""
-        return _delta_trampoline if self.kind == "delta" else _tuple_trampoline
+        if self.kind == "delta":
+            return _delta_trampoline
+        if self.kind == "typed_delta":
+            return _typed_delta_trampoline
+        return _tuple_trampoline
 
     def drain(self) -> list[Event]:
         """Pop and return all queued events. If a callback raised, the
@@ -165,4 +200,4 @@ def __del__(self) -> None:
             pass
 
 
-__all__ = ["CallbackHandle", "DeltaEvent", "TupleEvent", "Event"]
+__all__ = ["CallbackHandle", "DeltaEvent", "TupleEvent", "TypedDeltaEvent", "Event"]

src/pyrewire/_core/errors.py

@@ -110,6 +110,40 @@ class ExecError(WirelogError):
     code = int(ErrorCode.EXEC)
 
 
+class TypedRowError(ExecError):
+    """Raised when `wirelog_session_insert_typed` / `remove_typed` rejects a row.
+
+    A subclass of `ExecError` so existing `except ExecError` handlers keep
+    working, but it carries what the generic path throws away: which row
+    and column the engine objected to, and its own bounded diagnostic.
+
+    Attributes:
+        typed_code: The `TypedErrorCode` wirelog reported - `DESCRIPTOR`
+            (the descriptor is malformed), `SCHEMA` (it disagrees with the
+            relation) or `VALUE` (a lane holds an unrepresentable value).
+        row_index: Index of the offending row within the batch, or `None`
+            when the engine reported no specific row.
+        column: Logical column index within that row, or `None` when the
+            engine reported no specific column.
+        engine_message: wirelog's own message, or `None` if it supplied none.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        typed_code: int,
+        row_index: int | None = None,
+        column: int | None = None,
+        engine_message: str | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.typed_code = typed_code
+        self.row_index = row_index
+        self.column = column
+        self.engine_message = engine_message
+
+
 class WirelogMemoryError(WirelogError):
     """Raised when the wirelog engine cannot allocate memory (``WIRELOG_ERR_MEMORY``,
     code 4).
@@ -282,6 +316,7 @@ def check(rc: int) -> None:
     "WirelogError",
     "ParseError",
     "InvalidIRError",
+    "TypedRowError",
     "ExecError",
     "WirelogMemoryError",
     "WirelogIOError",

src/pyrewire/_core/lanes.py

@@ -0,0 +1,199 @@
+# SPDX-License-Identifier: Apache-2.0 OR GPL-3.0-or-later
+"""Conversion between Python values and wirelog's 64-bit typed lanes.
+
+wirelog's typed row ABI carries every column as a `uint64_t` lane plus a
+`wirelog_column_type_t` saying how to read it. A FLOAT lane holds
+host-order IEEE-754 binary64 bits rather than a C `double`, so a float
+only survives the boundary through an explicit bit reinterpret - which is
+exactly what the untyped `int64_t` entry points cannot do.
+
+These helpers live here rather than on `Session` because both the typed
+insert path (`pyrewire.session`) and the typed delta trampoline
+(`pyrewire._core.callbacks`) need them, and the trampoline cannot import
+the session module without a cycle.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import numbers
+import operator
+import struct
+from typing import Any, SupportsFloat, SupportsIndex
+
+from .._ffi._enums import ColumnType
+
+# What a single lane can hold once decoded. BOOL decodes to `bool`, which
+# is an `int` subclass, so the union is a documentation aid rather than a
+# discriminator.
+LaneValue = int | float | bool
+
+# What `encode_lane` accepts: anything integer-like or float-like. Wider
+# than `LaneValue`, so a NumPy scalar or any `__index__` type goes in.
+LaneInput = SupportsIndex | SupportsFloat
+
+_FLOAT = struct.Struct("<d")
+_BITS = struct.Struct("<Q")
+
+_UINT64_MASK = 0xFFFFFFFFFFFFFFFF
+
+
+def encode_lane(value: LaneInput, column_type: ColumnType) -> int:
+    """Pack one Python value into its 64-bit lane.
+
+    Integer columns take the value's two's-complement bits. A float that
+    names an exact integer (`3.0`) is accepted there; a non-integral one
+    is rejected rather than silently truncated.
+
+    Anything implementing `__index__` or `__float__` is accepted, so a
+    NumPy scalar works here as it does through `insert_batch`. `str` is
+    the deliberate exclusion. A float-like value heading for an integer
+    column is converted through `float`; if it is also a real numeric
+    type, the result is checked against the original and refused when the
+    conversion lost information - `Decimal(2**63 - 1)` names an integer a
+    binary64 cannot hold, and is refused rather than stored one off.
+
+    Raises:
+        ValueError: a float-like value was given for an integer column and
+            it does not name an exactly representable integer.
+        TypeError: a `str`, or a value that is neither integer-like nor
+            float-like, was given.
+        OverflowError: the value is too large to convert to a float at
+            all, as `10**400` is.
+    """
+    # Concrete types first. `isinstance` against a runtime-checkable
+    # Protocol probes for attributes and has no negative-result cache, so
+    # leaving it in front of the common case costs ~25% of a large typed
+    # insert. `type(...) is` deliberately excludes bool, which falls
+    # through to the __index__ path below and encodes as 0/1 either way.
+    is_int = type(value) is int
+    is_float = type(value) is float
+    if not is_int and not is_float:
+        if isinstance(value, str):
+            raise TypeError(
+                f"cannot encode str {value!r} into a lane: a STRING column carries an "
+                f"intern id, and the advanced Session has no forward-intern API. Pass "
+                f"the int64 id, seeding it with seed_intern(value, id) if you need to "
+                f"decode it back."
+            )
+        if not isinstance(value, (SupportsIndex, SupportsFloat)):
+            raise TypeError(f"cannot encode {type(value).__name__} {value!r} into a lane")
+
+    if column_type == ColumnType.FLOAT:
+        bits: int = _BITS.unpack(_FLOAT.pack(_as_float(value)))[0]
+        return bits
+
+    if is_int:
+        return value & _UINT64_MASK  # type: ignore[operator]
+    if not is_float and isinstance(value, SupportsIndex):
+        # `operator.index` enforces that __index__ returned an int, where
+        # calling the dunder directly would let a bogus return value reach
+        # the `&` below as some other type.
+        return operator.index(value) & _UINT64_MASK
+
+    # Float-like heading for an integer column: accept it only when it
+    # names an exact integer, and only when converting through float did
+    # not already lose the value.
+    as_float = _as_float(value)
+    if not as_float.is_integer():
+        raise ValueError(f"cannot store float {value!r} in a {ColumnType(column_type).name} column")
+    as_int = int(as_float)
+    if not is_float and _float_conversion_lost_value(value, as_int):
+        raise ValueError(
+            f"cannot store {type(value).__name__} {value!r} in a "
+            f"{ColumnType(column_type).name} column: it is not exactly representable "
+            f"as a 64-bit float, so the stored value would be {as_int}"
+        )
+    return as_int & _UINT64_MASK
+
+
+def _float_conversion_lost_value(value: object, as_int: int) -> bool:
+    """Whether routing `value` through `float` lost information.
+
+    The check runs only for a type registered with `numbers.Number`.
+    `Decimal` and `Fraction` carry an exact value that `float` may round
+    - `Decimal(2**63 - 1)` becomes `2**63` - and they compare numerically
+    against `int`, so for them the question is meaningful and the answer
+    is trustworthy.
+
+    Everything else is accepted without asking, because for most
+    `__float__` providers the protocol is the only thing exposed:
+    `float(value)` IS the value and nothing can have been lost. Asking
+    anyway would compare an `int` against an object with no numeric
+    `__eq__`, fall back to identity, and reject every such value while
+    claiming it was unrepresentable - which is how this guard read
+    before, and it was wrong.
+
+    `numbers.Number` is a proxy for "can answer the question", not the
+    thing itself, and it is imprecise in both directions: an exact
+    numeric type that declines to register is not checked, and a
+    non-numeric type that registers without defining `__eq__` is checked
+    and wrongly refused. Gating on `type(value).__eq__` instead trades
+    one of those for a worse one - a wrapper whose `__eq__` returns
+    `False` rather than `NotImplemented` for an `int` goes back to being
+    falsely rejected. The documented contract is therefore scoped to real
+    numeric types rather than to this mechanism.
+
+    A comparison that raises is treated as "cannot tell, accept", so a
+    user-defined `__eq__` cannot throw out of `encode_lane` past its
+    documented contract.
+    """
+    if not isinstance(value, numbers.Number):
+        return False
+    try:
+        # `!=` through operator, not the literal, so mypy does not reject
+        # the intentional cross-type comparison the ABCs make valid.
+        return bool(operator.ne(as_int, value))
+    except Exception:
+        return False
+
+
+def _as_float(value: LaneInput) -> float:
+    """`float(value)`, with a non-numeric `__float__` reported as a TypeError.
+
+    `ArithmeticError` passes through: `float(10**400)` overflows, and that
+    is an honest `OverflowError` about the value rather than a fault in
+    some `__float__` the type does not even have.
+    """
+    try:
+        return float(value)
+    except (TypeError, ValueError, ArithmeticError):
+        raise
+    except Exception as exc:  # a __float__ that raises something else
+        raise TypeError(
+            f"cannot encode {type(value).__name__} {value!r} into a lane: "
+            f"__float__ raised {exc!r}"
+        ) from exc
+
+
+def decode_lane(lane: int, column_type: ColumnType) -> LaneValue:
+    """Unpack one 64-bit lane back into a Python value.
+
+    FLOAT becomes `float`, BOOL becomes `bool`, UINT32 / UINT64 stay
+    unsigned, and everything else is read as a signed 64-bit integer.
+    """
+    if column_type == ColumnType.FLOAT:
+        value: float = _FLOAT.unpack(_BITS.pack(lane))[0]
+        return value
+    if column_type == ColumnType.BOOL:
+        return bool(lane)
+    if column_type in (ColumnType.UINT32, ColumnType.UINT64):
+        return int(lane)
+    return int(ctypes.c_int64(lane).value)
+
+
+def decode_typed_row(row: Any) -> tuple[LaneValue, ...]:
+    """Decode a borrowed `TypedRowStruct` into a tuple of Python values.
+
+    The descriptor and its lane storage belong to wirelog only for the
+    duration of the callback that received them, so this copies every
+    value out before returning.
+    """
+    values: list[LaneValue] = []
+    for c in range(row.logical_ncols):
+        lane = int(row.lanes[row.lane_offsets[c]])
+        values.append(decode_lane(lane, ColumnType(row.types[c])))
+    return tuple(values)
+
+
+__all__ = ["LaneValue", "LaneInput", "encode_lane", "decode_lane", "decode_typed_row"]