fix: remove arena from remote executors, use heap-owned ResultSet data (#33)

Remote executors (MySQL, PostgreSQL, Multi) no longer take an Arena&
parameter. Instead, ResultSet owns its row value arrays and string data
on the heap via owned_value_arrays and owned_strings, eliminating the
arena memory leak under sustained load.

Author: renecannao

include/sql_engine/multi_remote_executor.h

@@ -5,7 +5,6 @@
 #include "sql_engine/mysql_remote_executor.h"
 #include "sql_engine/pgsql_remote_executor.h"
 #include "sql_engine/backend_config.h"
-#include "sql_parser/arena.h"
 #include "sql_parser/common.h"
 
 #include <unordered_map>
@@ -15,7 +14,7 @@ namespace sql_engine {
 
 class MultiRemoteExecutor : public RemoteExecutor {
 public:
-    explicit MultiRemoteExecutor(sql_parser::Arena& arena);
+    MultiRemoteExecutor();
     ~MultiRemoteExecutor() override;
 
     void add_backend(const BackendConfig& config);

include/sql_engine/mysql_remote_executor.h

@@ -5,7 +5,7 @@
 #include "sql_engine/backend_config.h"
 #include "sql_engine/value.h"
 #include "sql_engine/row.h"
-#include "sql_parser/arena.h"
+#include "sql_engine/result_set.h"
 
 #include <mysql/mysql.h>
 #include <unordered_map>
@@ -15,7 +15,7 @@ namespace sql_engine {
 
 class MySQLRemoteExecutor : public RemoteExecutor {
 public:
-    explicit MySQLRemoteExecutor(sql_parser::Arena& arena);
+    MySQLRemoteExecutor();
     ~MySQLRemoteExecutor() override;
 
     void add_backend(const BackendConfig& config);
@@ -31,11 +31,11 @@ class MySQLRemoteExecutor : public RemoteExecutor {
     };
 
     std::unordered_map<std::string, Connection> backends_;
-    sql_parser::Arena& arena_;
 
     Connection& get_or_connect(const std::string& name);
     ResultSet mysql_result_to_resultset(MYSQL_RES* res);
-    Value mysql_field_to_value(const char* data, unsigned long length,
+    Value mysql_field_to_value(ResultSet& rs, const char* data,
+                               unsigned long length,
                                enum_field_types type, bool is_null);
 };
 

include/sql_engine/pgsql_remote_executor.h

@@ -5,7 +5,7 @@
 #include "sql_engine/backend_config.h"
 #include "sql_engine/value.h"
 #include "sql_engine/row.h"
-#include "sql_parser/arena.h"
+#include "sql_engine/result_set.h"
 
 #include <libpq-fe.h>
 #include <unordered_map>
@@ -15,7 +15,7 @@ namespace sql_engine {
 
 class PgSQLRemoteExecutor : public RemoteExecutor {
 public:
-    explicit PgSQLRemoteExecutor(sql_parser::Arena& arena);
+    PgSQLRemoteExecutor();
     ~PgSQLRemoteExecutor() override;
 
     void add_backend(const BackendConfig& config);
@@ -31,11 +31,11 @@ class PgSQLRemoteExecutor : public RemoteExecutor {
     };
 
     std::unordered_map<std::string, Connection> backends_;
-    sql_parser::Arena& arena_;
 
     Connection& get_or_connect(const std::string& name);
     ResultSet pg_result_to_resultset(PGresult* res);
-    Value pg_field_to_value(const char* data, int length, Oid type, bool is_null);
+    Value pg_field_to_value(ResultSet& rs, const char* data, int length,
+                            Oid type, bool is_null);
 };
 
 } // namespace sql_engine

include/sql_engine/result_set.h

@@ -4,6 +4,8 @@
 #include "sql_engine/row.h"
 #include <vector>
 #include <string>
+#include <algorithm>
+#include <utility>
 
 namespace sql_engine {
 
@@ -12,8 +14,62 @@ struct ResultSet {
     std::vector<std::string> column_names;
     uint16_t column_count = 0;
 
+    // Heap storage for row values and string data owned by this ResultSet.
+    // Used by remote executors to avoid arena lifetime issues.
+    std::vector<Value*> owned_value_arrays;
+    std::vector<std::string> owned_strings;
+
+    ResultSet() = default;
+    ~ResultSet() {
+        for (auto* arr : owned_value_arrays) ::operator delete(arr);
+    }
+
+    // Move-only (owns heap arrays)
+    ResultSet(ResultSet&& o) noexcept
+        : rows(std::move(o.rows)),
+          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)) {
+        o.column_count = 0;
+    }
+
+    ResultSet& operator=(ResultSet&& o) noexcept {
+        if (this != &o) {
+            for (auto* arr : owned_value_arrays) ::operator delete(arr);
+            rows = std::move(o.rows);
+            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);
+            o.column_count = 0;
+        }
+        return *this;
+    }
+
+    // Disable copy to prevent double-free of owned arrays
+    ResultSet(const ResultSet&) = delete;
+    ResultSet& operator=(const ResultSet&) = delete;
+
     size_t row_count() const { return rows.size(); }
     bool empty() const { return rows.empty(); }
+
+    // Allocate a heap-owned row and append it to rows. Returns a reference
+    // to the Row (which points into owned_value_arrays).
+    Row& add_heap_row(uint16_t col_count) {
+        Value* vals = static_cast<Value*>(::operator new(sizeof(Value) * col_count));
+        for (uint16_t i = 0; i < col_count; ++i) vals[i] = value_null();
+        owned_value_arrays.push_back(vals);
+        rows.push_back(Row{vals, col_count});
+        return rows.back();
+    }
+
+    // Store a string in owned_strings and return a StringRef pointing into it.
+    sql_parser::StringRef own_string(const char* data, uint32_t len) {
+        owned_strings.emplace_back(data, len);
+        const std::string& s = owned_strings.back();
+        return sql_parser::StringRef{s.c_str(), static_cast<uint32_t>(s.size())};
+    }
 };
 
 } // namespace sql_engine

src/sql_engine/multi_remote_executor.cpp

@@ -2,8 +2,7 @@
 
 namespace sql_engine {
 
-MultiRemoteExecutor::MultiRemoteExecutor(sql_parser::Arena& arena)
-    : mysql_exec_(arena), pgsql_exec_(arena) {}
+MultiRemoteExecutor::MultiRemoteExecutor() {}
 
 MultiRemoteExecutor::~MultiRemoteExecutor() {
     disconnect_all();

src/sql_engine/mysql_remote_executor.cpp

@@ -6,8 +6,7 @@
 
 namespace sql_engine {
 
-MySQLRemoteExecutor::MySQLRemoteExecutor(sql_parser::Arena& arena)
-    : arena_(arena) {}
+MySQLRemoteExecutor::MySQLRemoteExecutor() {}
 
 MySQLRemoteExecutor::~MySQLRemoteExecutor() {
     disconnect_all();
@@ -131,21 +130,20 @@ ResultSet MySQLRemoteExecutor::mysql_result_to_resultset(MYSQL_RES* res) {
     MYSQL_ROW mysql_row;
     while ((mysql_row = mysql_fetch_row(res)) != nullptr) {
         unsigned long* lengths = mysql_fetch_lengths(res);
-        Row row = make_row(arena_, rs.column_count);
+        Row& row = rs.add_heap_row(rs.column_count);
         for (unsigned int i = 0; i < num_fields; ++i) {
             bool is_null = (mysql_row[i] == nullptr);
             Value v = mysql_field_to_value(
-                mysql_row[i], is_null ? 0 : lengths[i],
+                rs, mysql_row[i], is_null ? 0 : lengths[i],
                 fields[i].type, is_null);
             row.set(static_cast<uint16_t>(i), v);
         }
-        rs.rows.push_back(row);
     }
     return rs;
 }
 
 Value MySQLRemoteExecutor::mysql_field_to_value(
-    const char* data, unsigned long length,
+    ResultSet& rs, const char* data, unsigned long length,
     enum_field_types type, bool is_null) {
 
     if (is_null) return value_null();
@@ -165,8 +163,7 @@ Value MySQLRemoteExecutor::mysql_field_to_value(
 
         case MYSQL_TYPE_DECIMAL:
         case MYSQL_TYPE_NEWDECIMAL: {
-            // Store as decimal (numeric string) in arena
-            sql_parser::StringRef s = arena_.allocate_string(data, static_cast<uint32_t>(length));
+            sql_parser::StringRef s = rs.own_string(data, static_cast<uint32_t>(length));
             return value_decimal(s);
         }
 
@@ -197,18 +194,18 @@ Value MySQLRemoteExecutor::mysql_field_to_value(
         case MYSQL_TYPE_TINY_BLOB:
         case MYSQL_TYPE_MEDIUM_BLOB:
         case MYSQL_TYPE_LONG_BLOB: {
-            sql_parser::StringRef s = arena_.allocate_string(data, static_cast<uint32_t>(length));
+            sql_parser::StringRef s = rs.own_string(data, static_cast<uint32_t>(length));
             return value_bytes(s);
         }
 
         case MYSQL_TYPE_JSON: {
-            sql_parser::StringRef s = arena_.allocate_string(data, static_cast<uint32_t>(length));
+            sql_parser::StringRef s = rs.own_string(data, static_cast<uint32_t>(length));
             return value_json(s);
         }
 
         default: {
             // STRING, VAR_STRING, ENUM, SET, etc. — treat as string
-            sql_parser::StringRef s = arena_.allocate_string(data, static_cast<uint32_t>(length));
+            sql_parser::StringRef s = rs.own_string(data, static_cast<uint32_t>(length));
             return value_string(s);
         }
     }

src/sql_engine/pgsql_remote_executor.cpp

@@ -29,8 +29,7 @@
 
 namespace sql_engine {
 
-PgSQLRemoteExecutor::PgSQLRemoteExecutor(sql_parser::Arena& arena)
-    : arena_(arena) {}
+PgSQLRemoteExecutor::PgSQLRemoteExecutor() {}
 
 PgSQLRemoteExecutor::~PgSQLRemoteExecutor() {
     disconnect_all();
@@ -152,22 +151,21 @@ ResultSet PgSQLRemoteExecutor::pg_result_to_resultset(PGresult* res) {
     }
 
     for (int r = 0; r < num_rows; ++r) {
-        Row row = make_row(arena_, rs.column_count);
+        Row& row = rs.add_heap_row(rs.column_count);
         for (int c = 0; c < num_fields; ++c) {
             bool is_null = PQgetisnull(res, r, c);
             Oid oid = PQftype(res, c);
             const char* data = PQgetvalue(res, r, c);
             int length = PQgetlength(res, r, c);
-            Value v = pg_field_to_value(data, length, oid, is_null);
+            Value v = pg_field_to_value(rs, data, length, oid, is_null);
             row.set(static_cast<uint16_t>(c), v);
         }
-        rs.rows.push_back(row);
     }
     return rs;
 }
 
 Value PgSQLRemoteExecutor::pg_field_to_value(
-    const char* data, int length, Oid type, bool is_null) {
+    ResultSet& rs, const char* data, int length, Oid type, bool is_null) {
 
     if (is_null) return value_null();
 
@@ -186,7 +184,7 @@ Value PgSQLRemoteExecutor::pg_field_to_value(
             return value_double(std::strtod(data, nullptr));
 
         case NUMERICOID: {
-            sql_parser::StringRef s = arena_.allocate_string(data, static_cast<uint32_t>(length));
+            sql_parser::StringRef s = rs.own_string(data, static_cast<uint32_t>(length));
             return value_string(s);
         }
 
@@ -213,19 +211,19 @@ Value PgSQLRemoteExecutor::pg_field_to_value(
         }
 
         case BYTEAOID: {
-            sql_parser::StringRef s = arena_.allocate_string(data, static_cast<uint32_t>(length));
+            sql_parser::StringRef s = rs.own_string(data, static_cast<uint32_t>(length));
             return value_bytes(s);
         }
 
         case JSONOID:
         case JSONBOID: {
-            sql_parser::StringRef s = arena_.allocate_string(data, static_cast<uint32_t>(length));
+            sql_parser::StringRef s = rs.own_string(data, static_cast<uint32_t>(length));
             return value_json(s);
         }
 
         default: {
             // TEXT, VARCHAR, BPCHAR, NAME, and everything else -- treat as string
-            sql_parser::StringRef s = arena_.allocate_string(data, static_cast<uint32_t>(length));
+            sql_parser::StringRef s = rs.own_string(data, static_cast<uint32_t>(length));
             return value_string(s);
         }
     }

tests/test_distributed_real.cpp

@@ -77,8 +77,7 @@ void register_test_catalog(InMemoryCatalog& catalog) {
 class DistributedRealMySQLTest : public ::testing::Test {
 protected:
     void SetUp() override {
-        exec_arena_ = std::make_unique<Arena>(65536, 1048576);
-        exec_ = std::make_unique<MySQLRemoteExecutor>(*exec_arena_);
+        exec_ = std::make_unique<MySQLRemoteExecutor>();
 
         BackendConfig cfg;
         cfg.name = "shard_1";
@@ -129,7 +128,6 @@ class DistributedRealMySQLTest : public ::testing::Test {
         return executor.execute(dist_plan);
     }
 
-    std::unique_ptr<Arena> exec_arena_;
     std::unique_ptr<MySQLRemoteExecutor> exec_;
     InMemoryCatalog catalog_;
     FunctionRegistry<Dialect::MySQL> functions_;
@@ -167,15 +165,13 @@ TEST_F(DistributedRealMySQLTest, SelectFromOrders) {
 class MultiExecutorTest : public ::testing::Test {
 protected:
     void SetUp() override {
-        arena_ = std::make_unique<Arena>(65536, 1048576);
-        exec_ = std::make_unique<MultiRemoteExecutor>(*arena_);
+        exec_ = std::make_unique<MultiRemoteExecutor>();
     }
 
     void TearDown() override {
         exec_->disconnect_all();
     }
 
-    std::unique_ptr<Arena> arena_;
     std::unique_ptr<MultiRemoteExecutor> exec_;
 };
 

tests/test_mysql_executor.cpp

@@ -1,7 +1,6 @@
 #include <gtest/gtest.h>
 #include "sql_engine/mysql_remote_executor.h"
 #include "sql_engine/backend_config.h"
-#include "sql_parser/arena.h"
 
 #include <mysql/mysql.h>
 
@@ -41,18 +40,15 @@ sql_engine::BackendConfig make_mysql_config(const char* name = "test_mysql") {
 class MySQLExecutorTest : public ::testing::Test {
 protected:
     void SetUp() override {
-        arena_ = std::make_unique<sql_parser::Arena>(65536, 1048576);
-        exec_ = std::make_unique<sql_engine::MySQLRemoteExecutor>(*arena_);
+        exec_ = std::make_unique<sql_engine::MySQLRemoteExecutor>();
         exec_->add_backend(make_mysql_config());
     }
 
     void TearDown() override {
         exec_->disconnect_all();
         exec_.reset();
-        arena_.reset();
     }
 
-    std::unique_ptr<sql_parser::Arena> arena_;
     std::unique_ptr<sql_engine::MySQLRemoteExecutor> exec_;
 };
 

tests/test_pgsql_executor.cpp

@@ -1,7 +1,6 @@
 #include <gtest/gtest.h>
 #include "sql_engine/pgsql_remote_executor.h"
 #include "sql_engine/backend_config.h"
-#include "sql_parser/arena.h"
 
 #include <libpq-fe.h>
 
@@ -43,18 +42,15 @@ sql_engine::BackendConfig make_pgsql_config(const char* name = "test_pgsql") {
 class PgSQLExecutorTest : public ::testing::Test {
 protected:
     void SetUp() override {
-        arena_ = std::make_unique<sql_parser::Arena>(65536, 1048576);
-        exec_ = std::make_unique<sql_engine::PgSQLRemoteExecutor>(*arena_);
+        exec_ = std::make_unique<sql_engine::PgSQLRemoteExecutor>();
         exec_->add_backend(make_pgsql_config());
     }
 
     void TearDown() override {
         exec_->disconnect_all();
         exec_.reset();
-        arena_.reset();
     }
 
-    std::unique_ptr<sql_parser::Arena> arena_;
     std::unique_ptr<sql_engine::PgSQLRemoteExecutor> exec_;
 };