feat(session): route WITH queries through Session::execute_query (issue 07)

CTE execution existed via PlanExecutor::execute_with_cte but the main
Session::execute_query path didn't use it -- WITH queries through the
user API silently produced wrong results because PlanBuilder::build_cte
built only the main SELECT plan without registering CTE materialisations.
Test coverage validated the executor's special path, not the API users
actually call.

Fix:

- Session::execute_query now detects pr.ast->type == NODE_CTE after
  parsing. CTE queries route to executor.execute_with_cte(pr.ast) and
  deliberately bypass the plan cache. Non-CTE queries follow the
  existing parse -> plan -> optimize -> distribute -> execute -> cache
  pipeline unchanged.
- PlanExecutor::execute_with_cte gains distribute_fn_ application: the
  inner SELECT of each CTE definition and the main query both go
  through the distributor when one is set, so CTE queries that scan
  sharded tables work end-to-end. Materialised CTE rows themselves
  live locally; the distributor leaves them alone.
- The CTE path is intentionally not cached. Per-call re-parse +
  re-materialise is the trade-off for correctness; caching build_cte's
  plan would silently skip materialisation on every cache hit.

Tests:

- tests/test_session.cpp adds four cases that exercise CTEs through
  the user API:
  * CteSimple              -- WITH ... SELECT *
  * CteWithAggregation     -- WITH ... GROUP BY then outer filter
  * CteMultipleDefinitions -- two CTEs, one consumed
  * CteFilteredAfterMaterialisation -- outer WHERE against the CTE
- tests/test_cte.cpp unchanged; the executor-direct path still works
  and is regression-tested.

Verification:

- make test: 1212 passed (+4 new), 37 skipped (live-backend), 0 failed.

Future work (separate issues / follow-ups):

- Recursive WITH RECURSIVE (parser-side gaps tracked under issue 06).
- Plan-cache awareness for CTE queries (cache the AST + build pipeline
  so we re-materialise without re-parsing).

Author: renecannao

docs/issues/07-session-cte-integration.md

@@ -36,3 +36,32 @@ CTE execution exists through `PlanExecutor::execute_with_cte()`, but the main `S
 - Existing CTE tests can be migrated or expanded to exercise the main path
 - CTE support remains intentionally limited until recursive support is explicitly tackled
 
+## Status
+
+Resolved 2026-04-18.
+
+## Resolution Notes
+
+`Session::execute_query` now detects `NODE_CTE` at the AST root and routes those queries through `PlanExecutor::execute_with_cte`. CTE materialisation runs per-call: each CTE definition is built and executed, then registered into the catalog as a synthetic in-memory table; the main `SELECT` then runs against those synthetic tables.
+
+CTE queries deliberately bypass the plan cache. `PlanBuilder::build_cte` only builds the main `SELECT` — caching that plan would silently return wrong results because it would not re-materialise the CTE rows. Per-call re-parse + re-materialise is the trade-off for correctness; recursive CTE optimisation remains explicitly out of scope.
+
+`PlanExecutor::execute_with_cte` was extended to apply the executor's distribute callback to (a) the inner SELECT for each CTE definition and (b) the main query. CTE queries against sharded tables now go through the distributed planner the same way non-CTE queries do; the materialised CTE rows themselves live locally.
+
+### Test Coverage
+
+`tests/test_session.cpp` adds four cases that exercise CTEs through the user API:
+
+- `CteSimple` — `WITH … SELECT *`
+- `CteWithAggregation` — `WITH … GROUP BY` then outer filter
+- `CteMultipleDefinitions` — two CTEs, one consumed
+- `CteFilteredAfterMaterialisation` — outer `WHERE` against the materialised CTE
+
+The existing `tests/test_cte.cpp` cases (which call `execute_with_cte` directly) are unchanged and still pass.
+
+### Future Work
+
+- Recursive `WITH RECURSIVE` (still deferred — see issue 06 for parser-side coverage gaps).
+- Plan-cache awareness for CTE queries (cache the AST + build pipeline so we re-materialise without re-parsing).
+- Push CTE materialisation across the boundary into ProxySQL when this engine is embedded — large-result CTEs may want to live on a backend rather than be pulled local.
+

docs/issues/README.md

@@ -20,7 +20,7 @@ This directory is the local working backlog for implementation gaps identified f
 
 5. [Tighten expression and type semantics](05-expression-and-type-semantics.md) — partial; arrays/tuples/field-access landed (commit a90d147), CHAR_LENGTH UTF-8 + PostgreSQL string casts landed (commit 9f090e5); non-literal arrays through planner, decimal int128, broader decimal semantics still open
 6. [Close parser gaps around MySQL `SELECT ... INTO` and recursive CTE handling](06-parser-gaps-select-into-and-recursive-cte.md)
-7. [Integrate CTE handling into the main `Session` query path](07-session-cte-integration.md)
+7. [Integrate CTE handling into the main `Session` query path](07-session-cte-integration.md) — resolved 2026-04-18
 
 ## Notes
 

include/sql_engine/plan_executor.h

@@ -111,6 +111,7 @@ class PlanExecutor {
             // Not a CTE, build plan normally
             PlanBuilder<D> builder(catalog_, arena_);
             PlanNode* plan = builder.build(ast);
+            if (plan && distribute_fn_) plan = distribute_fn_(plan);
             return execute(plan);
         }
 
@@ -127,6 +128,10 @@ class PlanExecutor {
             PlanBuilder<D> cte_builder(catalog_, arena_);
             PlanNode* cte_plan = cte_builder.build(inner_select);
             if (!cte_plan) continue;
+            // Apply distributed planning to the CTE definition's inner
+            // SELECT so a CTE that scans a sharded table still materialises
+            // correctly. The materialised CTE rows live locally afterwards.
+            if (distribute_fn_) cte_plan = distribute_fn_(cte_plan);
 
             ResultSet cte_rs = execute(cte_plan);
 
@@ -242,6 +247,11 @@ class PlanExecutor {
 
         PlanBuilder<D> builder(catalog_, arena_);
         PlanNode* plan = builder.build(main_query);
+        // The main query may itself reference sharded tables; distribute
+        // it the same way Session::execute_query would for a non-CTE
+        // SELECT. Materialised CTEs registered above show up as local
+        // data sources, so the distributor leaves them alone.
+        if (plan && distribute_fn_) plan = distribute_fn_(plan);
         ResultSet result = execute(plan);
 
         // CTE sources are kept alive by cte_sources until we return

include/sql_engine/session.h

@@ -105,6 +105,20 @@ class Session {
             return {};
         }
 
+        // CTE queries route through PlanExecutor::execute_with_cte so the
+        // WITH clause's CTE definitions are materialized into in-memory
+        // data sources before the main SELECT runs. We deliberately do
+        // NOT cache CTE plans: build_cte() produces only the main SELECT
+        // plan with no CTE materialization, so a cache hit would silently
+        // return wrong results. Per issue 07's "intentionally limited"
+        // scope, CTEs re-parse + re-materialize on every call. Recursive
+        // CTEs are still out of scope.
+        if (pr.ast->type == sql_parser::NodeType::NODE_CTE) {
+            PlanExecutor<D> executor(functions_, catalog_, cached_parser->arena());
+            wire_executor(executor);
+            return executor.execute_with_cte(pr.ast);
+        }
+
         PlanBuilder<D> builder(catalog_, cached_parser->arena());
         PlanNode* plan = builder.build(pr.ast);
         if (!plan) return {};

tests/test_session.cpp

@@ -551,3 +551,46 @@ TEST(SessionAutoEnlistment, ScatterDmlRoutedThroughRouteDml) {
 
     txn_mgr.rollback();
 }
+
+// ===== CTE via the user API (issue 07) =====
+
+// Simple WITH clause executed through Session::execute_query — same SQL
+// that previously only worked through PlanExecutor::execute_with_cte.
+TEST_F(SessionTest, CteSimple) {
+    auto rs = session->execute_query(
+        "WITH adults AS (SELECT * FROM users WHERE age >= 25) SELECT * FROM adults");
+    // Both rows in the fixture are >= 25 (Alice 25, Bob 30).
+    EXPECT_EQ(rs.row_count(), 2u);
+}
+
+// CTE with aggregation; the inner SELECT runs first, materialises into
+// the catalog as a synthetic table, then the outer SELECT filters it.
+TEST_F(SessionTest, CteWithAggregation) {
+    // Insert one more row so the GROUP BY has multiple groups.
+    session->execute_statement("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 30)");
+    auto rs = session->execute_query(
+        "WITH age_groups AS ("
+        "  SELECT age, COUNT(*) AS cnt FROM users GROUP BY age"
+        ") SELECT * FROM age_groups WHERE cnt > 1");
+    // age=30 has 2 users (Bob, Carol); age=25 has 1 user (Alice).
+    EXPECT_EQ(rs.row_count(), 1u);
+}
+
+// Multiple CTE definitions, only one consumed by the main SELECT.
+TEST_F(SessionTest, CteMultipleDefinitions) {
+    auto rs = session->execute_query(
+        "WITH young AS (SELECT * FROM users WHERE age < 30), "
+        "     senior AS (SELECT * FROM users WHERE age >= 30) "
+        "SELECT * FROM senior");
+    // Only Bob (age 30) is senior.
+    EXPECT_EQ(rs.row_count(), 1u);
+}
+
+// CTE result then filtered by an outer WHERE.
+TEST_F(SessionTest, CteFilteredAfterMaterialisation) {
+    auto rs = session->execute_query(
+        "WITH all_users AS (SELECT name, age FROM users) "
+        "SELECT * FROM all_users WHERE age >= 30");
+    EXPECT_EQ(rs.row_count(), 1u);
+    EXPECT_EQ(rs.rows[0].get(0).tag, Value::TAG_STRING);
+}