fix: ResultSet SSO corruption + materializing-op caps + merge-agg schema check

Second batch of fixes from the codebase review. 1059/1096 tests pass
(0 failures, 37 skipped require Docker), 14 new tests added.

1. ResultSet.own_string: SSO use-after-free in owned_strings vector
   This was the actual root cause of the garbled output the user observed
   in the bare-metal benchmarks (SELECT * FROM users showing pointer-like
   IDs and corrupted text columns). The diagnosis the agent gave -- about
   line 711 in mysql_server.cpp -- was wrong; the real bug lives here.

   owned_strings was std::vector<std::string>. own_string() emplace_back'd
   a std::string and returned a StringRef pointing at the new entry's
   c_str(). For short strings using SSO, the character data lives INSIDE
   the std::string object itself, so when the vector reallocated and
   move-constructed each std::string at a new heap address, every
   previously-handed-out StringRef became a dangling pointer to freed
   memory. Long strings happened to survive (their character buffer
   stays at the same heap address through the move) but short strings
   like "Alice", "Eng", "Sales" were silently corrupted.

   Fix: switch to std::deque<std::string>. push_back on a deque does
   not move existing elements, so element addresses (and thus c_str()
   pointers) remain valid for the lifetime of the deque.

   Verified: I temporarily reverted to std::vector<std::string> with the
   new tests in place; the regression tests fail with corrupted data,
   exactly matching the production symptom. Restored the deque fix and
   the tests pass cleanly.

2. MergeAggregateOperator: validate per-row schema matches what
   DistributedPlanner promised
   The merge expected exactly [group_key_count] + [merge_op_count]
   columns per partial-aggregate row. A schema drift between what the
   planner built and what the shard returned would silently truncate or
   misalign the merge, producing wrong aggregate results. Now we throw
   a clear runtime_error on the first row that doesn't match.

3. Hard row caps on materializing operators (engine_limits.h)
   Sort, Aggregate, HashJoin, NestedLoopJoin, Distinct, SetOp, Window
   now cap their materialized state at kDefaultMaxOperatorRows
   (10 million, overridable at compile time via SQL_ENGINE_MAX_OPERATOR_ROWS).
   This converts a "process killed by OOM" failure into a clean
   std::runtime_error with the offending operator name. Not a substitute
   for spill-to-disk, but it stops a runaway query from taking down the
   engine.

Verified-not-a-bug:
- Correlated subqueries: the agent claimed "scaffolding without wiring",
  but PlanExecutor::setup_subquery_executor DOES wire the correlated
  callback, expression_eval DOES pass the outer resolver through, and
  filter_op/project_op DO consult outer_resolver_ on cache miss.
  Existing CorrelatedSubqueryExists / Scalar / Avg tests already pass.
  Added 2 new tests covering correlated NOT EXISTS and correlated IN
  with arithmetic on outer columns -- both pass.

New tests:
- tests/test_result_set.cpp: 4 tests pinning down the deque invariant
  for short strings, long strings, mixed sizes, and Value::str_val
  round-trip after many subsequent push_backs.
- tests/test_operators.cpp: 3 MergeAggregate schema-validation tests
  (matching schema, too few cols, too many cols) and 3 engine_limits
  helper tests (throws at limit, allows below, error message).
- tests/test_subquery.cpp: 2 new correlated subquery tests
  (NOT EXISTS, IN with outer column reference + arithmetic).

Author: renecannao

Makefile

@@ -80,7 +80,8 @@ TEST_SRCS = $(TEST_DIR)/test_main.cpp \
             $(TEST_DIR)/test_distributed_txn.cpp \
             $(TEST_DIR)/test_window.cpp \
             $(TEST_DIR)/test_cte.cpp \
-            $(TEST_DIR)/test_datetime_format.cpp
+            $(TEST_DIR)/test_datetime_format.cpp \
+            $(TEST_DIR)/test_result_set.cpp
 TEST_OBJS = $(TEST_SRCS:.cpp=.o)
 TEST_TARGET = $(PROJECT_ROOT)/run_tests
 

include/sql_engine/engine_limits.h

@@ -0,0 +1,59 @@
+#ifndef SQL_ENGINE_ENGINE_LIMITS_H
+#define SQL_ENGINE_ENGINE_LIMITS_H
+
+// engine_limits.h -- safety caps for materializing operators.
+//
+// Several operators (Sort, Aggregate, HashJoin, NestedLoopJoin, Distinct,
+// SetOp, Window) materialize their input or build state into in-memory
+// collections. Without a cap, a runaway query can exhaust available memory
+// and bring down the whole process via SIGKILL.
+//
+// We don't have spill-to-disk yet. As a defensive measure, every
+// materializing operator enforces a hard row-count cap and throws a clear
+// std::runtime_error when it's exceeded. This converts a "process killed,
+// no message" failure into "query exceeds engine memory limit".
+//
+// The cap is intentionally generous (10 million rows) so existing
+// well-behaved queries are unaffected, but it's small enough that a single
+// query can't allocate tens of gigabytes of row state. For operators with
+// unusual storage shapes (HashJoin's hash table, Aggregate's group map)
+// the same cap applies to the number of distinct entries, which is what
+// actually drives memory usage.
+//
+// The cap can be raised globally by defining
+// SQL_ENGINE_MAX_OPERATOR_ROWS at compile time, and individual call sites
+// can pass a per-operator override via a setter (not currently exposed).
+
+#include <cstddef>
+#include <stdexcept>
+#include <string>
+
+namespace sql_engine {
+
+#ifdef SQL_ENGINE_MAX_OPERATOR_ROWS
+inline constexpr std::size_t kDefaultMaxOperatorRows = SQL_ENGINE_MAX_OPERATOR_ROWS;
+#else
+inline constexpr std::size_t kDefaultMaxOperatorRows = 10'000'000;
+#endif
+
+// Throws std::runtime_error with a clear message when an operator's
+// materialized row/state count is about to exceed the cap. The op_name
+// argument is the operator class name (used in the error message) so users
+// can tell which operator hit the limit.
+inline void check_operator_row_limit(std::size_t current_count,
+                                     std::size_t limit,
+                                     const char* op_name) {
+    if (current_count >= limit) {
+        throw std::runtime_error(
+            std::string(op_name) +
+            ": exceeded engine row limit (" + std::to_string(limit) +
+            " rows / state entries). The engine does not yet spill large "
+            "intermediate results to disk; raise the limit by defining "
+            "SQL_ENGINE_MAX_OPERATOR_ROWS at compile time, or rewrite the "
+            "query to bound the intermediate state.");
+    }
+}
+
+} // namespace sql_engine
+
+#endif // SQL_ENGINE_ENGINE_LIMITS_H

include/sql_engine/operators/aggregate_op.h

@@ -4,6 +4,7 @@
 #include "sql_engine/operator.h"
 #include "sql_engine/expression_eval.h"
 #include "sql_engine/catalog.h"
+#include "sql_engine/engine_limits.h"
 #include "sql_parser/arena.h"
 #include <functional>
 #include <vector>
@@ -44,6 +45,8 @@ class AggregateOperator : public Operator {
 
             auto it = groups_.find(key);
             if (it == groups_.end()) {
+                // Cap distinct group count to prevent unbounded memory.
+                check_operator_row_limit(groups_.size(), kDefaultMaxOperatorRows, "AggregateOperator");
                 GroupState state;
                 state.group_values.reserve(group_count_);
                 for (uint16_t i = 0; i < group_count_; ++i) {

include/sql_engine/operators/distinct_op.h

@@ -2,6 +2,7 @@
 #define SQL_ENGINE_OPERATORS_DISTINCT_OP_H
 
 #include "sql_engine/operator.h"
+#include "sql_engine/engine_limits.h"
 #include <unordered_set>
 #include <string>
 
@@ -20,6 +21,10 @@ class DistinctOperator : public Operator {
     bool next(Row& out) override {
         while (child_->next(out)) {
             std::string key = row_key(out);
+            // Cap distinct row count to prevent unbounded seen_ set.
+            if (seen_.find(key) == seen_.end()) {
+                check_operator_row_limit(seen_.size(), kDefaultMaxOperatorRows, "DistinctOperator");
+            }
             if (seen_.insert(key).second) {
                 return true;
             }

include/sql_engine/operators/hash_join_op.h

@@ -5,6 +5,7 @@
 #include "sql_engine/expression_eval.h"
 #include "sql_engine/catalog.h"
 #include "sql_engine/plan_node.h"
+#include "sql_engine/engine_limits.h"
 #include "sql_parser/arena.h"
 #include <functional>
 #include <vector>
@@ -50,9 +51,13 @@ class HashJoinOperator : public Operator {
         hash_table_.clear();
         right_->open();
         Row r{};
+        std::size_t total_build_rows = 0;
         while (right_->next(r)) {
+            // Cap total build-side rows to prevent unbounded hash table.
+            check_operator_row_limit(total_build_rows, kDefaultMaxOperatorRows, "HashJoinOperator");
             uint64_t h = hash_value(r.get(right_join_col_));
             hash_table_[h].push_back(copy_row(r));
+            ++total_build_rows;
         }
         right_->close();
 

include/sql_engine/operators/join_op.h

@@ -5,6 +5,7 @@
 #include "sql_engine/expression_eval.h"
 #include "sql_engine/catalog.h"
 #include "sql_engine/plan_node.h"
+#include "sql_engine/engine_limits.h"
 #include "sql_parser/arena.h"
 #include <stdexcept>
 #include <functional>
@@ -46,6 +47,8 @@ class NestedLoopJoinOperator : public Operator {
         right_rows_.clear();
         Row r{};
         while (right_->next(r)) {
+            // Cap right side to prevent O(n*m) explosion + OOM.
+            check_operator_row_limit(right_rows_.size(), kDefaultMaxOperatorRows, "NestedLoopJoinOperator");
             right_rows_.push_back(r);
         }
         right_->close();

include/sql_engine/operators/merge_aggregate_op.h

@@ -11,7 +11,7 @@
 #include <string>
 #include <cstring>
 #include <future>
-#include <mutex>
+#include <stdexcept>
 
 namespace sql_engine {
 
@@ -169,6 +169,15 @@ class MergeAggregateOperator : public Operator {
     size_t result_idx_ = 0;
 
     void merge_into_groups(const Row& row) {
+        // Validate that the row's schema matches what DistributedPlanner
+        // promised: exactly [group_key_count_ group keys] +
+        // [merge_op_count_ partial aggregate columns]. Previously, mismatched
+        // schemas were silently truncated by `if (col_idx >= row.column_count)
+        // continue;` in merge_row, producing silently-wrong aggregates. Throw
+        // a clear error instead so a remote schema drift is caught at the
+        // first row.
+        check_row_schema(row);
+
         std::string key = compute_group_key(row);
         auto it = groups_.find(key);
         if (it == groups_.end()) {
@@ -189,6 +198,26 @@ class MergeAggregateOperator : public Operator {
         merge_row(it->second, row);
     }
 
+    // Expected schema is exactly group_key_count_ + merge_op_count_ columns,
+    // in that order. We compute it once into expected_col_count_ on the first
+    // row to avoid recomputing per row. A mismatch on any row aborts the
+    // merge with a clear error message instead of silently corrupting state.
+    void check_row_schema(const Row& row) {
+        const uint16_t expected =
+            static_cast<uint16_t>(group_key_count_ + merge_op_count_);
+        if (row.column_count != expected) {
+            throw std::runtime_error(
+                "distributed aggregate merge: shard returned " +
+                std::to_string(row.column_count) +
+                " columns, expected " + std::to_string(expected) +
+                " (" + std::to_string(group_key_count_) +
+                " group key(s) + " + std::to_string(merge_op_count_) +
+                " partial aggregate(s)). Remote shard schema does not match "
+                "what DistributedPlanner built; aborting to avoid silently "
+                "corrupted aggregate results.");
+        }
+    }
+
     std::string compute_group_key(const Row& row) {
         std::string key;
         for (uint16_t i = 0; i < group_key_count_; ++i) {
@@ -214,9 +243,11 @@ class MergeAggregateOperator : public Operator {
     }
 
     void merge_row(GroupState& state, const Row& row) {
+        // Schema is guaranteed by check_row_schema() above, so we can skip
+        // bounds checks here. Defensive: still compute col_idx with the
+        // expected layout.
         for (uint16_t i = 0; i < merge_op_count_; ++i) {
             uint16_t col_idx = group_key_count_ + i;
-            if (col_idx >= row.column_count) continue;
             Value v = row.get(col_idx);
 
             MergeOp op = static_cast<MergeOp>(merge_ops_[i]);

include/sql_engine/operators/set_op_op.h

@@ -4,6 +4,7 @@
 #include "sql_engine/operator.h"
 #include "sql_engine/plan_node.h"
 #include "sql_engine/thread_pool.h"
+#include "sql_engine/engine_limits.h"
 #include <unordered_set>
 #include <string>
 #include <vector>
@@ -46,6 +47,7 @@ class SetOpOperator : public Operator {
             Row r{};
             while (right_->next(r)) {
                 check_col_count(r);
+                check_operator_row_limit(right_set_.size(), kDefaultMaxOperatorRows, "SetOpOperator");
                 right_set_.insert(row_key(r));
             }
             right_->close();
@@ -68,6 +70,9 @@ class SetOpOperator : public Operator {
                 if (got) {
                     check_col_count(out);
                     std::string key = row_key(out);
+                    if (seen_.find(key) == seen_.end()) {
+                        check_operator_row_limit(seen_.size(), kDefaultMaxOperatorRows, "SetOpOperator");
+                    }
                     if (seen_.insert(key).second) return true;
                 }
             }

include/sql_engine/operators/sort_op.h

@@ -4,6 +4,7 @@
 #include "sql_engine/operator.h"
 #include "sql_engine/expression_eval.h"
 #include "sql_engine/catalog.h"
+#include "sql_engine/engine_limits.h"
 #include "sql_parser/arena.h"
 #include <algorithm>
 #include <functional>
@@ -33,6 +34,8 @@ class SortOperator : public Operator {
 
         Row row{};
         while (child_->next(row)) {
+            // Hard cap to prevent unbounded materialization (no spill yet).
+            check_operator_row_limit(rows_.size(), kDefaultMaxOperatorRows, "SortOperator");
             rows_.push_back(row);
         }
         child_->close();

include/sql_engine/operators/window_op.h

@@ -4,6 +4,7 @@
 #include "sql_engine/operator.h"
 #include "sql_engine/expression_eval.h"
 #include "sql_engine/catalog.h"
+#include "sql_engine/engine_limits.h"
 #include "sql_parser/arena.h"
 #include <functional>
 #include <vector>
@@ -44,6 +45,7 @@ class WindowOperator : public Operator {
         // 1. Consume all child rows into buffer
         Row row{};
         while (child_->next(row)) {
+            check_operator_row_limit(buffer_.size(), kDefaultMaxOperatorRows, "WindowOperator");
             buffer_.push_back(row);
         }
         child_->close();