fix: resolve SegFaults via proper PMR propagation and record boundaries

Author: poyrazK

include/executor/operator.hpp

@@ -74,12 +74,12 @@ class Operator {
     [[nodiscard]] Transaction* get_txn() const { return txn_; }
     [[nodiscard]] LockManager* get_lock_manager() const { return lock_manager_; }
 
-    void set_memory_resource(std::pmr::memory_resource* mr) { mr_ = mr; }
+    virtual void set_memory_resource(std::pmr::memory_resource* mr) { mr_ = mr; }
     [[nodiscard]] std::pmr::memory_resource* get_memory_resource() const {
         return mr_ ? mr_ : std::pmr::get_default_resource();
     }
 
-    void set_params(const std::vector<common::Value>* params) { params_ = params; }
+    virtual void set_params(const std::vector<common::Value>* params) { params_ = params; }
     [[nodiscard]] const std::vector<common::Value>* get_params() const { return params_; }
 
     virtual bool init() { return true; }
@@ -202,6 +202,9 @@ class FilterOperator : public Operator {
     void close() override;
     [[nodiscard]] Schema& output_schema() override;
     void add_child(std::unique_ptr<Operator> child) override;
+
+    void set_memory_resource(std::pmr::memory_resource* mr) override;
+    void set_params(const std::vector<common::Value>* params) override;
 };
 
 /**
@@ -223,6 +226,9 @@ class ProjectOperator : public Operator {
     void close() override;
     [[nodiscard]] Schema& output_schema() override;
     void add_child(std::unique_ptr<Operator> child) override;
+
+    void set_memory_resource(std::pmr::memory_resource* mr) override;
+    void set_params(const std::vector<common::Value>* params) override;
 };
 
 /**
@@ -247,6 +253,9 @@ class SortOperator : public Operator {
     bool next(Tuple& out_tuple) override;
     void close() override;
     [[nodiscard]] Schema& output_schema() override;
+
+    void set_memory_resource(std::pmr::memory_resource* mr) override;
+    void set_params(const std::vector<common::Value>* params) override;
 };
 
 /**
@@ -281,6 +290,9 @@ class AggregateOperator : public Operator {
     bool next(Tuple& out_tuple) override;
     void close() override;
     [[nodiscard]] Schema& output_schema() override;
+
+    void set_memory_resource(std::pmr::memory_resource* mr) override;
+    void set_params(const std::vector<common::Value>* params) override;
 };
 
 /**
@@ -330,6 +342,9 @@ class HashJoinOperator : public Operator {
     void close() override;
     [[nodiscard]] Schema& output_schema() override;
     void add_child(std::unique_ptr<Operator> child) override;
+
+    void set_memory_resource(std::pmr::memory_resource* mr) override;
+    void set_params(const std::vector<common::Value>* params) override;
 };
 
 /**
@@ -352,6 +367,9 @@ class LimitOperator : public Operator {
     void close() override;
     [[nodiscard]] Schema& output_schema() override;
     void add_child(std::unique_ptr<Operator> child) override;
+
+    void set_memory_resource(std::pmr::memory_resource* mr) override;
+    void set_params(const std::vector<common::Value>* params) override;
 };
 
 }  // namespace cloudsql::executor

include/executor/types.hpp

@@ -11,12 +11,12 @@
 #define CLOUDSQL_EXECUTOR_TYPES_HPP
 
 #include <cstdint>
-#include <initializer_list>
 #include <memory>
 #include <memory_resource>
 #include <stdexcept>
 #include <string>
 #include <vector>
+#include <initializer_list>
 
 #include "common/value.hpp"
 
@@ -131,24 +131,29 @@ class Tuple {
 
    public:
     Tuple() = default;
-
+    
     // Explicit PMR vector constructor
     explicit Tuple(std::pmr::vector<common::Value> values) : values_(std::move(values)) {}
-
-    // Initializer list constructor to resolve ambiguity for {...}
+    
+    // Initializer list constructor
     Tuple(std::initializer_list<common::Value> list) : values_(list) {}
 
     // Support allocation from a custom memory resource
-    explicit Tuple(std::pmr::memory_resource* mr) : values_(mr) {}
-
-    // Support construction from standard vector
-    Tuple(const std::vector<common::Value>& values) : values_(values.begin(), values.end()) {}
-    Tuple(std::vector<common::Value>&& values)
+    explicit Tuple(std::pmr::memory_resource* mr)
+        : values_(mr ? mr : std::pmr::get_default_resource()) {}
+    
+    // Support construction from standard vector or PMR vector with specific resource
+    template <typename VectorType,
+              typename = std::enable_if_t<!std::is_same_v<std::decay_t<VectorType>, Tuple>>,
+              typename std::enable_if_t<!std::is_same_v<std::decay_t<VectorType>, std::pmr::memory_resource*>>* = nullptr>
+    Tuple(const VectorType& values, std::pmr::memory_resource* mr = nullptr)
+        : values_(values.begin(), values.end(), mr ? mr : std::pmr::get_default_resource()) {}
+
+    template <typename VectorType,
+              typename = std::enable_if_t<!std::is_same_v<std::decay_t<VectorType>, Tuple>>>
+    explicit Tuple(VectorType&& values)
         : values_(std::make_move_iterator(values.begin()), std::make_move_iterator(values.end())) {}
 
-    Tuple(const std::vector<common::Value>& values, std::pmr::memory_resource* mr)
-        : values_(values.begin(), values.end(), mr) {}
-
     Tuple(const Tuple& other) = default;
     Tuple(Tuple&& other) noexcept = default;
     Tuple& operator=(const Tuple& other) = default;

include/storage/heap_table.hpp

@@ -104,7 +104,12 @@ class HeapTable {
 
        public:
         explicit Iterator(HeapTable& table, std::pmr::memory_resource* mr = nullptr);
+        ~Iterator() = default;
 
+        Iterator(const Iterator&) = default;
+        Iterator& operator=(const Iterator&) = default;
+        Iterator(Iterator&&) noexcept = default;
+        Iterator& operator=(Iterator&&) noexcept = default;
         /**
          * @brief Fetches the next non-deleted record from the heap
          * @param[out] out_tuple Container for the retrieved record