From 3e61e6b7a53fa847c786a763f3e5c29907c82d3f Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Fri, 7 Nov 2025 11:22:04 -0800
Subject: [PATCH 01/17] Capture query plan in pds-h output

Currently, pds-h benchmarks will optionally print out the query plan if
`--explain` is passed. With this PR, we'll also persist the query plan
in the `pdsh_results.jsonl` file. This will facilitate downstream
analysis.

To make it easier for downstream tools to work with, we persist a
structured version of the query plan, rather than the tree-like textual
representation.
---
 .../cudf_polars/cudf_polars/dsl/nodebase.py   |  21 ++
 .../experimental/benchmarks/utils.py          |  47 +++-
 .../cudf_polars/experimental/explain.py       | 260 +++++++++++++++++-
 python/cudf_polars/docs/overview.md           |  45 +++
 .../tests/experimental/test_explain.py        |  67 ++++-
 5 files changed, 425 insertions(+), 15 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/dsl/nodebase.py b/python/cudf_polars/cudf_polars/dsl/nodebase.py
index 23aaf4138a08..7f197a99f564 100644
--- a/python/cudf_polars/cudf_polars/dsl/nodebase.py
+++ b/python/cudf_polars/cudf_polars/dsl/nodebase.py
@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import hashlib
 from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar
 
 if TYPE_CHECKING:
@@ -81,6 +82,26 @@ def get_hashable(self) -> Hashable:
         """
         return (type(self), self._ctor_arguments(self.children))
 
+    def get_stable_id(self) -> int:
+        """
+        Compute a stable identifier for Node.
+
+        Uses MD5 hash of the node's hashable representation for determinism
+        across process boundaries (Python's hash() uses PYTHONHASHSEED).
+
+        Parameters
+        ----------
+        ir_node
+            The IR node.
+
+        Returns
+        -------
+        int
+            A stable 32-bit identifier for this node.
+        """
+        content = repr(self.get_hashable()).encode("utf-8")
+        return int(hashlib.md5(content).hexdigest()[:8], 16)
+
     def __hash__(self) -> int:
         """
         Hash of an expression with caching.
diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils.py
index 9d59650b9688..dc9b211cd6c9 100644
--- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils.py
+++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils.py
@@ -58,6 +58,8 @@
 if TYPE_CHECKING:
     from collections.abc import Callable, Sequence
 
+    from cudf_polars.experimental.explain import DAG
+
 
 try:
     import structlog
@@ -236,6 +238,7 @@ class RunConfig:
         default_factory=PackageVersions.collect
     )
     records: dict[int, list[Record]] = dataclasses.field(default_factory=dict)
+    plans: dict[int, DAG] = dataclasses.field(default_factory=dict)
     dataset_path: Path
     scale_factor: int | float
     shuffle: Literal["rapidsmpf", "tasks"] | None = None
@@ -502,28 +505,37 @@ def print_query_plan(
     args: argparse.Namespace,
     run_config: RunConfig,
     engine: None | pl.GPUEngine = None,
-) -> None:
+    *,
+    print_plans: bool = True,
+) -> tuple[str | None, str | None]:
     """Print the query plan."""
+    logical_plan = plan = None
     if run_config.executor == "cpu":
         if args.explain_logical:
-            print(f"\nQuery {q_id} - Logical plan\n")
-            print(q.explain())
+            logical_plan = q.explain()
         if args.explain:
-            print(f"\nQuery {q_id} - Physical plan\n")
-            print(q.show_graph(engine="streaming", plan_stage="physical"))
+            plan = q.show_graph(engine="streaming", plan_stage="physical")
     elif CUDF_POLARS_AVAILABLE:
         assert isinstance(engine, pl.GPUEngine)
         if args.explain_logical:
-            print(f"\nQuery {q_id} - Logical plan\n")
-            print(explain_query(q, engine, physical=False))
+            logical_plan = explain_query(q, engine, physical=False)
         if args.explain and run_config.executor == "streaming":
-            print(f"\nQuery {q_id} - Physical plan\n")
-            print(explain_query(q, engine))
+            plan = explain_query(q, engine)
     else:
         raise RuntimeError(
             "Cannot provide the logical or physical plan because cudf_polars is not installed."
         )
 
+    if print_plans:
+        if logical_plan:
+            print(f"\nQuery {q_id} - Logical plan\n")
+            print(logical_plan)
+        if plan:
+            print(f"\nQuery {q_id} - Physical plan\n")
+            print(plan)
+
+    return logical_plan, plan
+
 
 def initialize_dask_cluster(run_config: RunConfig, args: argparse.Namespace):  # type: ignore[no-untyped-def]
     """
@@ -942,6 +954,12 @@ def parse_args(
         help="Print an outline of the logical plan",
         default=False,
     )
+    parser.add_argument(
+        "--print-plans",
+        action=argparse.BooleanOptionalAction,
+        help="Print the query plans",
+        default=True,
+    )
     parser.add_argument(
         "--validate",
         action=argparse.BooleanOptionalAction,
@@ -1036,6 +1054,7 @@ def run_polars(
         run_config = dataclasses.replace(run_config, n_workers=actual_n_workers)
 
     records: defaultdict[int, list[Record]] = defaultdict(list)
+    plans: dict[int, DAG] = {}
     engine: pl.GPUEngine | None = None
 
     if run_config.executor != "cpu":
@@ -1063,7 +1082,13 @@ def run_polars(
         except AttributeError as err:
             raise NotImplementedError(f"Query {q_id} not implemented.") from err
 
-        print_query_plan(q_id, q, args, run_config, engine)
+        print_query_plan(
+            q_id, q, args, run_config, engine, print_plans=args.print_plans
+        )
+        if (args.explain or args.explain_logical) and engine is not None:
+            from cudf_polars.experimental.explain import serialize_query
+
+            plans[q_id] = serialize_query(q, engine)
 
         records[q_id] = []
         for i in range(args.iterations):
@@ -1124,7 +1149,7 @@ def run_polars(
             )
             records[q_id].append(record)
 
-    run_config = dataclasses.replace(run_config, records=dict(records))
+    run_config = dataclasses.replace(run_config, records=dict(records), plans=plans)
 
     # consolidate logs
     if _HAS_STRUCTLOG and run_config.collect_traces:
diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index 17cba1d17cb4..577027cbe204 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: Apache-2.0
@@ -6,17 +6,23 @@
 
 from __future__ import annotations
 
+import dataclasses
 import functools
+from collections.abc import Mapping, Sequence
 from itertools import groupby
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Self, TypeAlias
 
 from cudf_polars.dsl.ir import (
+    Filter,
     GroupBy,
+    HStack,
     Join,
     Scan,
+    Select,
     Sort,
 )
 from cudf_polars.dsl.translate import Translator
+from cudf_polars.dsl.traversal import traversal
 from cudf_polars.experimental.base import ColumnStat
 from cudf_polars.experimental.parallel import lower_ir_graph
 from cudf_polars.experimental.statistics import (
@@ -33,6 +39,17 @@
     from cudf_polars.experimental.base import PartitionInfo, StatsCollector
 
 
+Serializable: TypeAlias = (
+    str
+    | int
+    | float
+    | bool
+    | Sequence["Serializable"]
+    | Mapping[str, "Serializable"]
+    | None
+)
+
+
 def explain_query(
     q: pl.LazyFrame,
     engine: pl.GPUEngine,
@@ -81,6 +98,44 @@ def explain_query(
             return _repr_ir_tree(ir)
 
 
+def serialize_query(
+    q: pl.LazyFrame,
+    engine: pl.GPUEngine,
+    *,
+    physical: bool = True,
+) -> DAG:
+    """
+    Return a structured, serializable representation of the IR plan.
+
+    Parameters
+    ----------
+    q : pl.LazyFrame
+        The LazyFrame to serialize.
+    engine : pl.GPUEngine
+        The configured GPU engine to use.
+    physical : bool, default True
+        If True, serialize the physical (lowered) plan with partition info.
+        If False, serialize the logical (pre-lowering) plan.
+
+    Returns
+    -------
+    DAG
+        A structured DAG representation of the query plan that can be
+        serialized to JSON.
+
+    Examples
+    --------
+    >>> import polars as pl
+    >>> import json
+    >>> q = pl.LazyFrame({"a": [1, 2, 3]}).select("a")
+    >>> engine = pl.GPUEngine(executor="streaming")
+    >>> dag = serialize_query(q, engine, physical=False)
+    >>> json.dumps(dag.to_dict())  # doctest: +SKIP
+    '{"roots": [...], "nodes": {...}, "partition_info": null}'
+    """
+    return DAG.from_query(q, engine, lowered=physical)
+
+
 def _fmt_row_count(value: int | None) -> str:
     """Format a row count as a readable string."""
     if value is None:
@@ -168,3 +223,204 @@ def _(ir: Sort, *, offset: str = "") -> str:
 def _(ir: Scan, *, offset: str = "") -> str:
     label = f"SCAN {ir.typ.upper()}"
     return _repr_header(offset, label, ir.schema)
+
+
+# --------------------------------------------------------------------------
+# Property serialization for structured query plan export
+# --------------------------------------------------------------------------
+
+
+@functools.singledispatch
+def _serialize_properties(ir: IR) -> dict[str, Serializable]:
+    """Extract serializable properties from an IR node."""
+    return {}
+
+
+@_serialize_properties.register
+def _(ir: Scan) -> dict[str, Serializable]:
+    return {
+        "typ": ir.typ,
+        "paths": ir.paths,
+    }
+
+
+@_serialize_properties.register
+def _(ir: Join) -> dict[str, Serializable]:
+    return {
+        "how": ir.options[0],
+        "left_on": [ne.name for ne in ir.left_on],
+        "right_on": [ne.name for ne in ir.right_on],
+    }
+
+
+@_serialize_properties.register
+def _(ir: GroupBy) -> dict[str, Serializable]:
+    return {
+        "keys": [ne.name for ne in ir.keys],
+    }
+
+
+@_serialize_properties.register
+def _(ir: Sort) -> dict[str, Serializable]:
+    return {
+        "by": [ne.name for ne in ir.by],
+        "order": [str(o) for o in ir.order],
+    }
+
+
+@_serialize_properties.register
+def _(ir: Filter) -> dict[str, Serializable]:
+    return {
+        "predicate": ir.mask.name,
+    }
+
+
+@_serialize_properties.register
+def _(ir: Select) -> dict[str, Serializable]:
+    return {
+        "columns": [ne.name for ne in ir.exprs],
+    }
+
+
+@_serialize_properties.register
+def _(ir: HStack) -> dict[str, Serializable]:
+    return {
+        "columns": [ne.name for ne in ir.columns],
+    }
+
+
+@dataclasses.dataclass
+class SerializableIRNode:
+    """
+    A node in the DAG.
+
+    This node is *serializable* and cannot be executed like a
+    cudf_polars.dsl.ir.IR node.
+    """
+
+    id: int
+    children: list[int]
+    schema: dict[str, Serializable]
+    properties: dict[str, Serializable]
+    type: str
+
+    @classmethod
+    def from_ir(cls, ir: IR) -> Self:
+        """Build a Node from an IR Node."""
+        return cls(
+            id=ir.get_stable_id(),
+            children=[child.get_stable_id() for child in ir.children],
+            schema={k: v.id().name for k, v in ir.schema.items()},
+            properties=_serialize_properties(ir),
+            type=type(ir).__name__,
+        )
+
+    def to_dict(self) -> dict[str, Serializable]:
+        """Convert to a JSON-serializable dictionary."""
+        return {
+            "id": self.id,
+            "children": self.children,
+            "schema": self.schema,
+            "properties": self.properties,
+            "type": self.type,
+        }
+
+
+@dataclasses.dataclass
+class SerializablePartitionInfo:
+    """Serializable information about a partition."""
+
+    count: int
+    partitioned_on: tuple[Serializable, ...]
+
+    def to_dict(self) -> dict[str, Serializable]:
+        """Convert to a JSON-serializable dictionary."""
+        return {
+            "count": self.count,
+            "partitioned_on": list(self.partitioned_on),
+        }
+
+
+@dataclasses.dataclass
+class DAG:
+    """A DAG of nodes."""
+
+    roots: list[int]
+    nodes: dict[int, SerializableIRNode]
+    partition_info: dict[int, SerializablePartitionInfo] | None = None
+
+    @classmethod
+    def from_query(
+        cls,
+        q: pl.LazyFrame,
+        engine: pl.GPUEngine,
+        *,
+        lowered: bool = False,
+    ) -> Self:
+        """
+        Build a DAG from a LazyFrame query.
+
+        Parameters
+        ----------
+        q
+            The LazyFrame to serialize.
+        engine
+            The GPU engine to use. If None, uses default streaming executor.
+        lowered
+            If True, lower the IR to the physical plan and include partition info.
+
+        Returns
+        -------
+        DAG
+            A serializable DAG representation of the query plan.
+        """
+        config = ConfigOptions.from_polars_engine(engine)
+        ir = Translator(q._ldf.visit(), engine).translate_ir()
+
+        partition_info_dict: dict[int, SerializablePartitionInfo] | None = None
+        if lowered:
+            if (
+                config.executor.name == "streaming"
+                and config.executor.runtime == "rapidsmpf"
+            ):
+                from cudf_polars.experimental.rapidsmpf.core import (
+                    lower_ir_graph as rapidsmpf_lower_ir_graph,
+                )
+
+                ir, partition_info_d, _ = rapidsmpf_lower_ir_graph(ir, config)
+            else:
+                ir, partition_info_d, _ = lower_ir_graph(ir, config)
+            partition_info_dict = {}
+
+        nodes: dict[int, SerializableIRNode] = {}
+        for ir_node in traversal([ir]):
+            stable_id = ir_node.get_stable_id()
+            nodes[stable_id] = SerializableIRNode.from_ir(ir_node)
+            if lowered and partition_info_dict is not None:
+                partition_info_dict[stable_id] = SerializablePartitionInfo(
+                    count=partition_info_d[ir_node].count,
+                    partitioned_on=tuple(
+                        expr.name for expr in partition_info_d[ir_node].partitioned_on
+                    ),
+                )
+
+        return cls(
+            roots=[ir.get_stable_id()], nodes=nodes, partition_info=partition_info_dict
+        )
+
+    def to_dict(self) -> dict[str, Serializable]:
+        """Convert to a JSON-serializable dictionary."""
+        return {
+            "roots": self.roots,
+            "nodes": {
+                str(node_id): node.to_dict() for node_id, node in self.nodes.items()
+            },
+            "partition_info": (
+                {
+                    str(node_id): info.to_dict()
+                    for node_id, info in self.partition_info.items()
+                }
+                if self.partition_info is not None
+                else None
+            ),
+        }
diff --git a/python/cudf_polars/docs/overview.md b/python/cudf_polars/docs/overview.md
index 1814ebc0c393..b0f922cbab9e 100644
--- a/python/cudf_polars/docs/overview.md
+++ b/python/cudf_polars/docs/overview.md
@@ -792,3 +792,48 @@ These provide a higher-level grouping over the lower-level libcudf calls (e.g.
 Finally, if using [rapidsmpf](https://docs.rapids.ai/api/rapidsmpf/nightly/)
 for shuffling, the methods inserting and extracting partitions to shuffle are
 annotated with nvtx ranges.
+
+# Query Plans
+
+The module `cudf_polars.experimental.explain` contains functions for dumping
+the query for a given `LazyFrame`.
+
+
+## Structured Output
+
+`cudf_polars.experimental.explain.serialize_query` can be used to output
+the query plan in a structured format.
+
+```python
+>>> import polars as pl
+>>> from cudf_polars.experimental.explain import serialize_query
+>>> q = pl.LazyFrame({"a": ['a', 'b', 'a'], "b": [1, 2, 3]}).group_by("a").agg(pl.len())
+serialize_query(df, engine=pl.GPUEngine())
+{'roots': [2376323819],
+ 'nodes': {'2376323819': {'id': 2376323819,
+   'children': [1086359873],
+   'schema': {'a': 'STRING', 'len': 'UINT32'},
+   'properties': {'columns': ['a', 'len']},
+   'type': 'Select'},
+  '1086359873': {'id': 1086359873,
+   'children': [2102236744],
+   'schema': {'a': 'STRING', '___0': 'UINT32'},
+   'properties': {'keys': ['a']},
+   'type': 'GroupBy'},
+  '2102236744': {'id': 2102236744,
+   'children': [],
+   'schema': {'a': 'STRING'},
+   'properties': {},
+   'type': 'DataFrameScan'}},
+ 'partition_info': {'2376323819': {'count': 1, 'partitioned_on': []},
+  '1086359873': {'count': 1, 'partitioned_on': []},
+  '2102236744': {'count': 1, 'partitioned_on': []}}}
+```
+
+The structured schema has three top-level fields:
+
+1. `roots`: the integer ID for the "root" (final) nodes in the query plan
+2. `partition_info`: partitioning information at each stage of the query
+3. `nodes`: A mapping from integer node id to node details. Each node ID
+   that appears in the output will be present in this mapping.
+   Inspect `children` to understand which nodes this node depends on.
diff --git a/python/cudf_polars/tests/experimental/test_explain.py b/python/cudf_polars/tests/experimental/test_explain.py
index 19b14a545618..6f9225fa7926 100644
--- a/python/cudf_polars/tests/experimental/test_explain.py
+++ b/python/cudf_polars/tests/experimental/test_explain.py
@@ -1,15 +1,20 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: Apache-2.0
 
 from __future__ import annotations
 
+import json
 import re
 
 import pytest
 
 import polars as pl
 
-from cudf_polars.experimental.explain import _fmt_row_count, explain_query
+from cudf_polars.experimental.explain import (
+    _fmt_row_count,
+    explain_query,
+    serialize_query,
+)
 from cudf_polars.testing.asserts import (
     DEFAULT_CLUSTER,
     DEFAULT_RUNTIME,
@@ -495,3 +500,61 @@ def _gb(df):
         assert re.search(
             rf"^\s*SORT.*row_count=\'~{final_count_2}\'\s*$", repr, re.MULTILINE
         )
+
+
+def test_serialize_query():
+    left = pl.LazyFrame({"a": ["a", "b", "a"], "b": [1, 2, 3]})
+    right = pl.LazyFrame({"a": ["a", "b", "c"], "c": [4, 5, 6]})
+
+    q = (
+        left.join(right, on="a", how="inner")
+        .group_by("a")
+        .agg(pl.col("b").sum(), pl.col("c").max())
+    )
+    engine = pl.GPUEngine(executor="streaming")
+    dag = serialize_query(q, engine)
+
+    # We don't know the exact node IDs, but we can check the structure.
+    assert len(dag.roots) == 1
+    assert len(dag.nodes) == 5
+    assert len(dag.partition_info) == 5
+    node_ids = set(dag.nodes)
+
+    for node_id, node in dag.nodes.items():
+        assert node_id in node_ids
+        assert node.id in node_ids
+        assert set(node.children) <= node_ids
+
+        match node.type:
+            case "DataFrameScan":
+                assert node.children == []
+                assert node.schema == {"a": "STRING", "b": "INT64"} or node.schema == {
+                    "a": "STRING",
+                    "c": "INT64",
+                }
+                assert node_id not in dag.roots
+
+            case "GroupBy":
+                assert len(node.children) == 1
+                assert node.schema == {"a": "STRING", "b": "INT64", "c": "INT64"}
+                assert node.properties == {"keys": ["a"]}
+                assert node_id not in dag.roots
+
+            case "Select":
+                assert len(node.children) == 1
+                assert node.schema == {"a": "STRING", "b": "INT64", "c": "INT64"}
+                assert node.properties == {"columns": ["a", "b", "c"]}
+                assert node_id in dag.roots
+
+            case "Join":
+                assert len(node.children) == 2
+                assert node.schema == {"a": "STRING", "b": "INT64", "c": "INT64"}
+                assert node.properties == {
+                    "how": "Inner",
+                    "left_on": ["a"],
+                    "right_on": ["a"],
+                }
+                assert node_id not in dag.roots
+
+    # smoke test to ensure that the output is JSON serializable
+    json.dumps(dag.to_dict())

From aed6708cd6db009b254edbab0e4dfb8858e7efd8 Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Tue, 3 Feb 2026 14:57:09 -0800
Subject: [PATCH 02/17] use string IDs

---
 .../cudf_polars/experimental/explain.py       | 37 ++++++-------------
 1 file changed, 11 insertions(+), 26 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index 577027cbe204..baf44905a9f5 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -298,7 +298,7 @@ class SerializableIRNode:
     cudf_polars.dsl.ir.IR node.
     """
 
-    id: int
+    id: str
     children: list[int]
     schema: dict[str, Serializable]
     properties: dict[str, Serializable]
@@ -308,7 +308,7 @@ class SerializableIRNode:
     def from_ir(cls, ir: IR) -> Self:
         """Build a Node from an IR Node."""
         return cls(
-            id=ir.get_stable_id(),
+            id=str(ir.get_stable_id()),
             children=[child.get_stable_id() for child in ir.children],
             schema={k: v.id().name for k, v in ir.schema.items()},
             properties=_serialize_properties(ir),
@@ -345,9 +345,9 @@ def to_dict(self) -> dict[str, Serializable]:
 class DAG:
     """A DAG of nodes."""
 
-    roots: list[int]
-    nodes: dict[int, SerializableIRNode]
-    partition_info: dict[int, SerializablePartitionInfo] | None = None
+    roots: list[str]
+    nodes: dict[str, SerializableIRNode]
+    partition_info: dict[str, SerializablePartitionInfo] | None = None
 
     @classmethod
     def from_query(
@@ -377,7 +377,7 @@ def from_query(
         config = ConfigOptions.from_polars_engine(engine)
         ir = Translator(q._ldf.visit(), engine).translate_ir()
 
-        partition_info_dict: dict[int, SerializablePartitionInfo] | None = None
+        partition_info_dict: dict[str, SerializablePartitionInfo] | None = None
         if lowered:
             if (
                 config.executor.name == "streaming"
@@ -392,9 +392,9 @@ def from_query(
                 ir, partition_info_d, _ = lower_ir_graph(ir, config)
             partition_info_dict = {}
 
-        nodes: dict[int, SerializableIRNode] = {}
+        nodes: dict[str, SerializableIRNode] = {}
         for ir_node in traversal([ir]):
-            stable_id = ir_node.get_stable_id()
+            stable_id = str(ir_node.get_stable_id())
             nodes[stable_id] = SerializableIRNode.from_ir(ir_node)
             if lowered and partition_info_dict is not None:
                 partition_info_dict[stable_id] = SerializablePartitionInfo(
@@ -405,22 +405,7 @@ def from_query(
                 )
 
         return cls(
-            roots=[ir.get_stable_id()], nodes=nodes, partition_info=partition_info_dict
+            roots=[str(ir.get_stable_id())],
+            nodes=nodes,
+            partition_info=partition_info_dict,
         )
-
-    def to_dict(self) -> dict[str, Serializable]:
-        """Convert to a JSON-serializable dictionary."""
-        return {
-            "roots": self.roots,
-            "nodes": {
-                str(node_id): node.to_dict() for node_id, node in self.nodes.items()
-            },
-            "partition_info": (
-                {
-                    str(node_id): info.to_dict()
-                    for node_id, info in self.partition_info.items()
-                }
-                if self.partition_info is not None
-                else None
-            ),
-        }

From 66e3356eb7da6d6a81babc3769ad41d642595ce8 Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Tue, 3 Feb 2026 15:02:07 -0800
Subject: [PATCH 03/17] fix

---
 python/cudf_polars/docs/overview.md | 21 +++++++++++----------
 1 file changed, 11 insertions(+), 10 deletions(-)

diff --git a/python/cudf_polars/docs/overview.md b/python/cudf_polars/docs/overview.md
index b0f922cbab9e..b6e49cae956b 100644
--- a/python/cudf_polars/docs/overview.md
+++ b/python/cudf_polars/docs/overview.md
@@ -805,29 +805,30 @@ the query for a given `LazyFrame`.
 the query plan in a structured format.
 
 ```python
+>>> import dataclasses
 >>> import polars as pl
 >>> from cudf_polars.experimental.explain import serialize_query
 >>> q = pl.LazyFrame({"a": ['a', 'b', 'a'], "b": [1, 2, 3]}).group_by("a").agg(pl.len())
-serialize_query(df, engine=pl.GPUEngine())
-{'roots': [2376323819],
- 'nodes': {'2376323819': {'id': 2376323819,
-   'children': [1086359873],
+>>> dataclasses.asdict(serialize_query(q, engine=pl.GPUEngine()))
+{'roots': ['2408237918'],
+ 'nodes': {'2408237918': {'id': '2408237918',
+   'children': [1009833075],
    'schema': {'a': 'STRING', 'len': 'UINT32'},
    'properties': {'columns': ['a', 'len']},
    'type': 'Select'},
-  '1086359873': {'id': 1086359873,
-   'children': [2102236744],
+  '1009833075': {'id': '1009833075',
+   'children': [3262131978],
    'schema': {'a': 'STRING', '___0': 'UINT32'},
    'properties': {'keys': ['a']},
    'type': 'GroupBy'},
-  '2102236744': {'id': 2102236744,
+  '3262131978': {'id': '3262131978',
    'children': [],
    'schema': {'a': 'STRING'},
    'properties': {},
    'type': 'DataFrameScan'}},
- 'partition_info': {'2376323819': {'count': 1, 'partitioned_on': []},
-  '1086359873': {'count': 1, 'partitioned_on': []},
-  '2102236744': {'count': 1, 'partitioned_on': []}}}
+ 'partition_info': {'2408237918': {'count': 1, 'partitioned_on': ()},
+  '1009833075': {'count': 1, 'partitioned_on': ()},
+  '3262131978': {'count': 1, 'partitioned_on': ()}}}
 ```
 
 The structured schema has three top-level fields:

From 01a42ced1f16263a978579e020b3dfdf21e4923c Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Tue, 3 Feb 2026 15:04:15 -0800
Subject: [PATCH 04/17] fixes

---
 python/cudf_polars/cudf_polars/experimental/explain.py | 4 ++--
 python/cudf_polars/tests/experimental/test_explain.py  | 3 ++-
 2 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index baf44905a9f5..b4849cbd15e7 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -299,7 +299,7 @@ class SerializableIRNode:
     """
 
     id: str
-    children: list[int]
+    children: list[str]
     schema: dict[str, Serializable]
     properties: dict[str, Serializable]
     type: str
@@ -309,7 +309,7 @@ def from_ir(cls, ir: IR) -> Self:
         """Build a Node from an IR Node."""
         return cls(
             id=str(ir.get_stable_id()),
-            children=[child.get_stable_id() for child in ir.children],
+            children=[str(child.get_stable_id()) for child in ir.children],
             schema={k: v.id().name for k, v in ir.schema.items()},
             properties=_serialize_properties(ir),
             type=type(ir).__name__,
diff --git a/python/cudf_polars/tests/experimental/test_explain.py b/python/cudf_polars/tests/experimental/test_explain.py
index 6f9225fa7926..f7ba6712a338 100644
--- a/python/cudf_polars/tests/experimental/test_explain.py
+++ b/python/cudf_polars/tests/experimental/test_explain.py
@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import json
 import re
 
@@ -557,4 +558,4 @@ def test_serialize_query():
                 assert node_id not in dag.roots
 
     # smoke test to ensure that the output is JSON serializable
-    json.dumps(dag.to_dict())
+    json.dumps(dataclasses.asdict(dag))

From fd6a52c56e8b0c33a5e4c2aa1c5814a9bd3cf9a1 Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Wed, 4 Feb 2026 05:42:42 -0800
Subject: [PATCH 05/17] test debugging

---
 python/cudf_polars/tests/experimental/test_explain.py | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/python/cudf_polars/tests/experimental/test_explain.py b/python/cudf_polars/tests/experimental/test_explain.py
index f7ba6712a338..3d94fa8e1385 100644
--- a/python/cudf_polars/tests/experimental/test_explain.py
+++ b/python/cudf_polars/tests/experimental/test_explain.py
@@ -512,12 +512,15 @@ def test_serialize_query():
         .group_by("a")
         .agg(pl.col("b").sum(), pl.col("c").max())
     )
-    engine = pl.GPUEngine(executor="streaming")
+    engine = pl.GPUEngine(executor="streaming", raise_on_fail=True)
     dag = serialize_query(q, engine)
 
     # We don't know the exact node IDs, but we can check the structure.
     assert len(dag.roots) == 1
-    assert len(dag.nodes) == 5
+    node_types = sorted({x.type for x in dag.nodes.values()})
+    assert node_types == ["DataFrameScan", "GroupBy", "Join", "Select"]
+    # On some systems (CI), this can apparently be length 6.
+    assert len(dag.nodes) >= 5
     assert len(dag.partition_info) == 5
     node_ids = set(dag.nodes)
 

From 748731fe3c237367f747b62bcc1d1aecdcc3442b Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Wed, 4 Feb 2026 07:04:17 -0800
Subject: [PATCH 06/17] polars compat

---
 .../tests/experimental/test_explain.py         | 18 +++++++++++++-----
 1 file changed, 13 insertions(+), 5 deletions(-)

diff --git a/python/cudf_polars/tests/experimental/test_explain.py b/python/cudf_polars/tests/experimental/test_explain.py
index 3d94fa8e1385..ff50d5298b55 100644
--- a/python/cudf_polars/tests/experimental/test_explain.py
+++ b/python/cudf_polars/tests/experimental/test_explain.py
@@ -504,6 +504,10 @@ def _gb(df):
 
 
 def test_serialize_query():
+    # this test is sensitive to the polars version.
+    # we get a different query plan for polars < 1.35.0.
+    pytest.importorskip("polars", minversion="1.35.0")
+
     left = pl.LazyFrame({"a": ["a", "b", "a"], "b": [1, 2, 3]})
     right = pl.LazyFrame({"a": ["a", "b", "c"], "c": [4, 5, 6]})
 
@@ -518,15 +522,14 @@ def test_serialize_query():
     # We don't know the exact node IDs, but we can check the structure.
     assert len(dag.roots) == 1
     node_types = sorted({x.type for x in dag.nodes.values()})
-    assert node_types == ["DataFrameScan", "GroupBy", "Join", "Select"]
-    # On some systems (CI), this can apparently be length 6.
-    assert len(dag.nodes) >= 5
-    assert len(dag.partition_info) == 5
+    assert node_types == ["DataFrameScan", "GroupBy", "Join", "Projection", "Select"]
+    assert len(dag.nodes) == 6
+    assert len(dag.partition_info) == 6
     node_ids = set(dag.nodes)
 
     for node_id, node in dag.nodes.items():
+        assert node.id == node_id
         assert node_id in node_ids
-        assert node.id in node_ids
         assert set(node.children) <= node_ids
 
         match node.type:
@@ -538,6 +541,11 @@ def test_serialize_query():
                 }
                 assert node_id not in dag.roots
 
+            case "Projection":
+                assert len(node.children) == 1
+                assert node.schema == {"b": "INT64", "c": "INT64", "a": "STRING"}
+                assert node.properties == {}
+
             case "GroupBy":
                 assert len(node.children) == 1
                 assert node.schema == {"a": "STRING", "b": "INT64", "c": "INT64"}

From 53f30b1317c88b2259ffa6c15c2b54c85f8546ce Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Wed, 4 Feb 2026 09:24:52 -0800
Subject: [PATCH 07/17] cache stable id

---
 python/cudf_polars/cudf_polars/dsl/nodebase.py | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/dsl/nodebase.py b/python/cudf_polars/cudf_polars/dsl/nodebase.py
index 7f197a99f564..6453c798f7fc 100644
--- a/python/cudf_polars/cudf_polars/dsl/nodebase.py
+++ b/python/cudf_polars/cudf_polars/dsl/nodebase.py
@@ -34,8 +34,9 @@ class Node(Generic[T]):
     *children).``
     """
 
-    __slots__ = ("_hash_value", "_repr_value", "children")
+    __slots__ = ("_hash_value", "_repr_value", "_stable_hash_value", "children")
     _hash_value: int
+    _stable_hash_value: int
     _repr_value: str
     children: tuple[T, ...]
     _non_child: ClassVar[tuple[str, ...]] = ()
@@ -99,8 +100,12 @@ def get_stable_id(self) -> int:
         int
             A stable 32-bit identifier for this node.
         """
-        content = repr(self.get_hashable()).encode("utf-8")
-        return int(hashlib.md5(content).hexdigest()[:8], 16)
+        try:
+            return self._stable_hash_value
+        except AttributeError:
+            content = repr(self.get_hashable()).encode("utf-8")
+            self._stable_hash_value = int(hashlib.md5(content).hexdigest()[:8], 16)
+            return self._stable_hash_value
 
     def __hash__(self) -> int:
         """

From 5924b82e3d25ec96f1295623f1c6a06cfd1e8437 Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Wed, 4 Feb 2026 09:45:17 -0800
Subject: [PATCH 08/17] test coverage

---
 .../cudf_polars/experimental/explain.py       | 28 +++++++-
 .../tests/experimental/test_explain.py        | 64 +++++++++++++++++++
 2 files changed, 90 insertions(+), 2 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index b4849cbd15e7..162e9a6e5305 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -264,16 +264,40 @@ def _(ir: GroupBy) -> dict[str, Serializable]:
 def _(ir: Sort) -> dict[str, Serializable]:
     return {
         "by": [ne.name for ne in ir.by],
-        "order": [str(o) for o in ir.order],
+        "order": [o.name for o in ir.order],
     }
 
 
 @_serialize_properties.register
 def _(ir: Filter) -> dict[str, Serializable]:
-    return {
+    import cudf_polars.dsl.expressions.binaryop
+    import cudf_polars.dsl.expressions.literal
+
+    value = ir.mask.value
+    properties: dict[str, Serializable] = {
         "predicate": ir.mask.name,
     }
 
+    match value:
+        case cudf_polars.dsl.expressions.binaryop.BinOp():
+            properties["op"] = value.op.name
+            match value.children[0]:
+                case cudf_polars.dsl.expressions.base.Col(name=name):
+                    properties["left"] = {"type": "Col", "name": name}
+                case cudf_polars.dsl.expressions.literal.Literal(value=value):
+                    properties["left"] = {"type": "Literal", "value": value}
+                case _:  # pragma: no cover
+                    properties["left"] = {"type": type(value.children[0]).__name__}
+            match value.children[1]:
+                case cudf_polars.dsl.expressions.base.Col(name=name):
+                    properties["right"] = {"type": "Col", "name": name}
+                case cudf_polars.dsl.expressions.literal.Literal(value=value):
+                    properties["right"] = {"type": "Literal", "value": value}
+                case _:  # pragma: no cover
+                    properties["left"] = {"type": type(value.children[0]).__name__}
+
+    return properties
+
 
 @_serialize_properties.register
 def _(ir: Select) -> dict[str, Serializable]:
diff --git a/python/cudf_polars/tests/experimental/test_explain.py b/python/cudf_polars/tests/experimental/test_explain.py
index ff50d5298b55..9ee7cd2f4e50 100644
--- a/python/cudf_polars/tests/experimental/test_explain.py
+++ b/python/cudf_polars/tests/experimental/test_explain.py
@@ -6,6 +6,7 @@
 import dataclasses
 import json
 import re
+from typing import TYPE_CHECKING
 
 import pytest
 
@@ -23,6 +24,9 @@
 )
 from cudf_polars.testing.io import make_lazy_frame, make_partitioned_source
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
 
 @pytest.fixture(scope="module")
 def df():
@@ -570,3 +574,63 @@ def test_serialize_query():
 
     # smoke test to ensure that the output is JSON serializable
     json.dumps(dataclasses.asdict(dag))
+
+
+def test_scan_properties(tmp_path: Path):
+    pl.DataFrame({"a": [1, 2, 3]}).write_parquet(tmp_path / "test.parquet")
+
+    q = pl.scan_parquet(tmp_path / "test.parquet")
+    engine = pl.GPUEngine(executor="streaming", raise_on_fail=True)
+    dag = serialize_query(q, engine)
+
+    # walk Union -> Scan
+    node = dag.nodes[dag.nodes[dag.roots[0]].children[0]]
+    assert node.type == "Scan"
+    assert node.properties == {
+        "paths": [str(tmp_path / "test.parquet")],
+        "typ": "parquet",
+    }
+
+
+@pytest.mark.parametrize("descending", [False, True])
+def test_sort_properties(*, descending: bool):
+    q = pl.LazyFrame({"a": [1, 3, 2]}).sort("a", descending=descending)
+    dag = serialize_query(q, pl.GPUEngine(executor="streaming"))
+
+    order = "DESCENDING" if descending else "ASCENDING"
+    node = dag.nodes[dag.roots[0]]
+    assert node.type == "Sort"
+    assert node.properties == {"by": ["a"], "order": [order]}
+
+
+def test_filter_properties():
+    q = pl.LazyFrame({"a": [1, 2, 3]}).filter(pl.col("a") > 1)
+    dag = serialize_query(q, pl.GPUEngine(executor="streaming"))
+
+    node = dag.nodes[dag.roots[0]]
+    assert node.type == "Filter"
+    assert node.properties == {
+        "predicate": "a",
+        "op": "GREATER",
+        "left": {"type": "Col", "name": "a"},
+        "right": {"type": "Literal", "value": 1},
+    }
+
+
+def test_select_properties():
+    q = pl.LazyFrame({"a": [1, 2, 3]}).select(pl.col("a") + 1)
+    dag = serialize_query(q, pl.GPUEngine(executor="streaming"))
+
+    node = dag.nodes[dag.roots[0]]
+    assert node.type == "Select"
+    assert node.properties == {"columns": ["a"]}
+
+
+def test_hstack_properties():
+    left = pl.LazyFrame({"a": [1, 2, 3]})
+    q = left.with_columns(pl.col("a"), (pl.col("a") + 1).alias("b"))
+    dag = serialize_query(q, pl.GPUEngine(executor="streaming"))
+
+    node = dag.nodes[dag.roots[0]]
+    assert node.type == "HStack"
+    assert node.properties == {"columns": ["a", "b"]}

From 89f34ddfea0f49531fa9788fe54107526788921f Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Wed, 4 Feb 2026 14:18:18 -0800
Subject: [PATCH 09/17] testing

---
 .../cudf_polars/experimental/explain.py       | 22 ++++++++++--
 .../tests/experimental/test_explain.py        | 34 ++++++++++++++-----
 2 files changed, 45 insertions(+), 11 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index 162e9a6e5305..3dcf89bd7b65 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -12,6 +12,8 @@
 from itertools import groupby
 from typing import TYPE_CHECKING, Self, TypeAlias
 
+import cudf_polars.dsl.expressions.binaryop
+import cudf_polars.dsl.expressions.literal
 from cudf_polars.dsl.ir import (
     Filter,
     GroupBy,
@@ -35,6 +37,7 @@
 
     import polars as pl
 
+    from cudf_polars.dsl.expressions.base import Expr
     from cudf_polars.dsl.ir import IR
     from cudf_polars.experimental.base import PartitionInfo, StatsCollector
 
@@ -268,11 +271,24 @@ def _(ir: Sort) -> dict[str, Serializable]:
     }
 
 
+def _serialize_expr(expr: Expr) -> dict[str, Serializable]:
+    match expr:
+        case cudf_polars.dsl.expressions.base.Col(name=name):
+            return {"type": "Col", "name": name}
+        case cudf_polars.dsl.expressions.literal.Literal(value=value):
+            return {"type": "Literal", "value": value}
+        case cudf_polars.dsl.expressions.binaryop.BinOp():
+            return {
+                "op": expr.op.name,
+                "left": _serialize_expr(expr.children[0]),
+                "right": _serialize_expr(expr.children[1]),
+            }
+        case _:
+            return {"type": type(expr).__name__}
+
+
 @_serialize_properties.register
 def _(ir: Filter) -> dict[str, Serializable]:
-    import cudf_polars.dsl.expressions.binaryop
-    import cudf_polars.dsl.expressions.literal
-
     value = ir.mask.value
     properties: dict[str, Serializable] = {
         "predicate": ir.mask.name,
diff --git a/python/cudf_polars/tests/experimental/test_explain.py b/python/cudf_polars/tests/experimental/test_explain.py
index 9ee7cd2f4e50..2653be2e7c3e 100644
--- a/python/cudf_polars/tests/experimental/test_explain.py
+++ b/python/cudf_polars/tests/experimental/test_explain.py
@@ -603,18 +603,36 @@ def test_sort_properties(*, descending: bool):
     assert node.properties == {"by": ["a"], "order": [order]}
 
 
-def test_filter_properties():
-    q = pl.LazyFrame({"a": [1, 2, 3]}).filter(pl.col("a") > 1)
+@pytest.mark.parametrize(
+    "predicate, expected",
+    [
+        (
+            pl.col("a") > 1,
+            {
+                "predicate": "a",
+                "op": "GREATER",
+                "left": {"type": "Col", "name": "a"},
+                "right": {"type": "Literal", "value": 1},
+            },
+        ),
+        (
+            pl.col("a") == pl.col("b"),
+            {
+                "predicate": "a",
+                "op": "EQUAL",
+                "left": {"type": "Col", "name": "a"},
+                "right": {"type": "Col", "name": "b"},
+            },
+        ),
+    ],
+)
+def test_filter_properties(predicate: pl.Expr, expected: dict):
+    q = pl.LazyFrame({"a": [1, 2, 3], "b": [2, 2, 2]}).filter(predicate)
     dag = serialize_query(q, pl.GPUEngine(executor="streaming"))
 
     node = dag.nodes[dag.roots[0]]
     assert node.type == "Filter"
-    assert node.properties == {
-        "predicate": "a",
-        "op": "GREATER",
-        "left": {"type": "Col", "name": "a"},
-        "right": {"type": "Literal", "value": 1},
-    }
+    assert node.properties == expected
 
 
 def test_select_properties():

From 0429ef1be8ee9f9aec713f21acdd4a347c906c1c Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Wed, 4 Feb 2026 14:25:00 -0800
Subject: [PATCH 10/17] coverage

---
 .../cudf_polars/experimental/explain.py       | 46 ++-----------------
 1 file changed, 4 insertions(+), 42 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index 3dcf89bd7b65..aff555a0adf3 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -133,8 +133,6 @@ def serialize_query(
     >>> q = pl.LazyFrame({"a": [1, 2, 3]}).select("a")
     >>> engine = pl.GPUEngine(executor="streaming")
     >>> dag = serialize_query(q, engine, physical=False)
-    >>> json.dumps(dag.to_dict())  # doctest: +SKIP
-    '{"roots": [...], "nodes": {...}, "partition_info": null}'
     """
     return DAG.from_query(q, engine, lowered=physical)
 
@@ -283,34 +281,15 @@ def _serialize_expr(expr: Expr) -> dict[str, Serializable]:
                 "left": _serialize_expr(expr.children[0]),
                 "right": _serialize_expr(expr.children[1]),
             }
-        case _:
+        case _:  # pragma: no cover
             return {"type": type(expr).__name__}
 
 
 @_serialize_properties.register
 def _(ir: Filter) -> dict[str, Serializable]:
     value = ir.mask.value
-    properties: dict[str, Serializable] = {
-        "predicate": ir.mask.name,
-    }
-
-    match value:
-        case cudf_polars.dsl.expressions.binaryop.BinOp():
-            properties["op"] = value.op.name
-            match value.children[0]:
-                case cudf_polars.dsl.expressions.base.Col(name=name):
-                    properties["left"] = {"type": "Col", "name": name}
-                case cudf_polars.dsl.expressions.literal.Literal(value=value):
-                    properties["left"] = {"type": "Literal", "value": value}
-                case _:  # pragma: no cover
-                    properties["left"] = {"type": type(value.children[0]).__name__}
-            match value.children[1]:
-                case cudf_polars.dsl.expressions.base.Col(name=name):
-                    properties["right"] = {"type": "Col", "name": name}
-                case cudf_polars.dsl.expressions.literal.Literal(value=value):
-                    properties["right"] = {"type": "Literal", "value": value}
-                case _:  # pragma: no cover
-                    properties["left"] = {"type": type(value.children[0]).__name__}
+    properties = _serialize_expr(value)
+    properties["predicate"] = ir.mask.name
 
     return properties
 
@@ -355,16 +334,6 @@ def from_ir(cls, ir: IR) -> Self:
             type=type(ir).__name__,
         )
 
-    def to_dict(self) -> dict[str, Serializable]:
-        """Convert to a JSON-serializable dictionary."""
-        return {
-            "id": self.id,
-            "children": self.children,
-            "schema": self.schema,
-            "properties": self.properties,
-            "type": self.type,
-        }
-
 
 @dataclasses.dataclass
 class SerializablePartitionInfo:
@@ -373,13 +342,6 @@ class SerializablePartitionInfo:
     count: int
     partitioned_on: tuple[Serializable, ...]
 
-    def to_dict(self) -> dict[str, Serializable]:
-        """Convert to a JSON-serializable dictionary."""
-        return {
-            "count": self.count,
-            "partitioned_on": list(self.partitioned_on),
-        }
-
 
 @dataclasses.dataclass
 class DAG:
@@ -422,7 +384,7 @@ def from_query(
             if (
                 config.executor.name == "streaming"
                 and config.executor.runtime == "rapidsmpf"
-            ):
+            ):  # pragma: no cover; rapidsmpf runtime not tested in CI yet
                 from cudf_polars.experimental.rapidsmpf.core import (
                     lower_ir_graph as rapidsmpf_lower_ir_graph,
                 )

From fe6e794e70e7359cf813914e275cd94c7c04a23a Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Thu, 5 Feb 2026 05:28:53 -0800
Subject: [PATCH 11/17] doc strings

---
 python/cudf_polars/docs/overview.md | 21 ++++++++++++---------
 1 file changed, 12 insertions(+), 9 deletions(-)

diff --git a/python/cudf_polars/docs/overview.md b/python/cudf_polars/docs/overview.md
index b6e49cae956b..e7aecf364cf3 100644
--- a/python/cudf_polars/docs/overview.md
+++ b/python/cudf_polars/docs/overview.md
@@ -810,25 +810,25 @@ the query plan in a structured format.
 >>> from cudf_polars.experimental.explain import serialize_query
 >>> q = pl.LazyFrame({"a": ['a', 'b', 'a'], "b": [1, 2, 3]}).group_by("a").agg(pl.len())
 >>> dataclasses.asdict(serialize_query(q, engine=pl.GPUEngine()))
-{'roots': ['2408237918'],
- 'nodes': {'2408237918': {'id': '2408237918',
-   'children': [1009833075],
+{'roots': ['526964741'],
+ 'nodes': {'526964741': {'id': '526964741',
+   'children': ['1694929589'],
    'schema': {'a': 'STRING', 'len': 'UINT32'},
    'properties': {'columns': ['a', 'len']},
    'type': 'Select'},
-  '1009833075': {'id': '1009833075',
-   'children': [3262131978],
+  '1694929589': {'id': '1694929589',
+   'children': ['2632275007'],
    'schema': {'a': 'STRING', '___0': 'UINT32'},
    'properties': {'keys': ['a']},
    'type': 'GroupBy'},
-  '3262131978': {'id': '3262131978',
+  '2632275007': {'id': '2632275007',
    'children': [],
    'schema': {'a': 'STRING'},
    'properties': {},
    'type': 'DataFrameScan'}},
- 'partition_info': {'2408237918': {'count': 1, 'partitioned_on': ()},
-  '1009833075': {'count': 1, 'partitioned_on': ()},
-  '3262131978': {'count': 1, 'partitioned_on': ()}}}
+ 'partition_info': {'526964741': {'count': 1, 'partitioned_on': ()},
+  '1694929589': {'count': 1, 'partitioned_on': ()},
+  '2632275007': {'count': 1, 'partitioned_on': ()}}}
 ```
 
 The structured schema has three top-level fields:
@@ -838,3 +838,6 @@ The structured schema has three top-level fields:
 3. `nodes`: A mapping from integer node id to node details. Each node ID
    that appears in the output will be present in this mapping.
    Inspect `children` to understand which nodes this node depends on.
+
+Note that all integers are stored as strings to make round-tripping
+to JSON easier.

From 6295afe3899d82821d73b6127b3b906649e298cb Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Thu, 5 Feb 2026 05:34:32 -0800
Subject: [PATCH 12/17] more docs

---
 .../cudf_polars/experimental/explain.py       | 30 ++++++++++++++++++-
 1 file changed, 29 insertions(+), 1 deletion(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index aff555a0adf3..106680caa8d7 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -345,7 +345,35 @@ class SerializablePartitionInfo:
 
 @dataclasses.dataclass
 class DAG:
-    """A DAG of nodes."""
+    """
+    A DAG of plan nodes, which is serializable to JSON.
+
+    Parameters
+    ----------
+    roots
+        The IDs of the root nodes of the DAG.
+    nodes
+        A mapping from node ID to node details.
+    partition_info
+        Information about the partitions of the DAG.
+
+    Notes
+    -----
+    All integers node IDs are stored as strings to make round-tripping
+    to JSON easier. Node IDs will appear in
+
+    - ``roots``
+    - the keys of ``nodes``
+    - the ``children`` of each node in ``nodes``
+    - the keys in ``partition_info``
+
+    You can safely rely on every key being present in ``nodes``.
+
+    See Also
+    --------
+    serialize_query
+        A function that builds a DAG from a LazyFrame query.
+    """
 
     roots: list[str]
     nodes: dict[str, SerializableIRNode]

From bb57d6f0eb6b15ab43b488f97dafea4c51d78018 Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Thu, 5 Feb 2026 07:07:19 -0800
Subject: [PATCH 13/17] polars 1.30 compat

---
 python/cudf_polars/cudf_polars/experimental/explain.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index 106680caa8d7..13cfb359457d 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -239,9 +239,11 @@ def _serialize_properties(ir: IR) -> dict[str, Serializable]:
 
 @_serialize_properties.register
 def _(ir: Scan) -> dict[str, Serializable]:
+    # for polars<1.31, paths is a list[Path]
+    # for polars>=1.31, paths is a list[str]
     return {
         "typ": ir.typ,
-        "paths": ir.paths,
+        "paths": [str(path) for path in ir.paths],
     }
 
 

From f3f908a03cfe07ec50d388ea282705d7f8c61e51 Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Thu, 5 Feb 2026 12:06:43 -0800
Subject: [PATCH 14/17] Move to get_stable_id

---
 .../experimental/rapidsmpf/tracing.py         | 26 ++-----------------
 .../experimental/rapidsmpf/utils.py           |  3 +--
 2 files changed, 3 insertions(+), 26 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py
index 31ee9c113bb2..bf7b8b1cbc53 100644
--- a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py
+++ b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py
@@ -4,7 +4,6 @@
 
 from __future__ import annotations
 
-import hashlib
 from typing import TYPE_CHECKING
 
 from cudf_polars.dsl.tracing import LOG_TRACES, Scope
@@ -16,27 +15,6 @@
     from cudf_polars.dsl.ir import IR
 
 
-def _stable_ir_id(ir_node: IR) -> int:
-    """
-    Compute a stable identifier for an IR node.
-
-    Uses MD5 hash of the node's hashable representation for determinism
-    across process boundaries (Python's hash() uses PYTHONHASHSEED).
-
-    Parameters
-    ----------
-    ir_node
-        The IR node.
-
-    Returns
-    -------
-    int
-        A stable 32-bit identifier for this node.
-    """
-    content = repr(ir_node.get_hashable()).encode("utf-8")
-    return int(hashlib.md5(content).hexdigest()[:8], 16)
-
-
 class ActorTracer:
     """
     Tracer for a single streaming actor (IR node).
@@ -124,9 +102,9 @@ def log_query_plan(ir: IR) -> None:
 
     nodes = [
         {
-            "ir_id": _stable_ir_id(node),
+            "ir_id": node.get_stable_id(),
             "ir_type": type(node).__name__,
-            "children_ir_ids": [_stable_ir_id(c) for c in node.children],
+            "children_ir_ids": [c.get_stable_id() for c in node.children],
         }
         for node in traversal([ir])
     ]
diff --git a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/utils.py b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/utils.py
index 7972404e1e67..739d03b4313a 100644
--- a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/utils.py
+++ b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/utils.py
@@ -79,10 +79,9 @@ async def shutdown_on_error(
     if LOG_TRACES and trace_ir is not None:
         from cudf_polars.experimental.rapidsmpf.tracing import (
             ActorTracer,
-            _stable_ir_id,
         )
 
-        ir_id = _stable_ir_id(trace_ir)
+        ir_id = trace_ir.get_stable_id()
         ir_type = type(trace_ir).__name__
         tracer = ActorTracer(ir_id, ir_type)
         structlog.contextvars.bind_contextvars(actor_ir_id=ir_id, actor_ir_type=ir_type)

From a1ea77824c69c383fe6e098368889b539311f80c Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Thu, 5 Feb 2026 14:18:10 -0800
Subject: [PATCH 15/17] Updates for logging changes

use this serialize when logging.
---
 .../cudf_polars/experimental/explain.py       | 58 +++++++++++++------
 .../experimental/rapidsmpf/core.py            |  2 +-
 .../experimental/rapidsmpf/tracing.py         | 20 +++----
 3 files changed, 50 insertions(+), 30 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index 13cfb359457d..c47da978bc60 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -382,22 +382,18 @@ class DAG:
     partition_info: dict[str, SerializablePartitionInfo] | None = None
 
     @classmethod
-    def from_query(
-        cls,
-        q: pl.LazyFrame,
-        engine: pl.GPUEngine,
-        *,
-        lowered: bool = False,
+    def from_ir(
+        cls, ir: IR, *, config_options: ConfigOptions, lowered: bool = False
     ) -> Self:
         """
-        Build a DAG from a LazyFrame query.
+        Construct a DAG from an IR node.
 
         Parameters
         ----------
-        q
-            The LazyFrame to serialize.
-        engine
-            The GPU engine to use. If None, uses default streaming executor.
+        ir
+            The IR node to construct the DAG from.
+        config_options
+            The configuration options.
         lowered
             If True, lower the IR to the physical plan and include partition info.
 
@@ -406,22 +402,19 @@ def from_query(
         DAG
             A serializable DAG representation of the query plan.
         """
-        config = ConfigOptions.from_polars_engine(engine)
-        ir = Translator(q._ldf.visit(), engine).translate_ir()
-
         partition_info_dict: dict[str, SerializablePartitionInfo] | None = None
         if lowered:
             if (
-                config.executor.name == "streaming"
-                and config.executor.runtime == "rapidsmpf"
+                config_options.executor.name == "streaming"
+                and config_options.executor.runtime == "rapidsmpf"
             ):  # pragma: no cover; rapidsmpf runtime not tested in CI yet
                 from cudf_polars.experimental.rapidsmpf.core import (
                     lower_ir_graph as rapidsmpf_lower_ir_graph,
                 )
 
-                ir, partition_info_d, _ = rapidsmpf_lower_ir_graph(ir, config)
+                ir, partition_info_d, _ = rapidsmpf_lower_ir_graph(ir, config_options)
             else:
-                ir, partition_info_d, _ = lower_ir_graph(ir, config)
+                ir, partition_info_d, _ = lower_ir_graph(ir, config_options)
             partition_info_dict = {}
 
         nodes: dict[str, SerializableIRNode] = {}
@@ -441,3 +434,32 @@ def from_query(
             nodes=nodes,
             partition_info=partition_info_dict,
         )
+
+    @classmethod
+    def from_query(
+        cls,
+        q: pl.LazyFrame,
+        engine: pl.GPUEngine,
+        *,
+        lowered: bool = False,
+    ) -> Self:
+        """
+        Build a DAG from a LazyFrame query.
+
+        Parameters
+        ----------
+        q
+            The LazyFrame to serialize.
+        engine
+            The GPU engine to use. If None, uses default streaming executor.
+        lowered
+            If True, lower the IR to the physical plan and include partition info.
+
+        Returns
+        -------
+        DAG
+            A serializable DAG representation of the query plan.
+        """
+        config_options = ConfigOptions.from_polars_engine(engine)
+        ir = Translator(q._ldf.visit(), engine).translate_ir()
+        return cls.from_ir(ir, config_options=config_options, lowered=lowered)
diff --git a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/core.py b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/core.py
index cdc8bd04d952..e405a383456a 100644
--- a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/core.py
+++ b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/core.py
@@ -98,7 +98,7 @@ def evaluate_logical_plan(
     ir, partition_info, stats = lower_ir_graph(ir, config_options)
 
     # Log the query plan structure for tracing (no-op if tracing disabled)
-    log_query_plan(ir)
+    log_query_plan(ir, config_options)
 
     # Reserve shuffle IDs for the entire pipeline execution
     with ReserveOpIDs(ir, config_options) as collective_id_map:
diff --git a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py
index bf7b8b1cbc53..c867bb5316ee 100644
--- a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py
+++ b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py
@@ -4,15 +4,17 @@
 
 from __future__ import annotations
 
+import dataclasses
 from typing import TYPE_CHECKING
 
 from cudf_polars.dsl.tracing import LOG_TRACES, Scope
-from cudf_polars.dsl.traversal import traversal
+from cudf_polars.experimental.explain import DAG
 
 if TYPE_CHECKING:
     import pylibcudf as plc
 
     from cudf_polars.dsl.ir import IR
+    from cudf_polars.utils.config import ConfigOptions
 
 
 class ActorTracer:
@@ -78,7 +80,7 @@ def set_duplicated(self, *, duplicated: bool = True) -> None:
         self.duplicated = duplicated
 
 
-def log_query_plan(ir: IR) -> None:
+def log_query_plan(ir: IR, config_options: ConfigOptions) -> None:
     """
     Log the IR tree structure as a structlog event.
 
@@ -90,6 +92,8 @@ def log_query_plan(ir: IR) -> None:
     ----------
     ir
         The root IR node of the lowered query plan.
+    config_options
+        The GPU engine configuration options.
 
     Notes
     -----
@@ -100,14 +104,8 @@ def log_query_plan(ir: IR) -> None:
 
     import structlog
 
-    nodes = [
-        {
-            "ir_id": node.get_stable_id(),
-            "ir_type": type(node).__name__,
-            "children_ir_ids": [c.get_stable_id() for c in node.children],
-        }
-        for node in traversal([ir])
-    ]
+    dag = DAG.from_ir(ir, config_options=config_options)
+    raw = dataclasses.asdict(dag)
 
     log = structlog.get_logger()
-    log.info("Query Plan", scope=Scope.PLAN.value, nodes=nodes)
+    log.info("Query Plan", scope=Scope.PLAN.value, plan=raw)

From 75027f5d346e41c13f502c97431b94c6303c1522 Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Fri, 6 Feb 2026 14:41:46 -0800
Subject: [PATCH 16/17] dag -> plan

---
 .../experimental/benchmarks/utils.py          |  6 ++--
 .../cudf_polars/experimental/explain.py       | 34 +++++++++----------
 .../experimental/rapidsmpf/tracing.py         |  4 +--
 3 files changed, 22 insertions(+), 22 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils.py
index 56763bfa34e3..dfa65b3da16d 100644
--- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils.py
+++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils.py
@@ -59,7 +59,7 @@
 if TYPE_CHECKING:
     from collections.abc import Callable, Sequence
 
-    from cudf_polars.experimental.explain import DAG
+    from cudf_polars.experimental.explain import SerializablePlan
 
 
 try:
@@ -239,7 +239,7 @@ class RunConfig:
         default_factory=PackageVersions.collect
     )
     records: dict[int, list[Record]] = dataclasses.field(default_factory=dict)
-    plans: dict[int, DAG] = dataclasses.field(default_factory=dict)
+    plans: dict[int, SerializablePlan] = dataclasses.field(default_factory=dict)
     dataset_path: Path
     scale_factor: int | float
     shuffle: Literal["rapidsmpf", "tasks"] | None = None
@@ -1055,7 +1055,7 @@ def run_polars(
         run_config = dataclasses.replace(run_config, n_workers=actual_n_workers)
 
     records: defaultdict[int, list[Record]] = defaultdict(list)
-    plans: dict[int, DAG] = {}
+    plans: dict[int, SerializablePlan] = {}
     engine: pl.GPUEngine | None = None
 
     if run_config.executor != "cpu":
diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index c47da978bc60..e89edc14f6f2 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -106,7 +106,7 @@ def serialize_query(
     engine: pl.GPUEngine,
     *,
     physical: bool = True,
-) -> DAG:
+) -> SerializablePlan:
     """
     Return a structured, serializable representation of the IR plan.
 
@@ -122,8 +122,8 @@ def serialize_query(
 
     Returns
     -------
-    DAG
-        A structured DAG representation of the query plan that can be
+    plan
+        A structured representation of the query plan that can be
         serialized to JSON.
 
     Examples
@@ -134,7 +134,7 @@ def serialize_query(
     >>> engine = pl.GPUEngine(executor="streaming")
     >>> dag = serialize_query(q, engine, physical=False)
     """
-    return DAG.from_query(q, engine, lowered=physical)
+    return SerializablePlan.from_query(q, engine, lowered=physical)
 
 
 def _fmt_row_count(value: int | None) -> str:
@@ -313,7 +313,7 @@ def _(ir: HStack) -> dict[str, Serializable]:
 @dataclasses.dataclass
 class SerializableIRNode:
     """
-    A node in the DAG.
+    A node in the plan.
 
     This node is *serializable* and cannot be executed like a
     cudf_polars.dsl.ir.IR node.
@@ -346,18 +346,18 @@ class SerializablePartitionInfo:
 
 
 @dataclasses.dataclass
-class DAG:
+class SerializablePlan:
     """
-    A DAG of plan nodes, which is serializable to JSON.
+    A serializable representation of a query plan.
 
     Parameters
     ----------
     roots
-        The IDs of the root nodes of the DAG.
+        The IDs of the root nodes of the plan.
     nodes
         A mapping from node ID to node details.
     partition_info
-        Information about the partitions of the DAG.
+        Information about the partitions of the plan.
 
     Notes
     -----
@@ -374,7 +374,7 @@ class DAG:
     See Also
     --------
     serialize_query
-        A function that builds a DAG from a LazyFrame query.
+        A function that builds a serializable plan from a LazyFrame query.
     """
 
     roots: list[str]
@@ -386,12 +386,12 @@ def from_ir(
         cls, ir: IR, *, config_options: ConfigOptions, lowered: bool = False
     ) -> Self:
         """
-        Construct a DAG from an IR node.
+        Construct a serializable plan from an IR node.
 
         Parameters
         ----------
         ir
-            The IR node to construct the DAG from.
+            The IR node to construct the serializable plan from.
         config_options
             The configuration options.
         lowered
@@ -399,8 +399,8 @@ def from_ir(
 
         Returns
         -------
-        DAG
-            A serializable DAG representation of the query plan.
+        plan
+            A serializable representation of the query plan.
         """
         partition_info_dict: dict[str, SerializablePartitionInfo] | None = None
         if lowered:
@@ -444,7 +444,7 @@ def from_query(
         lowered: bool = False,
     ) -> Self:
         """
-        Build a DAG from a LazyFrame query.
+        Build a serializable plan from a LazyFrame query.
 
         Parameters
         ----------
@@ -457,8 +457,8 @@ def from_query(
 
         Returns
         -------
-        DAG
-            A serializable DAG representation of the query plan.
+        plan
+            A serializable representation of the query plan.
         """
         config_options = ConfigOptions.from_polars_engine(engine)
         ir = Translator(q._ldf.visit(), engine).translate_ir()
diff --git a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py
index c867bb5316ee..ec24bfab3b7a 100644
--- a/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py
+++ b/python/cudf_polars/cudf_polars/experimental/rapidsmpf/tracing.py
@@ -8,7 +8,7 @@
 from typing import TYPE_CHECKING
 
 from cudf_polars.dsl.tracing import LOG_TRACES, Scope
-from cudf_polars.experimental.explain import DAG
+from cudf_polars.experimental.explain import SerializablePlan
 
 if TYPE_CHECKING:
     import pylibcudf as plc
@@ -104,7 +104,7 @@ def log_query_plan(ir: IR, config_options: ConfigOptions) -> None:
 
     import structlog
 
-    dag = DAG.from_ir(ir, config_options=config_options)
+    dag = SerializablePlan.from_ir(ir, config_options=config_options)
     raw = dataclasses.asdict(dag)
 
     log = structlog.get_logger()

From 5728aa9710c6529a68e975002e026c4ac73ba691 Mon Sep 17 00:00:00 2001
From: Tom Augspurger <toaugspurger@nvidia.com>
Date: Mon, 9 Feb 2026 06:41:33 -0800
Subject: [PATCH 17/17] example

---
 .../cudf_polars/experimental/explain.py       | 38 ++++++++++++++++++-
 1 file changed, 36 insertions(+), 2 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/experimental/explain.py b/python/cudf_polars/cudf_polars/experimental/explain.py
index 7d0a5207653b..ecd72ef46c04 100644
--- a/python/cudf_polars/cudf_polars/experimental/explain.py
+++ b/python/cudf_polars/cudf_polars/experimental/explain.py
@@ -128,9 +128,43 @@ def serialize_query(
     --------
     >>> import polars as pl
     >>> import json
-    >>> q = pl.LazyFrame({"a": [1, 2, 3]}).select("a")
+    >>> import dataclasses
+    >>> q = pl.LazyFrame({"a": [1, 2, 3]}).select(pl.col("a") * 2)
     >>> engine = pl.GPUEngine(executor="streaming")
-    >>> dag = serialize_query(q, engine, physical=False)
+    >>> plan = serialize_query(q, engine, physical=False)
+    >>> print(json.dumps(dataclasses.asdict(plan), indent=2))
+    {
+      "roots": [
+        "1739020873"
+      ],
+      "nodes": {
+        "1739020873": {
+          "id": "1739020873",
+          "children": [
+            "2653195019"
+          ],
+          "schema": {
+            "a": "INT64"
+          },
+          "properties": {
+            "columns": [
+              "a"
+            ]
+          },
+          "type": "Select"
+        },
+        "2653195019": {
+          "id": "2653195019",
+          "children": [],
+          "schema": {
+            "a": "INT64"
+          },
+          "properties": {},
+          "type": "DataFrameScan"
+        }
+      },
+      "partition_info": null
+    }
     """
     return SerializablePlan.from_query(q, engine, lowered=physical)