@@ -204,6 +204,15 @@ planning, as it is over tables of more than about a megabyte, a call
 plans the query again and runs as fast as `ctx.sql`. So does a call of
 a plan over a lazy table (Dask or a lazily opened file).
 
+DataFusion plans a CTE or a view in full at each place a query reads
+it. A model whose layers each read the layer before more than once
+grows with every layer: a one-layer GPT's loss plans as 9,882 operators,
+repeats of only 87 distinct subplans. `prepare` runs each CTE or view read more
+than once just once per call, and keeps its rows in a table for the
+rest of the query to read. That GPT's loss takes 16 ms per call, and
+2.7 s with `ctx.sql`. A query with `WITH RECURSIVE`, or a subquery with
+CTEs of its own, is planned as it's written.
+
 ## Round-trip faster with ORDER BY
 
 Results that arrive **grid-ordered** — sorted by the dimension columns,
 
@@ -262,3 +262,108 @@ def test_a_quoted_table_replaced_in_sql_counts_too():
     ctx.sql('CREATE OR REPLACE TABLE "Mixed" AS VALUES (1), (2)').collect()
     with pytest.raises(ValueError, match="registered again"):
         query()
+
+
+def _planned(query, a, b, *setup):
+    """The rows of ``query`` planned from scratch over ``a`` and ``b``."""
+    fresh = Context()
+    fresh.from_dataset("a", a)
+    fresh.from_dataset("b", b)
+    for statement in setup:
+        fresh.sql(statement)
+    return fresh.sql(query).to_arrow_table()
+
+
+def _calls(query, *setup):
+    """A prepared ``query``'s rows on new values of ``a``, against planning
+    it."""
+    rng = np.random.default_rng(1)
+    b = _b(rng.normal(size=(4, 3)))
+    ctx = Context()
+    ctx.from_dataset("a", _a(rng.normal(size=(4, 3))))
+    ctx.from_dataset("b", b)
+    for statement in setup:
+        ctx.sql(statement)
+    prepared = ctx.prepare(query)
+    for _ in range(3):
+        a = _a(rng.normal(size=(4, 3)))
+        yield prepared(a=a), _planned(query, a, b, *setup)
+
+
+# Each reads `t` twice, so a call runs `t` once and stores its rows.
+@pytest.mark.parametrize(
+    "query",
+    [
+        # A CTE read twice.
+        (
+            "WITH t AS (SELECT i, j, 2 * v AS v2 FROM a) "
+            "SELECT x.i, x.j, x.v2 * y.v2 AS p FROM t x JOIN t y "
+            "ON x.i = y.i AND x.j = y.j ORDER BY x.i, x.j"
+        ),
+        # Read twice, by a CTE read twice itself.
+        (
+            "WITH t AS (SELECT i, j, v FROM a), "
+            "u AS (SELECT x.i, x.j, x.v + y.v AS v FROM t x JOIN t y "
+            "ON x.i = y.i AND x.j = y.j), "
+            "s AS (SELECT u.i, SUM(u.v * b.w) AS s FROM u JOIN b "
+            "ON u.i = b.i AND u.j = b.j GROUP BY u.i) "
+            "SELECT p.i, p.s + q.s AS s FROM s p JOIN s q ON p.i = q.i ORDER BY p.i"
+        ),
+        # Rows whose number changes from call to call.
+        (
+            "WITH t AS (SELECT i, j, v FROM a WHERE v > 0) "
+            "SELECT (SELECT COUNT(*) FROM t) AS n, (SELECT SUM(v) FROM t) AS s"
+        ),
+        # Rows with nulls, and with strings, which aren't written in place.
+        (
+            "WITH t AS (SELECT i, j, CASE WHEN v > 0 THEN v END AS v, "
+            "CASE WHEN v > 0 THEN 'up' ELSE 'down' END AS d FROM a) "
+            "SELECT x.i, x.j, x.v, y.d FROM t x JOIN t y "
+            "ON x.i = y.i AND x.j = y.j ORDER BY x.i, x.j"
+        ),
+        # Columns named where the CTE is.
+        (
+            "WITH t(r, c, val) AS (SELECT i, j, v FROM a) "
+            "SELECT x.r, x.c, x.val + y.val AS z FROM t x JOIN t y "
+            "ON x.r = y.r AND x.c = y.c ORDER BY x.r, x.c"
+        ),
+    ],
+)
+def test_a_cte_read_more_than_once_gives_the_rows_planning_gives(query):
+    for rows, want in _calls(query):
+        assert rows.equals(want)
+
+
+def test_a_view_read_more_than_once_gives_the_rows_planning_gives():
+    view = (
+        "CREATE VIEW t AS WITH p AS (SELECT i, j, 2 * v AS v FROM a) "
+        "SELECT x.i, x.j, x.v + y.v AS v FROM p x JOIN p y "
+        "ON x.i = y.i AND x.j = y.j"
+    )
+    query = (
+        "SELECT x.i, SUM(x.v * y.v) AS s FROM t x JOIN t y "
+        "ON x.i = y.i AND x.j = y.j GROUP BY x.i ORDER BY x.i"
+    )
+    for rows, want in _calls(query, view):
+        assert rows.equals(want)
+
+
+def test_a_cte_named_like_a_view_hides_it():
+    view = "CREATE VIEW t AS SELECT i, j, -v AS v FROM a"
+    query = (
+        "WITH t AS (SELECT i, j, v FROM a) "
+        "SELECT x.i, x.j, x.v + y.v AS v FROM t x JOIN t y "
+        "ON x.i = y.i AND x.j = y.j ORDER BY x.i, x.j"
+    )
+    for rows, want in _calls(query, view):
+        assert rows.equals(want)
+
+
+def test_a_recursive_cte_runs_again():
+    query = (
+        "WITH RECURSIVE r(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM r "
+        "WHERE n < 4) SELECT SUM(n) AS s FROM r"
+    )
+    prepared = Context().prepare(query)
+    assert prepared().to_pydict() == {"s": [10]}
+    assert prepared().to_pydict() == {"s": [10]}
@@ -12,14 +12,14 @@
 from types import ModuleType
 
 from . import cftime as cft
+from ._native import share as _share_sql
 from .df import (
     Chunks,
     TableNames,
     group_vars_by_dims,
     resolve_table_names,
     shared_coord_arrays,
 )
-from ._native import share as _share_sql
 from .ds import XarrayDataFrame
 from .reader import _in_memory_partitions, read_xarray_table
 
@@ -525,20 +525,22 @@ def write(self, rows: pa.Table) -> bool:
         schema and number of rows, in numeric columns without nulls."""
         if rows.schema != self._schema or rows.num_rows != self._rows:
             return False
-        if any(
-            array is None or column.null_count
-            for array, column in zip(self._arrays, rows.columns)
+        arrays = [array for array in self._arrays if array is not None]
+        if len(arrays) < len(self._arrays) or any(
+            column.null_count for column in rows.columns
         ):
             return False
-        for array, column in zip(self._arrays, rows.columns):
+        for array, column in zip(arrays, rows.columns):
             array[...] = column.to_numpy()
         return True
 
 
 def _writable(column: pa.ChunkedArray) -> np.ndarray | None:
     """A copy of ``column`` that pyarrow can share memory with, if it's
     numeric and has no nulls."""
-    numeric = pa.types.is_integer(column.type) or pa.types.is_floating(column.type)
+    numeric = pa.types.is_integer(column.type) or pa.types.is_floating(
+        column.type
+    )
     if not numeric or column.null_count:
         return None
     return np.array(column.to_numpy(), copy=True)
@@ -563,9 +565,10 @@ def _share(
         ).to_pydict()
     except Exception:  # noqa: BLE001 - no information schema to read
         views = {"table_name": [], "definition": []}
-    return _share_sql(
+    shared: tuple[list[tuple[str, str]], str] | None = _share_sql(
         query, dict(zip(views["table_name"], views["definition"])), prefix
     )
+    return shared
 
 
 def _copy(ctx: SessionContext, plan: ExecutionPlan) -> ExecutionPlan: