fix(distributed): single-shard aggregate schema and use-after-free (issue 08)

Resolves the two stacked bugs surfaced by scripts/test_sqlengine.sh
single. Both manifested as the same user-visible symptom (catalog
column headers and "?" rendered values for SELECT COUNT(*) /
SELECT SUM(...) against a single-shard config), but each had a
different root cause and needed an independent fix.

Bug A — wrong column headers (schema)

make_unsharded_aggregate built a remote SQL like SELECT COUNT(*)
FROM users, sent it to the one backend, and returned a bare
REMOTE_SCAN tagged with the source users table. build_column_names
saw the REMOTE_SCAN and emitted all of users.columns[] as the
result schema, mis-labelling the projected aggregates with
unrelated catalog column names. The multi-shard path dodged this
because its MERGE_AGGREGATE node carried explicit output_exprs.

Fix:
- Add output_exprs + output_expr_count to PlanNode::remote_scan,
  mirroring the same fields on merge_aggregate.
- New helper make_remote_scan_with_outputs in distributed_planner.h
  populates them.
- make_unsharded_aggregate uses the new helper, attaching the
  group_by + agg_exprs projection list.
- build_column_names(REMOTE_SCAN) prefers output_exprs (rendered
  through Emitter) when present and falls back to the table's
  catalog columns only for SELECT * passthrough.

Bug B — "?" values (use-after-free)

After bug A was fixed and headers became correct, every aggregate
value still rendered as "?". Tracing showed Value::tag = 58
arriving at the renderer — an enum value out of range, i.e.
uninitialized memory.

PlanExecutor is a stack-local in Session::execute_query. When it
goes out of scope, every operator in operators_ is destroyed.
RemoteScanOperator owns an internal ResultSet (returned from
RemoteExecutor::execute) which heap-owns the Value arrays the
rows point into. When the operator died those arrays were freed,
leaving the outer ResultSet returned to the caller with rows
whose values pointer references freed memory.

This was invisible for any plan tree that wraps the REMOTE_SCAN
in a local operator that re-allocates rows in the executor's
arena (PROJECT, MERGE_AGGREGATE, etc.). The 1-shard aggregate
path is the one shape that yields rows directly from a
REMOTE_SCAN, so issue 08 was the only place it surfaced.

A first attempt moved heap arrays and strings out of the
operator's ResultSet into the outer ResultSet. That worked for
value arrays but corrupted SSO std::string content (every string
lost its first byte) because StringRefs into the source string's
inline buffer became dangling after the move.

Fix that worked: keep the operators alive as long as the returned
ResultSet. New std::vector<std::shared_ptr<void>> backing_lifetimes
on ResultSet holds type-erased ownership of operators released
from PlanExecutor::operators_. Operators (and all storage they
own — heap arrays, owned_strings deque, the inner ResultSet)
survive until the caller is done with the ResultSet. Pointers
stay valid; nothing moves.

Verification

- scripts/test_sqlengine.sh single — was 8/10, now 10/10.
- scripts/test_sqlengine.sh all — 34/34.

Files changed

- include/sql_engine/plan_node.h
- include/sql_engine/distributed_planner.h
- include/sql_engine/plan_executor.h
- include/sql_engine/result_set.h
- docs/issues/08-... (resolution notes)
- docs/issues/README.md (status flip)
- docs/architecture-and-status.md (issue 08 in §8)

Future work

- Consider extending output_exprs to every REMOTE_SCAN (including
  projected non-aggregate pushdown) so the renderer emits better
  column headers across the board.
- Audit other paths where rows might be returned from operator-local
  heap storage without lifetime extension.

Author: renecannao

docs/architecture-and-status.md

@@ -595,8 +595,8 @@ Source: `docs/issues/README.md`. Status reflects the working tree on 2026-04-17.
    *New module `tool_config_parser.{h,cpp}`. Replaces ~135 lines of duplication in each of `sqlengine`, `mysql_server`, `bench_distributed`, `engine_stress_test`, plus `tests/test_ssl_config.cpp`.*
 4. **[04] Close join execution coverage gaps** — 🟦 in working tree.
    *`NestedLoopJoinOperator` now executes RIGHT and FULL outer joins (with right-row match tracking) and resolves qualified column names in join conditions.*
-8. **[08] Aggregate projection schema wrong in single-shard mode** — 📋 open. Surfaced 2026-04-18 by `scripts/test_sqlengine.sh single`.
-   *`SELECT COUNT(*)` / `SELECT SUM(salary)` against a 1-shard config show catalog table columns as headers and put the aggregate value in the wrong slot. Two-shard mode is unaffected.*
+8. **[08] Aggregate projection schema wrong in single-shard mode** — ✅ resolved 2026-04-18.
+   *Two stacked bugs. (a) `make_unsharded_aggregate` returned a bare `REMOTE_SCAN` whose `build_column_names` defaulted to the source table's columns; fixed by adding `output_exprs` to the remote-scan plan node. (b) After (a), values still rendered as `?` because `RemoteScanOperator`'s heap-owned storage died with the stack-local `PlanExecutor` while the returned rows still pointed into it; fixed by adding `backing_lifetimes` to `ResultSet` so operators outlive the executor's stack frame.*
 
 ### P2 — deferred
 

docs/issues/08-aggregate-projection-schema-single-shard.md

@@ -6,7 +6,7 @@
 
 ## Status
 
-Open. Surfaced by `scripts/test_sqlengine.sh single` on 2026-04-18.
+Resolved 2026-04-18.
 
 ## Problem
 
@@ -91,7 +91,54 @@ Worth checking which shard-count branch in `distribute_node` runs when there is
 docker rm -f parsersql-single 2>/dev/null
 ./scripts/setup_single_backend.sh
 make build-sqlengine
-./scripts/test_sqlengine.sh single   # expect 10/10
-make test                            # full unit suite
-./scripts/test_sqlengine.sh all      # expect all-green
+./scripts/test_sqlengine.sh single   # 10/10
+./scripts/test_sqlengine.sh all      # 34/34
 ```
+
+## Resolution Notes
+
+The single bug surfaced by the test was actually two bugs stacked.
+
+### Bug A — wrong column headers (schema)
+
+`make_unsharded_aggregate` in `distributed_planner.h` builds a remote SQL like `SELECT COUNT(*) FROM users`, sends it to the one backend, and returns a bare `REMOTE_SCAN` plan node tagged with the source `users` table. `build_column_names` in `plan_executor.h` saw the `REMOTE_SCAN` and emitted *all of* `users.columns[]` (`id, name, age, dept, salary`) as the result schema. The user saw `id` where the result was actually `COUNT(*)`.
+
+The multi-shard path dodged this because its `MERGE_AGGREGATE` node carried explicit `output_exprs` and had its own `build_column_names` case that used them. The 1-shard path had no equivalent.
+
+**Fix:**
+
+- Added optional `output_exprs` + `output_expr_count` fields to `PlanNode::remote_scan` (mirroring the same fields on `merge_aggregate`).
+- New helper `make_remote_scan_with_outputs(...)` in `distributed_planner.h` populates them.
+- `make_unsharded_aggregate` now uses the new helper, attaching the projection list (`group_by + agg_exprs`) to the REMOTE_SCAN.
+- `build_column_names(REMOTE_SCAN)` prefers the `output_exprs` (rendered through `Emitter`) when present, and falls back to the table's catalog columns only for `SELECT *` passthrough.
+
+### Bug B — `?` rendered values (use-after-free)
+
+After bug A was fixed and headers became correct, every aggregate value still rendered as `?`. Tracing showed `Value::tag = 58` arriving at the renderer — a value far outside the enum range, i.e. uninitialized memory.
+
+`PlanExecutor` is a stack-local in `Session::execute_query`. When it goes out of scope, every operator in `operators_` is destroyed. `RemoteScanOperator` owns an internal `ResultSet` (returned from `RemoteExecutor::execute`) which heap-owns the `Value` arrays the rows point into. When the operator died, those arrays were freed; the outer `ResultSet` returned to the caller had `rows[i].values` pointing at freed memory.
+
+The bug is invisible for any query whose plan tree wraps the `REMOTE_SCAN` in a local operator that re-allocates rows in the executor's arena (PROJECT, MERGE_AGGREGATE, etc.). The 1-shard aggregate path is the one shape that yields rows directly from a `REMOTE_SCAN` to the caller — that's why issue 08 only surfaced there.
+
+A first attempt moved heap arrays and strings out of the operator's `ResultSet` into the outer `ResultSet`. That worked for value arrays but corrupted SSO `std::string` content (every string lost its first byte) because `StringRef`s captured into the source string's inline buffer became dangling after the move.
+
+**Fix that worked:** keep the operators themselves alive as long as the returned `ResultSet`. A new `std::vector<std::shared_ptr<void>> backing_lifetimes` on `ResultSet` holds type-erased ownership of operators released from `PlanExecutor::operators_`. The operators (and all storage they own — heap arrays, the `owned_strings` deque, the inner `ResultSet`) survive until the caller is done with the `ResultSet`. Pointers stay valid; nothing moves.
+
+## Test Coverage
+
+- `scripts/test_sqlengine.sh single` — 10/10 with the fix; the two failing assertions (`single: total user count` finds `10` for `SELECT COUNT(*) FROM users`; `single: SUM(salary) Engineering = 530000` finds the value) now pass.
+- `scripts/test_sqlengine.sh all` — 34/34.
+
+The shell suite is the regression guard. Anyone who reintroduces either bug will see it fail loudly within seconds of running `make test-sqlengine-single`.
+
+## Files Touched
+
+- `include/sql_engine/plan_node.h` — `remote_scan.output_exprs` + `output_expr_count`.
+- `include/sql_engine/distributed_planner.h` — `make_remote_scan_with_outputs`; `make_unsharded_aggregate` uses it.
+- `include/sql_engine/plan_executor.h` — `build_column_names(REMOTE_SCAN)` honours `output_exprs`; `execute()` releases operators into `rs.backing_lifetimes`.
+- `include/sql_engine/result_set.h` — `backing_lifetimes` field; move ctor / move assignment carry it.
+
+## Future Work
+
+- Consider extending `output_exprs` to *every* `REMOTE_SCAN` produced by the planner, including projected (non-aggregate) pushdown, so the renderer can emit better column headers across the board (e.g. `name` instead of catalog ordinal-0).
+- Audit other places where rows might be returned from operator-local heap storage without lifetime extension. `SCAN` against a `MutableDataSource` that returns owned strings could have a similar shape.

docs/issues/README.md

@@ -14,7 +14,7 @@ This directory is the local working backlog for implementation gaps identified f
 2. [Make 2PC phase timeouts deterministic rather than best-effort](02-distributed-2pc-deterministic-phase-timeouts.md) — implemented in current working tree
 3. [Extract shared backend and shard configuration parsing](03-shared-backend-config-parsing.md) — implemented in current working tree
 4. [Close join execution coverage gaps or reject unsupported joins earlier](04-join-operator-coverage.md) — implemented in current working tree
-8. [Aggregate projection schema wrong in single-shard mode](08-aggregate-projection-schema-single-shard.md) — open; surfaced 2026-04-18 by `scripts/test_sqlengine.sh single`
+8. [Aggregate projection schema wrong in single-shard mode](08-aggregate-projection-schema-single-shard.md) — resolved 2026-04-18 (REMOTE_SCAN.output_exprs + ResultSet.backing_lifetimes; surfaced two stacked bugs: schema and use-after-free)
 
 ### P2
 

include/sql_engine/distributed_planner.h

@@ -503,6 +503,26 @@ class DistributedPlanner {
         node->remote_scan.remote_sql = sql.ptr;
         node->remote_scan.remote_sql_len = static_cast<uint16_t>(sql.len);
         node->remote_scan.table = table;
+        // Caller is responsible for setting output_exprs when the remote SQL
+        // is not a passthrough SELECT *. make_plan_node() already zero-fills
+        // the union, so leaving these unset here means "fall back to the
+        // table's catalog columns".
+        return node;
+    }
+
+    PlanNode* make_remote_scan_with_outputs(
+        const char* backend, sql_parser::StringRef sql, const TableInfo* table,
+        const std::vector<const sql_parser::AstNode*>& output_exprs)
+    {
+        PlanNode* node = make_remote_scan(backend, sql, table);
+        if (!output_exprs.empty()) {
+            uint16_t n = static_cast<uint16_t>(output_exprs.size());
+            auto** arr = static_cast<const sql_parser::AstNode**>(
+                arena_.allocate(sizeof(sql_parser::AstNode*) * n));
+            for (uint16_t i = 0; i < n; ++i) arr[i] = output_exprs[i];
+            node->remote_scan.output_exprs = arr;
+            node->remote_scan.output_expr_count = n;
+        }
         return node;
     }
 
@@ -624,7 +644,12 @@ class DistributedPlanner {
             projs.data(), static_cast<uint16_t>(projs.size()),
             gb.data(), static_cast<uint16_t>(gb.size()),
             nullptr, nullptr, 0, -1, false);
-        return make_remote_scan(backend, sql, table);
+        // Carry the projection expressions on the REMOTE_SCAN so the result
+        // schema picks them up instead of mis-labelling with the source
+        // table's catalog columns. Without this, "SELECT COUNT(*) FROM users"
+        // against a single-shard config renders as the table's first
+        // column ("id") rather than the aggregate.
+        return make_remote_scan_with_outputs(backend, sql, table, projs);
     }
 
     void decompose_aggregate(const sql_parser::AstNode* expr,

include/sql_engine/plan_executor.h

@@ -270,6 +270,29 @@ class PlanExecutor {
         }
         root->close();
 
+        // Extend operator lifetimes to match the returned ResultSet.
+        // PlanExecutor is typically a stack-local in Session::execute_query,
+        // so without this every operator would die when the executor goes
+        // out of scope -- taking with it any heap storage that the rows'
+        // Value* / StringRef pointers reference. RemoteScanOperator is the
+        // canonical case: its rows live inside an internal ResultSet
+        // returned from a remote backend, owned by the operator. Bare
+        // single-shard REMOTE_SCAN paths (e.g. SELECT COUNT(*) FROM t
+        // against a one-backend "shard") have no local operator above
+        // them to copy values into the executor's arena, so without
+        // lifetime extension the caller sees garbage tag values rendered
+        // as "?" and zero-overwritten string starts.
+        for (auto& op : operators_) {
+            if (op) {
+                Operator* raw = op.release();
+                rs.backing_lifetimes.emplace_back(
+                    std::shared_ptr<void>(raw, [](void* p) {
+                        delete static_cast<Operator*>(p);
+                    }));
+            }
+        }
+        operators_.clear();
+
         if (!rs.rows.empty()) {
             rs.column_count = rs.rows[0].column_count;
         }
@@ -1203,7 +1226,27 @@ class PlanExecutor {
                 break;
             }
             case PlanNodeType::REMOTE_SCAN: {
-                if (plan->remote_scan.table) {
+                // Prefer the projection expressions the planner attached
+                // (set by make_remote_scan_with_outputs for aggregate or
+                // projected pushdown). Fall back to the table's catalog
+                // columns only for SELECT * passthrough.
+                if (plan->remote_scan.output_exprs &&
+                    plan->remote_scan.output_expr_count > 0) {
+                    for (uint16_t i = 0; i < plan->remote_scan.output_expr_count; ++i) {
+                        const sql_parser::AstNode* expr = plan->remote_scan.output_exprs[i];
+                        if (expr) {
+                            sql_parser::Emitter<D> emitter(arena_);
+                            emitter.emit(expr);
+                            sql_parser::StringRef ev = emitter.result();
+                            if (ev.ptr && ev.len > 0)
+                                rs.column_names.emplace_back(ev.ptr, ev.len);
+                            else
+                                rs.column_names.push_back("?column?");
+                        } else {
+                            rs.column_names.push_back("?column?");
+                        }
+                    }
+                } else if (plan->remote_scan.table) {
                     for (uint16_t i = 0; i < plan->remote_scan.table->column_count; ++i) {
                         auto& cn = plan->remote_scan.table->columns[i].name;
                         rs.column_names.emplace_back(cn.ptr, cn.len);

include/sql_engine/plan_node.h

@@ -102,7 +102,15 @@ struct PlanNode {
             const char* backend_name;
             const char* remote_sql;
             uint16_t remote_sql_len;
-            const TableInfo* table;       // expected result schema
+            const TableInfo* table;       // expected result schema (for SELECT *)
+            // Optional projection expressions used to derive result column
+            // names when the remote SQL is not a passthrough SELECT *. When
+            // non-null, build_column_names() prefers these over the table's
+            // catalog columns. Required for aggregate / projected pushdown
+            // to a single-shard backend; otherwise the catalog's table
+            // columns mis-label the result.
+            const sql_parser::AstNode** output_exprs;
+            uint16_t output_expr_count;
         } remote_scan;
 
         // Merge operations for distributed aggregation

include/sql_engine/result_set.h

@@ -4,12 +4,15 @@
 #include "sql_engine/row.h"
 #include <vector>
 #include <deque>
+#include <memory>
 #include <string>
 #include <algorithm>
 #include <utility>
 
 namespace sql_engine {
 
+class Operator;  // forward decl — ResultSet may extend operator lifetime
+
 struct ResultSet {
     std::vector<Row> rows;
     std::vector<std::string> column_names;
@@ -30,6 +33,16 @@ struct ResultSet {
     // existing elements on push_back, so pointers stay valid.
     std::deque<std::string> owned_strings;
 
+    // Operators that produced rows in this ResultSet may own heap or
+    // arena storage that the rows' Value* / StringRef pointers reference
+    // (notably RemoteScanOperator's internal ResultSet returned from a
+    // remote backend). Without keeping those operators alive past
+    // PlanExecutor's stack scope, rows would dangle the moment
+    // execute_query returned. Keeping them as opaque shared_ptr<void>
+    // here avoids a circular include with operator.h while still
+    // extending their lifetime to match this ResultSet.
+    std::vector<std::shared_ptr<void>> backing_lifetimes;
+
     ResultSet() = default;
     ~ResultSet() {
         for (auto* arr : owned_value_arrays) ::operator delete(arr);
@@ -41,7 +54,8 @@ struct ResultSet {
           column_names(std::move(o.column_names)),
           column_count(o.column_count),
           owned_value_arrays(std::move(o.owned_value_arrays)),
-          owned_strings(std::move(o.owned_strings)) {
+          owned_strings(std::move(o.owned_strings)),
+          backing_lifetimes(std::move(o.backing_lifetimes)) {
         o.column_count = 0;
     }
 
@@ -53,6 +67,7 @@ struct ResultSet {
             column_count = o.column_count;
             owned_value_arrays = std::move(o.owned_value_arrays);
             owned_strings = std::move(o.owned_strings);
+            backing_lifetimes = std::move(o.backing_lifetimes);
             o.column_count = 0;
         }
         return *this;