fix: silent correctness bugs and resource safety from codebase review

Six fixes flagged in the recent stubs/incomplete-features review.
None change behavior for queries that already worked; all tests still pass
(1047/1084, 37 skipped require Docker, 26 new tests, 0 failures).

1. mysql_server: format DATE/TIME/DATETIME/TIMESTAMP/INTERVAL values
   value_to_string had no cases for the temporal Value tags, so date and
   time columns rendered as empty strings over the wire protocol. Add
   format_date / format_time / format_datetime helpers in datetime_parse
   (using Howard Hinnant's civil_from_days algorithm) and wire them in.
   Also map TAG_DATE/TIME/TIMESTAMP to the proper MySQL field type codes
   in column metadata instead of falling back to VAR_STRING.

2. SetOpOperator: validate column counts across UNION/INTERSECT/EXCEPT
   Mismatched column counts on the two sides of a set operation used to
   silently truncate via row_key(), producing wrong results. Track the
   first row's column_count and throw runtime_error on any mismatch.

3. NestedLoopJoinOperator / HashJoinOperator: reject unsupported join types
   RIGHT and FULL joins (and SEMI/ANTI in HashJoin) silently fell through
   to the INNER code path, returning wrong rows. Throw runtime_error in
   open() with a clear "not yet supported" message until they're properly
   implemented.

4. ThreadSafeMultiRemoteExecutor: RAII for connection + result set
   Replace manual checkout/checkin with ConnectionGuard (returns on dtor,
   closes-and-discards if poisoned) and ResultGuard (frees MYSQL_RES on
   dtor). Connections that hit a query error or exception are now poisoned
   instead of being returned to the pool in a possibly-broken state, and
   MYSQL_RES can no longer leak on a throwing result conversion.

5. Session: bounded LRU plan cache
   Plan cache used to be an unordered_map with no eviction, growing
   forever for high-QPS workloads with varied SQL. Replace with an
   intrusive list + map LRU bounded at 1024 entries by default. Cache
   size is configurable via set_plan_cache_max_size(0 disables caching).

6. ThreadPool: document the exception-safety invariant
   The submit() path wraps callables in std::packaged_task, which catches
   exceptions and stores them in the future for later get(). The agent
   review claimed worker threads could die silently on a thrown task --
   that's not true today, but it would become true if a future change
   queued raw callables. Add a comment locking in the invariant so
   nobody breaks it accidentally.

New tests:
- tests/test_datetime_format.cpp: 19 tests covering parse/format
  round-trips for date, time, datetime including dates before the epoch
  and a 200-year sweep validating days_to_ymd against days_since_epoch.
- tests/test_operators.cpp: 4 new tests for SetOp column-count rejection
  (UNION ALL and INTERSECT) and JoinOp RIGHT/FULL rejection.
- tests/test_session.cpp: 3 new tests for plan cache size, LRU eviction,
  and disabled-cache mode.

Note: a separate review claim that mysql_server.cpp:711 contained a
"sequence number double-increment bug" was investigated and is incorrect.
The MySQL protocol requires server responses to use client_seq + 1, and
the increment at line 711 correctly advances from the client's seq to
the server's first response seq. No change made there.

Author: renecannao

Makefile

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

include/sql_engine/datetime_parse.h

@@ -2,6 +2,7 @@
 #define SQL_ENGINE_DATETIME_PARSE_H
 
 #include <cstdint>
+#include <cstddef>
 
 namespace sql_engine {
 namespace datetime_parse {
@@ -19,6 +20,22 @@ int64_t parse_time(const char* s);
 // Handles leap years correctly.
 int32_t days_since_epoch(int year, int month, int day);
 
+// Calendar math: decompose days since 1970-01-01 into year/month/day.
+// Handles leap years and dates before the epoch correctly.
+void days_to_ymd(int32_t days, int& year, int& month, int& day);
+
+// Format days since epoch as "YYYY-MM-DD" into buf (needs >= 11 bytes including NUL).
+// Returns number of bytes written (excluding NUL).
+size_t format_date(int32_t days, char* buf, size_t buf_len);
+
+// Format microseconds since midnight as "HH:MM:SS" or "HH:MM:SS.uuuuuu".
+// Returns number of bytes written (excluding NUL).
+size_t format_time(int64_t us_since_midnight, char* buf, size_t buf_len);
+
+// Format microseconds since epoch as "YYYY-MM-DD HH:MM:SS[.uuuuuu]".
+// Returns number of bytes written (excluding NUL).
+size_t format_datetime(int64_t us_since_epoch, char* buf, size_t buf_len);
+
 } // namespace datetime_parse
 } // namespace sql_engine
 

include/sql_engine/operators/hash_join_op.h

@@ -9,6 +9,7 @@
 #include <functional>
 #include <vector>
 #include <unordered_map>
+#include <stdexcept>
 
 namespace sql_engine {
 
@@ -33,6 +34,18 @@ class HashJoinOperator : public Operator {
           arena_(arena) {}
 
     void open() override {
+        // Only INNER and LEFT equi-joins are actually supported below.
+        // RIGHT/FULL/CROSS would silently fall through to INNER logic and
+        // return wrong results. Reject explicitly. (CROSS also doesn't
+        // belong here because it's not an equi-join; PlanExecutor's
+        // build_join should never route it to the hash join operator.)
+        if (join_type_ != JOIN_INNER && join_type_ != JOIN_LEFT) {
+            throw std::runtime_error(
+                "join type not supported by HashJoinOperator "
+                "(only INNER and LEFT equi-joins are implemented; "
+                "RIGHT, FULL and CROSS joins are not yet supported)");
+        }
+
         // Build phase: consume right side into hash table keyed by join column
         hash_table_.clear();
         right_->open();

include/sql_engine/operators/join_op.h

@@ -6,6 +6,7 @@
 #include "sql_engine/catalog.h"
 #include "sql_engine/plan_node.h"
 #include "sql_parser/arena.h"
+#include <stdexcept>
 #include <functional>
 #include <vector>
 
@@ -29,6 +30,17 @@ class NestedLoopJoinOperator : public Operator {
           functions_(functions), arena_(arena) {}
 
     void open() override {
+        // Only INNER, LEFT and CROSS joins are actually implemented below.
+        // RIGHT/FULL would previously fall through to the INNER code path,
+        // silently producing wrong results. Reject them explicitly instead
+        // of corrupting query answers.
+        if (join_type_ != JOIN_INNER && join_type_ != JOIN_LEFT && join_type_ != JOIN_CROSS) {
+            throw std::runtime_error(
+                "join type not supported by NestedLoopJoinOperator "
+                "(only INNER, LEFT, CROSS are implemented; "
+                "RIGHT and FULL joins are not yet supported)");
+        }
+
         // Materialize right side
         right_->open();
         right_rows_.clear();

include/sql_engine/operators/set_op_op.h

@@ -8,6 +8,7 @@
 #include <string>
 #include <vector>
 #include <future>
+#include <stdexcept>
 
 namespace sql_engine {
 
@@ -37,12 +38,14 @@ class SetOpOperator : public Operator {
         }
         reading_left_ = true;
         seen_.clear();
+        expected_col_count_ = -1;
 
         if (op_ == SET_OP_INTERSECT || op_ == SET_OP_EXCEPT) {
             // Materialize right side into a set
             right_set_.clear();
             Row r{};
             while (right_->next(r)) {
+                check_col_count(r);
                 right_set_.insert(row_key(r));
             }
             right_->close();
@@ -63,6 +66,7 @@ class SetOpOperator : public Operator {
                     if (!got) return false;
                 }
                 if (got) {
+                    check_col_count(out);
                     std::string key = row_key(out);
                     if (seen_.insert(key).second) return true;
                 }
@@ -72,15 +76,23 @@ class SetOpOperator : public Operator {
         if (op_ == SET_OP_UNION && all_) {
             // UNION ALL: yield left then right
             if (reading_left_) {
-                if (left_->next(out)) return true;
+                if (left_->next(out)) {
+                    check_col_count(out);
+                    return true;
+                }
                 reading_left_ = false;
             }
-            return right_->next(out);
+            if (right_->next(out)) {
+                check_col_count(out);
+                return true;
+            }
+            return false;
         }
 
         if (op_ == SET_OP_INTERSECT) {
             // Yield left rows that also appear in right
             while (left_->next(out)) {
+                check_col_count(out);
                 std::string key = row_key(out);
                 if (right_set_.count(key)) {
                     if (all_ || seen_.insert(key).second)
@@ -93,6 +105,7 @@ class SetOpOperator : public Operator {
         if (op_ == SET_OP_EXCEPT) {
             // Yield left rows that don't appear in right
             while (left_->next(out)) {
+                check_col_count(out);
                 std::string key = row_key(out);
                 if (!right_set_.count(key)) {
                     if (all_ || seen_.insert(key).second)
@@ -122,6 +135,26 @@ class SetOpOperator : public Operator {
     bool reading_left_ = true;
     std::unordered_set<std::string> seen_;
     std::unordered_set<std::string> right_set_;
+    // Column count established by the first row we see. Used to detect
+    // schema mismatches between the two sides of a UNION/INTERSECT/EXCEPT --
+    // which the SQL standard says must have the same number of columns.
+    // Previously, mismatched column counts silently produced wrong results
+    // because row_key() would iterate a different number of values on each
+    // side. Now we throw a clear error at execution time.
+    int16_t expected_col_count_ = -1;
+
+    void check_col_count(const Row& row) {
+        if (expected_col_count_ < 0) {
+            expected_col_count_ = static_cast<int16_t>(row.column_count);
+            return;
+        }
+        if (row.column_count != static_cast<uint16_t>(expected_col_count_)) {
+            throw std::runtime_error(
+                "set operation column count mismatch: expected " +
+                std::to_string(expected_col_count_) + " columns, got " +
+                std::to_string(row.column_count));
+        }
+    }
 
     static std::string row_key(const Row& row) {
         std::string key;

include/sql_engine/session.h

@@ -19,6 +19,7 @@
 #include <cstring>
 #include <string>
 #include <unordered_map>
+#include <list>
 #include <memory>
 
 namespace sql_engine {
@@ -72,17 +73,25 @@ class Session {
         pool_ = pool;
     }
 
+    // Configure the maximum number of cached plans. When the cache is full,
+    // the least-recently-used entry is evicted on insert. Default is 1024.
+    // Setting to 0 disables caching entirely.
+    void set_plan_cache_max_size(size_t n) { plan_cache_max_size_ = n; }
+    size_t plan_cache_size() const { return plan_cache_.size(); }
+
     // Execute a SELECT query. Returns a ResultSet.
     // Uses plan caching: repeated identical SQL strings skip parse/plan/optimize/distribute.
     ResultSet execute_query(const char* sql, size_t len) {
         // Check plan cache first
         std::string sql_key(sql, len);
         auto cache_it = plan_cache_.find(sql_key);
         if (cache_it != plan_cache_.end()) {
-            // Cache hit: reuse the cached plan. Use exec_arena_ for per-query
-            // allocations (rows, operator internals). Reset it each time.
+            // Cache hit: move this entry to the front of the LRU list.
+            plan_cache_order_.splice(plan_cache_order_.begin(),
+                                     plan_cache_order_,
+                                     cache_it->second);
             exec_arena_.reset();
-            auto& entry = cache_it->second;
+            auto& entry = *cache_it->second;
             PlanExecutor<D> executor(functions_, catalog_, exec_arena_);
             wire_executor(executor);
             return executor.execute(entry.plan);
@@ -115,11 +124,10 @@ class Session {
         wire_executor(executor);
         ResultSet rs = executor.execute(plan);
 
-        // Cache the plan (parser arena keeps plan tree and strings alive)
-        CachedPlan entry;
-        entry.parser = std::move(cached_parser);
-        entry.plan = plan;
-        plan_cache_.emplace(std::move(sql_key), std::move(entry));
+        // Cache the plan, enforcing the LRU bound. The parser arena is kept
+        // alive via unique_ptr stored in the CachedPlan so all plan/AST
+        // string pointers remain valid for subsequent cache hits.
+        insert_into_plan_cache(std::move(sql_key), std::move(cached_parser), plan);
 
         return rs;
     }
@@ -246,13 +254,45 @@ class Session {
     // Externally owned; set via set_thread_pool(). Shared across sessions.
     ThreadPool* pool_ = nullptr;
 
-    // Plan cache: maps SQL string → cached plan + parser arena.
-    // The parser's arena keeps all AST/plan string pointers valid.
+    // Plan cache: bounded LRU keyed by SQL string. Each entry owns the
+    // parser whose arena keeps the plan tree and all AST/plan string
+    // pointers alive for as long as the entry stays in the cache.
+    //
+    // Implementation: a list keeps insertion/use order (front = most
+    // recently used) and a hash map maps each SQL string to its iterator
+    // in the list, giving O(1) lookup, O(1) move-to-front on hit, and
+    // O(1) eviction of the LRU entry on insert.
     struct CachedPlan {
+        std::string key;
         std::unique_ptr<sql_parser::Parser<D>> parser;
         PlanNode* plan;
     };
-    std::unordered_map<std::string, CachedPlan> plan_cache_;
+    using CacheList = std::list<CachedPlan>;
+    using CacheIter = typename CacheList::iterator;
+    CacheList plan_cache_order_;
+    std::unordered_map<std::string, CacheIter> plan_cache_;
+    size_t plan_cache_max_size_ = 1024;
+
+    void insert_into_plan_cache(std::string key,
+                                std::unique_ptr<sql_parser::Parser<D>> parser,
+                                PlanNode* plan) {
+        if (plan_cache_max_size_ == 0) return;  // caching disabled
+        // Evict LRU entries until we're within budget. We evict before insert
+        // so the new entry counts against the cap and we never exceed it.
+        while (plan_cache_.size() >= plan_cache_max_size_ && !plan_cache_order_.empty()) {
+            const std::string& victim_key = plan_cache_order_.back().key;
+            plan_cache_.erase(victim_key);
+            plan_cache_order_.pop_back();
+        }
+        CachedPlan entry;
+        entry.key = std::move(key);
+        entry.parser = std::move(parser);
+        entry.plan = plan;
+        plan_cache_order_.push_front(std::move(entry));
+        // The map stores the iterator and a copy of the key (so the lookup
+        // string and the entry's owned key both remain valid through moves).
+        plan_cache_[plan_cache_order_.front().key] = plan_cache_order_.begin();
+    }
 
     void wire_executor(PlanExecutor<D>& executor) {
         for (auto& kv : sources_)

include/sql_engine/thread_pool.h

@@ -38,6 +38,13 @@ class ThreadPool {
                         task = std::move(tasks_.front());
                         tasks_.pop();
                     }
+                    // Exception safety: submit() wraps callables in
+                    // std::packaged_task, which catches any exception and
+                    // stores it in the shared state to be rethrown by
+                    // future::get(). `task()` therefore does not throw, so
+                    // the worker loop cannot die on a user exception. If
+                    // submit() is ever changed to queue raw callables, this
+                    // must be wrapped in try { task(); } catch (...) { log; }.
                     task();
                 }
             });