Skip to content

Creating an adapter (PlanFrame core)

This guide shows how to implement a PlanFrame adapter for an existing “dataframe-like” engine by implementing BaseAdapter.

What an adapter does

PlanFrame’s core (planframe) is backend-agnostic. It builds a typed plan, then calls an adapter to:

  • compile expressions (compile_expr)
  • execute plan nodes (select, filter, join, …)
  • materialize outputs (collect, to_dicts, to_dict, sink_*/write_*)

The adapter API is the abstract base class:

  • packages/planframe/planframe/backend/adapter.py (BaseAdapter)

For a small, published pass/fail suite you can run in your own CI (recommended for third-party adapters), see Adapter conformance kit.

Third-party adapter integration checklist (PlanFrame 1.3.x)

Use this list when wiring a new engine to PlanFrame. Revisit it when upgrading PlanFrame minors—adapter contracts and interpreter behavior can change (see CHANGELOG.md and Migrating since v1.1.0).

Design

  • [ ] Frame type: choose the backend’s native “frame” or lazy plan type (BackendFrameT).
  • [ ] Expression type: choose the backend’s expression type (BackendExprT) or a small wrapper.
  • [ ] Lazy transforms: return lazy objects from plan nodes; run backend work only inside collect / to_dicts / to_dict / write_* (or documented streaming paths).

BaseAdapter: implement vs override

You must implement every @abstractmethod on BaseAdapter (grouped by role):

  • [ ] Expression lowering: compile_expr
  • [ ] Materialization / export: collect, to_dicts, to_dict (accept options: ExecutionOptions | None on each)
  • [ ] Core transforms: select, project, drop, rename, with_column, cast, with_row_count, filter, sort, unique, duplicated, join, slice, head, tail, concat_vertical, concat_horizontal, pivot
  • [ ] Aggregations / reshaping: group_by_agg, group_by_dynamic_agg, rolling_agg, drop_nulls, fill_null, melt, explode, unnest, posexplode, drop_nulls_all, sample
  • [ ] Sinks (write_*): write_parquet, write_csv, write_ndjson, write_ipc, write_database, write_excel, write_delta, write_avro

You may rely on defaults (override only when needed):

  • reader / writer / areader / awriter: defaults wrap sync read_* / write_* and thread-pool async; override for custom IO surfaces or true async IO.
  • capabilities: defaults to empty flags; set conservatively so PlanFrame can fail fast on unsupported IO or declare advisory async behavior (native_async_materialize).
  • resolve_dtype: default merges ctx.schema with optional ctx.resolve_backend_dtype; override if you need richer dtype recovery for Col(...) during compile_expr.
  • resolve_backend_dtype_from_frame: default returns None; override so execute_plan can populate CompileExprContext.resolve_backend_dtype when the step schema omits a column still present on the backend frame.
  • acollect / ato_dicts / ato_dict: defaults run the sync methods in asyncio.to_thread; override for async-native engines.
  • hint: default no-op; override if the engine supports plan hints.

compile_expr, resolve_dtype, and unknown columns

  • [ ] Implement compile_expr so all Expr IR PlanFrame emits for your supported API surface lowers to BackendExprT.
  • [ ] Decide your policy for unknown column names (shipped adapters are permissive at compile time; see Unknown columns during compile_expr).
  • [ ] If projected step schemas can omit columns that still exist on the evaluated backend frame, implement resolve_backend_dtype_from_frame and/or resolve_dtype so dtypes stay accurate when possible.

Columnar boundaries: to_dict vs planframe.materialize

  • [ ] Implement to_dict (column-oriented) and to_dicts (row-oriented) on the adapter; they are distinct execution boundaries.
  • [ ] In wrapper libraries, prefer materialize_columns / materialize_into (and amaterialize_*) so imports and ExecutionOptions forwarding stay aligned with Frame.to_dict / Frame.ato_dict—see Columnar boundary helpers.

Sync vs async: execute_plan, execute_plan_async, and Frame a* methods

  • [ ] Plan evaluation: execute_plan is synchronous. execute_plan_async runs that interpreter in a worker thread—it does not make individual adapter calls async by itself.
  • [ ] Frame terminals: sync collect_backend / to_dicts / to_dict vs async acollect_backend / ato_dicts / ato_dict (and aliases like collect_async); async paths still build the plan synchronously, then await adapter async materializers.
  • [ ] If you only implement sync collect / to_dicts / to_dict, default acollect / ato_dicts / ato_dict give async API users thread-pooled behavior; override when you need native async I/O.
  • [ ] Do not bypass execute_plan from async entrypoints unless you intentionally opt out of PlanFrame’s execution semantics—see Async execution contract.

Typing (Resolve) and static analysis

  • [ ] For how static column types propagate on Frame / Expr, read Resolve typing design (Pyright-focused). Adapters implement runtime behavior; stubs and Resolve power editor/type-checker UX for host packages.

How plans reach your adapter

Chaining on Frame records a PlanNode tree and updates the derived schema. At materialization (collect, to_dicts, to_dict, write_*, …), PlanFrame runs execute_plan, which walks that tree and invokes the matching BaseAdapter methods. Expression IR is compiled through PlanCompileContext (planframe.compile_context) so the same rules apply when building plans and when executing them. For a file-level map (mixins, dispatch registry, stub location), see Core layout.

A minimal runnable adapter

Below we implement an adapter for a tiny engine that represents a “DataFrame” as list[dict[str, object]].

It’s not fast, but it’s a good template: each adapter method is a pure transformation returning a new “frame”.

Example script

Run:

./.venv/bin/python docs/planframe/guides/examples/rows_adapter_minimal.py

Expected output:

schema=('id', 'age')
collect=[UserRow(id=1, age=10), UserRow(id=2, age=20)]
dicts=[{'id': 1, 'age': 10}, {'id': 2, 'age': 20}]
dict={'id': [1, 2], 'age': [10, 20]}

Optional: row streaming and I/O skins

Beyond the integration checklist:

  • I/O: implement write_* (used by PlanFrame’s sink_*/write_*) or override BaseAdapter.writer with a custom writer implementation.
  • Async I/O (optional): override BaseAdapter.areader / BaseAdapter.awriter for true async IO (defaults wrap sync IO in asyncio.to_thread).
  • Row streaming (optional): implement AdapterRowStreamer (both stream_dicts and astream_dicts are required for detection) to support Frame.stream_dicts() / Frame.astream_dicts() without materializing all rows at once—see Optional: row streaming + true async IO below.

Unknown columns during compile_expr

PlanFrame’s own Frame builder keeps expression IR aligned with each plan step’s input schema, so a normal user plan should not reference a column that is absent from that step’s schema. Custom embedders or experimental paths can still hand adapters IR that disagrees with CompileExprContext.schema.

Policy (shipped adapters: Polars, pandas, sparkless):

  1. BaseAdapter.resolve_dtype(name, ctx=...) is an optional hint hook. A return value of None means “no dtype information”—not “this column is invalid.” The adapter may still lower Col(name) to a backend column reference.
  2. compile_expr is permissive: unknown names are not required to raise at compile time. The backend expression typically refers to the column by name; if the column does not exist on the frame, the engine reports the error at execution (collect, filter evaluation, etc.).
  3. When execute_plan runs, CompileExprContext.resolve_backend_dtype may supply dtypes for names missing from the step schema (see Migrating since v1.1.0 / issue #113). If that callback also returns None, the same permissive rule applies.

Third-party adapters may choose a stricter policy (e.g. raise PlanFrameBackendError from compile_expr when resolve_dtype and resolve_backend_dtype both yield None and the name is not in ctx.schema) if that matches their engine—document that choice clearly.

Debugging: If a filter(...).select(...)-style chain misbehaves, compare the failing step’s input schema (what compile_expr used) to the columns actually present on the backend frame after prior steps; mismatches usually mean plan/schema drift, not silent adapter guessing.

Execution boundaries and ExecutionOptions

Materialization and row export happen only at execution boundaries. On BaseAdapter, these methods take an optional options: ExecutionOptions | None (planframe.execution_options):

Method Role
collect(df, *, options=...) Eager materialization (or no-op for eager backends).
to_dicts(df, *, options=...) Row-oriented export.
to_dict(df, *, options=...) Column-oriented export.
acollect / ato_dicts / ato_dict Async variants; same options= contract (defaults delegate to the sync methods).

ExecutionOptions currently exposes:

  • streaming: user-level streaming hint (meaning is backend-defined).
  • engine_streaming: engine-level streaming hint (distinct from streaming where the backend distinguishes them).

Contract: adapters should accept options on these signatures so the public API stays stable. Forward only the hints your engine understands into the backend’s collect() / export APIs; ignore the rest. If you do not support any hints yet, it is fine to options unused (as many shipped adapters do today), but keep the parameter.

Frame.collect_backend, Frame.to_dicts, Frame.to_dict, and the async counterparts accept the same ExecutionOptions and pass them through to the adapter.

Columnar boundary helpers (planframe.materialize)

End-users often reach Frame.to_dict / Frame.ato_dict first; adapter and wrapper packages are encouraged to use planframe.materialize so imports stay stable and ExecutionOptions forwarding matches the Frame methods below.

Columnar export (options forwarded end-to-end)
  sync:  Frame.to_dict()  ←──  materialize_columns(frame)
  async: Frame.ato_dict() ←──  amaterialize_columns(frame)

For a stable import at the “lazy Frame → columnar dict” step (without pulling in Pydantic or other integrations in adapter code), use:

Helper Behavior
materialize_columns(frame, *, options=...) Same as frame.to_dict(...); forwards ExecutionOptions.
materialize_into(frame, factory, *, options=...) Columnar dict → your factory(dict[str, list[object]]) (Pydantic, dataclass batching, Arrow, etc.).
amaterialize_columns / amaterialize_into Async path (Frame.ato_dict); same options contract.

PlanFrame stays generic: it does not build models—adapters or host libraries supply the callable. Re-exported from from planframe import ... for discoverability.

Example (Polars-backed frame): examples/materialize_boundary_minimal.py (run from repo root with PYTHONPATH=packages/planframe, see the script docstring).

Large results / chunked columnar export: full in-memory dict[str, list[object]] is not always viable. See the design note Columnar streaming (chunked export) and optional AdapterColumnarStreamer in planframe.backend.io (spike; not yet called from materialize_columns).

Async execution contract (third-party adapters)

PlanFrame’s lazy chaining is always synchronous: building a Frame never does I/O and never awaits. Async support exists only at materialization boundaries.

Call graph (what runs on async terminals)

All materialization paths evaluate the plan the same way:

  • Sync: Frame.collect_backend() / Frame.to_dicts() / Frame.to_dict()
  • Plan evaluation: planned = execute_plan(frame.plan, adapter=..., schema=..., options=...) (via Frame._eval)
  • Adapter boundary: adapter.collect(planned, options=...) then adapter.to_dicts(...) / adapter.to_dict(...)
  • Async: Frame.acollect_backend() / Frame.ato_dicts() / Frame.ato_dict() / Frame.acollect()
  • Plan evaluation: still synchronous (planned = frame._eval(frame.plan) on the event loop thread)
  • Adapter boundary: awaits adapter.acollect(planned, options=...) / adapter.ato_dicts(...) / adapter.ato_dict(...)

Implication: async terminals still go through PlanFrame’s interpreter. Third-party adapters should not bypass execute_plan by delegating async paths directly to the underlying engine unless they are intentionally opting out of PlanFrame’s execution/options semantics.

What BaseAdapter must implement for async

  • Minimum: implement the sync boundaries (collect, to_dicts, to_dict) and accept options=....
  • You get async terminals “for free” via BaseAdapter.acollect / ato_dicts / ato_dict, which wrap sync work in asyncio.to_thread.
  • Async-native backends: override acollect and usually also ato_dicts / ato_dict to avoid blocking a worker thread.
  • Keep acollect focused on I/O / engine execution; plan translation should already be completed before it’s called.

Default async behavior: asyncio.to_thread

Location What runs
BaseAdapter.acollect / ato_dicts / ato_dict Defaults wrap the matching sync method (collect, to_dicts, to_dict) in asyncio.to_thread. The synchronous backend work does not run on the event-loop thread.
execute_plan_async Runs the synchronous execute_plan interpreter in a worker thread (asyncio.to_thread), so awaiting it avoids blocking the loop while walking the plan, but each adapter call inside the interpreter remains sync from Python’s perspective.

Combining calls: awaiting execute_plan_async(...) and then await adapter.acollect(...) may use multiple thread hops when the engine is fully synchronous. That still keeps the event loop responsive; it is not end-to-end async I/O unless your adapter overrides the async materializers.

Declaring native async materialization (advisory)

Set AdapterCapabilities.native_async_materialize to True only when your adapter overrides acollect / ato_dicts / ato_dict so the primary backend work uses native async/await (HTTP clients, asyncio database drivers, …) instead of BaseAdapter’s default asyncio.to_thread around sync methods.

Value Meaning for host libraries
False (default) Matches BaseAdapter defaults: async entrypoints may offload sync engine calls to a thread pool.
True The adapter claims async materialization does not rely on those default thread-pooled wrappers for the main backend path.

Contract: PlanFrame’s Frame layer does not read this flag to change behavior today—it is advisory for documentation, UI, or diagnostics. Shipped adapters (Polars, pandas, sparkless) keep the default False.

ExecutionOptions propagation

ExecutionOptions is forwarded to:

  • execute_plan(..., options=...) (plan execution / compilation context)
  • adapter materialization/export methods (collect / to_dicts / to_dict and async variants)

Adapters should treat ExecutionOptions fields as hints: forward only what your engine understands, ignore unknown hints, and keep the parameter on public signatures for forward compatibility.

Thread-safety expectations (important)

If you rely on the default async methods (the asyncio.to_thread wrappers), your adapter’s sync boundaries may be invoked concurrently in multiple worker threads if users run multiple async collections at once.

  • If your backend client / connection object is not thread-safe, either:
  • override async methods to use an async-native client, or
  • serialize internally (locks) / use per-task clients, and document the limitation.

Optional: row streaming + true async IO

If your engine can stream rows (cursor-based DB reads, chunked readers, etc.), implement AdapterRowStreamer on your adapter. Both sync and async entrypoints are required: PlanFrame uses isinstance(adapter, AdapterRowStreamer); an adapter that only defines stream_dicts is treated like a non-streaming adapter and will fall back to to_dicts() / ato_dicts().

from collections.abc import AsyncIterator, Iterator

from planframe.backend.io import AdapterRowStreamer
from planframe.execution_options import ExecutionOptions


class MyAdapter(..., AdapterRowStreamer[MyBackendFrame]):
    def stream_dicts(
        self, df: MyBackendFrame, *, options: ExecutionOptions | None = None
    ) -> Iterator[dict[str, object]]:
        # Yield rows without building a full list
        for row in df.iter_rows():
            yield row

    async def astream_dicts(
        self, df: MyBackendFrame, *, options: ExecutionOptions | None = None
    ) -> AsyncIterator[dict[str, object]]:
        async for row in df.aiter_rows():
            yield row

If your IO layer is truly async (native async HTTP/S3/DB drivers), override BaseAdapter.areader / BaseAdapter.awriter. If you don’t, PlanFrame’s defaults wrap the sync reader/writer via asyncio.to_thread.

Production readiness checklist (adapter authors)

If you plan to ship your adapter as a package used outside a single codebase:

  • Packaging: publish py.typed for typing support; set requires-python; pin compatible dependency ranges.
  • Contracts:
  • Implement ExecutionOptions passthrough on materialization boundaries (collect/exports) even if you ignore hints.
  • If you implement row streaming, implement both stream_dicts and astream_dicts (AdapterRowStreamer contract).
  • Prefer consistent async behavior: if you override acollect, consider also overriding ato_dicts / ato_dict (or ensure the defaults are acceptable).
  • Tests: add regression tests for join semantics, null ordering, and empty-shape exports (to_dict); include async boundary tests if you claim async-native support.
  • Docs: document limitations and backend-specific semantics (e.g. null placement, streaming guarantees, supported joins).

Example: accept and forward hints

from planframe.execution_options import ExecutionOptions

def collect(self, df: MyFrame, *, options: ExecutionOptions | None = None) -> MyFrame:
    kwargs = {}
    if options is not None:
        if options.streaming is not None:
            kwargs["streaming"] = options.streaming
        if options.engine_streaming is not None:
            kwargs["engine_streaming"] = options.engine_streaming
    return df.collect(**kwargs) if kwargs else df.collect()

join

BaseAdapter.join receives left_on and right_on tuples of equal length (for symmetric joins they are identical). For a how="cross" join from Frame.join, both tuples are empty—there are no key columns.

The last argument is optional options: JoinOptions | None (planframe.plan.join_options). JoinOptions fields are execution hints (not relational join semantics). Current fields:

Field Purpose (hint)
coalesce Backend-specific key coalescing.
validate Join validation strategy (backend-defined strings).
join_nulls Whether nulls compare equal in keys.
maintain_order Preserve input order where supported.
streaming User-level streaming / execution style hint.
engine_streaming Engine-level streaming (pairs with ExecutionOptions.engine_streaming conceptually).
allow_parallel Allow parallel join execution.
force_parallel Prefer forcing parallel execution.

Omit-None rule: only pass through kwargs for fields that are not None, so the engine’s defaults apply. Adapters may ignore hints they do not support.

Example: join with hints

from planframe.plan.join_options import JoinOptions

# Call site (conceptual): Frame.join(..., options=JoinOptions(...))
def join(self, left, right, *, left_on, right_on, how="inner", suffix="_right", options=None):
    if options is None:
        return backend_join(left, right, left_on=left_on, right_on=right_on, how=how, suffix=suffix)
    kwargs = {}
    if options.coalesce is not None:
        kwargs["coalesce"] = options.coalesce
    if options.engine_streaming is not None:
        kwargs["engine_streaming"] = options.engine_streaming
    # ... other non-None fields ...
    return backend_join(left, right, left_on=left_on, right_on=right_on, how=how, suffix=suffix, **kwargs)

group_by_agg

Frame.group_by(...).agg(...) lowers to a GroupBy node followed by Agg. Evaluation calls BaseAdapter.group_by_agg on the frame before grouping (the GroupBy predecessor), not on a backend-specific “grouped” handle.

Arguments

  • keys: tuple[CompiledJoinKey[BackendExprT], ...] — same structural type as join / sort keys (CompiledJoinKey is an alias of CompiledSortKey). Each element is either a column name (column= set, expr=None) or a compiled backend expression (column=None, expr= set). PlanFrame compiles JoinKeyExpr IR into the latter. Synthetic key column names in the result schema are __pf_g0, __pf_g1, … matching the index in this tuple.

  • named_aggs: dict[str, tuple[str, str] | BackendExprT] mapping output column names to either:

  • Tuple form: (op, column_name) with op one of count, sum, mean, min, max, n_unique (legacy reductions over a single input column).
  • Compiled expression form: a backend-native expression that is already a per-group aggregation suitable for the engine’s group_by(...).agg(...). PlanFrame produces this by compiling AggExpr IR (agg_sum(inner), agg_mean(inner), …) via compile_expr.

Preserve the iteration order of named_aggs when building the backend aggregation list (Python dict insertion order).

Minimal behavior

If you do not implement grouping yet, keep raising NotImplementedError from group_by_agg with a clear message, as in examples/rows_adapter_minimal.py.

Optional API skins (planframe.spark / planframe.pandas)

End-user packages may subclass a backend Frame and mix in planframe.spark.SparkFrame or planframe.pandas.PandasLikeFrame for familiar naming. That does not change the adapter contract: the plan is still standard Frame nodes, compiled and executed the same way. See PySpark-like API and pandas-like API.

Plan-level hints (Hint node and BaseAdapter.hint)

Plans may include a Hint node (e.g. via SparkFrame.hint(...)). During execution, execute_plan calls:

  • BaseAdapter.hint(self, df, *, hints: tuple[str, ...], kv: dict[str, object]) -> BackendFrameT

The default implementation is a no-op (returns df). If your engine supports broadcast / shuffle / similar hints, override hint and forward only what you understand; ignore the rest.

Notes

  • PlanFrame validates many schema invariants before calling the backend. Your adapter can assume the plan is well-formed, but it should still validate backend-specific constraints (e.g. “pivot requires on_columns when lazy”).