feat: prune OR/range, pool PostgreSQL, 2PC stress, honest docs

Prune shard_key OR-of-equalities and RANGE inequalities/BETWEEN.
Placeholders still scatter. ThreadSafeMultiRemoteExecutor pools
PostgreSQL as well as MySQL. engine_stress_test uses
DistributedTransactionManager. README matches current Session/ShardMap
and tool 2PC behavior.

Author: renecannao

AGENTS.md

@@ -7,7 +7,7 @@ Trust the `Makefile` over prose. Extension recipes live in `CLAUDE.md`. `docs/su
 - Parser: header-only templates in `include/sql_parser/` except `src/sql_parser/{arena,parser}.cpp`
 - Engine: headers in `include/sql_engine/` (`operators/`, `functions/`, `rules/`); compiled files are the explicit `ENGINE_SRCS` list
 - High-level API: `Session<D>` (`include/sql_engine/session.h`) — parse → plan → optimize → distribute → execute
-- Production remote path: `ThreadSafeMultiRemoteExecutor`, not the single-connection executors
+- Production remote path: `ThreadSafeMultiRemoteExecutor` (pooled MySQL **and** PostgreSQL), not the single-connection executors
 - All shard routing (SELECT prune and DML) goes through `ShardMap`. Do not add a private hash in the planner.
 - Backend URL / shard-spec parsing: `tool_config_parser` — do not add another copy in tools
 - Do not edit `third_party/`

README.md

@@ -93,7 +93,7 @@ echo "SELECT 1 + 2, UPPER('hello'), COALESCE(NULL, 42)" | ./sqlengine
 # Against a MySQL backend
 ./sqlengine --backend "mysql://root:pass@127.0.0.1:3306/mydb?name=primary"
 
-# Sharded across two backends
+# Sharded across two backends (2PC is on; optional --txn-log PATH)
 ./sqlengine \
   --backend "mysql://root:pass@host1:3306/db?name=shard1" \
   --backend "mysql://root:pass@host2:3306/db?name=shard2" \
@@ -171,20 +171,20 @@ ResultSet rs = executor.execute(plan);
 #include "sql_engine/session.h"
 #include "sql_engine/thread_safe_executor.h"
 #include "sql_engine/shard_map.h"
-#include "sql_engine/local_txn.h"
+#include "sql_engine/distributed_txn.h"
 
-// Backends (connection-pooled, thread-safe)
+// Backends (connection-pooled, thread-safe; MySQL or PostgreSQL)
 ThreadSafeMultiRemoteExecutor executor;
 executor.add_backend({.name = "shard1", .host = "h1", .port = 3306, ...});
 executor.add_backend({.name = "shard2", .host = "h2", .port = 3306, ...});
 
 // Sharding policy: "users" is sharded on "id" across shard1, shard2
 ShardMap shards;
-shards.add_sharded_table("users", "id", {"shard1", "shard2"});
+shards.add_table({"users", "id", {{"shard1"}, {"shard2"}}});
 
-// Catalog, transactions, session
+// Catalog + 2PC (required for atomic multi-shard DML)
 InMemoryCatalog catalog;  /* ... add_table(...) ... */
-LocalTransactionManager txn;
+DistributedTransactionManager txn(executor);
 Session<Dialect::MySQL> session(catalog, txn);
 session.set_remote_executor(&executor);
 session.set_shard_map(&shards);
@@ -354,11 +354,11 @@ auto report = recovery.recover();
 
 ### Distributed execution
 
-- **Shard routing** — shard-key lookups go to one backend; scatter queries go to all
-- **Distributed aggregation** — per-shard partial aggregates + coordinator merge (COUNT+SUM+MIN+MAX + AVG from SUM/COUNT)
-- **Distributed sort** — per-shard sort + coordinator merge
-- **Cross-shard joins** — hash-join coordinator; materialized subquery cache
-- **Cross-shard DML** — scatter INSERT/UPDATE/DELETE when no shard key; single-shard when key present
+- **Shard routing** — equality / `IN` / `OR` of equalities prune via `ShardMap`; RANGE also prunes `<`/`>`/`BETWEEN`. Placeholders scatter.
+- **Distributed aggregation** — per-shard partial aggregates + coordinator merge (`COUNT`/`SUM`/`MIN`/`MAX`/`AVG`). `COUNT(DISTINCT)` gathers then aggregates locally.
+- **Distributed sort** — per-shard sort + coordinator merge when keys are table columns
+- **Joins** — co-located same-key joins push down; otherwise gather both sides and join locally
+- **Cross-shard DML** — routed by `ShardMap`; missing/non-literal shard key and multi-table DML on shards fail closed
 - **Cross-shard INSERT ... SELECT** — source materialized, rows routed by destination shard key
 
 ### Transactions
@@ -372,10 +372,10 @@ auto report = recovery.recover();
 
 ### Backends & connectivity
 
-- **MySQL** — libmysqlclient with pooled and single-connection paths, UTF-8, configurable timeouts
-- **PostgreSQL** — libpq with statement_timeout, UTC-normalized TIMESTAMPTZ handling
+- **MySQL** — libmysqlclient with pooled (`ThreadSafeMultiRemoteExecutor`) and single-connection paths
+- **PostgreSQL** — libpq pooled on the same executor, plus a single-connection path; `statement_timeout` and UTC TIMESTAMPTZ
 - **SSL/TLS** — `ssl_mode`, `ssl_ca`, `ssl_cert`, `ssl_key` configurable per backend for both dialects
-- **Connection pool** — thread-safe with health checks, reconnection, RAII `ConnectionGuard`
+- **Connection pool** — thread-safe per dialect, RAII checkout, poison-on-error
 - **MySQL wire-protocol server** — `mysql_server` speaks the MySQL protocol; backends are ParserSQL engines
 
 ### Thread-safety
@@ -388,7 +388,7 @@ auto report = recovery.recover();
 
 | Tool | Build | Purpose |
 |---|---|---|
-| `sqlengine` | `make build-sqlengine` | Interactive SQL CLI; stdin, one-shot, or REPL; optional backends and sharding |
+| `sqlengine` | `make build-sqlengine` | Interactive SQL CLI; 2PC when `--backend` is set; optional `--txn-log` |
 | `mysql_server` | `make mysql-server` | MySQL wire-protocol server fronted by the ParserSQL engine |
 | `corpus_test` | `make build-corpus-test` | Read SQL from stdin/files, parse each, report OK/PARTIAL/ERROR |
 | `engine_stress_test` | `make engine-stress` | Direct-API engine stress test |

include/sql_engine/distributed_planner.h

@@ -18,6 +18,7 @@
 #include <cstring>
 #include <cstdio>
 #include <cstdlib>
+#include <climits>
 #include <vector>
 #include <unordered_map>
 #include <functional>
@@ -401,6 +402,62 @@ class DistributedPlanner {
                     }
                 }
             }
+            if (is_compare_op(op) &&
+                shards_.routing_strategy(table_name) == RoutingStrategy::RANGE) {
+                const sql_parser::AstNode* left_node = expr->first_child;
+                const sql_parser::AstNode* right_node = left_node ? left_node->next_sibling : nullptr;
+                if (left_node && right_node) {
+                    const sql_parser::AstNode* col = nullptr;
+                    const sql_parser::AstNode* lit = nullptr;
+                    bool key_on_left = false;
+                    if (is_shard_key_ref(left_node, shard_key) && is_literal(right_node)) {
+                        col = left_node; lit = right_node; key_on_left = true;
+                    } else if (is_shard_key_ref(right_node, shard_key) && is_literal(left_node)) {
+                        col = right_node; lit = left_node; key_on_left = false;
+                    }
+                    if (col && lit) {
+                        int64_t v = literal_to_int(lit);
+                        int64_t lo = INT64_MIN, hi = INT64_MAX;
+                        char c0 = op.ptr[0];
+                        bool has_eq = op.len == 2 && op.ptr[1] == '=';
+                        if (c0 == '<' && key_on_left) {
+                            hi = has_eq ? v : (v == INT64_MIN ? INT64_MIN : v - 1);
+                        } else if (c0 == '>' && key_on_left) {
+                            lo = has_eq ? v : (v == INT64_MAX ? INT64_MAX : v + 1);
+                        } else if (c0 == '<' && !key_on_left) {
+                            lo = has_eq ? v : (v == INT64_MAX ? INT64_MAX : v + 1);
+                        } else if (c0 == '>' && !key_on_left) {
+                            hi = has_eq ? v : (v == INT64_MIN ? INT64_MIN : v - 1);
+                        }
+                        shards_.collect_int_range_shards(table_name, lo, hi, target_indices);
+                        return;
+                    }
+                }
+            }
+            if (op.len == 2 &&
+                (op.ptr[0] == 'O' || op.ptr[0] == 'o') &&
+                (op.ptr[1] == 'R' || op.ptr[1] == 'r')) {
+                const sql_parser::AstNode* left_node = expr->first_child;
+                const sql_parser::AstNode* right_node = left_node ? left_node->next_sibling : nullptr;
+                std::vector<size_t> left_targets, right_targets;
+                extract_shard_targets(left_node, shard_key, table_name, num_shards, left_targets);
+                extract_shard_targets(right_node, shard_key, table_name, num_shards, right_targets);
+                if (left_targets.empty() || right_targets.empty()) return;
+                std::vector<bool> seen(num_shards, false);
+                for (auto i : left_targets) {
+                    if (i < num_shards && !seen[i]) {
+                        seen[i] = true;
+                        target_indices.push_back(i);
+                    }
+                }
+                for (auto i : right_targets) {
+                    if (i < num_shards && !seen[i]) {
+                        seen[i] = true;
+                        target_indices.push_back(i);
+                    }
+                }
+                return;
+            }
             // Recurse into AND branches
             if (op.len == 3 &&
                 (op.ptr[0] == 'A' || op.ptr[0] == 'a') &&
@@ -446,6 +503,29 @@ class DistributedPlanner {
                 }
             }
         }
+
+        if (expr->type == sql_parser::NodeType::NODE_BETWEEN &&
+            shards_.routing_strategy(table_name) == RoutingStrategy::RANGE) {
+            const sql_parser::AstNode* col = expr->first_child;
+            const sql_parser::AstNode* lo = col ? col->next_sibling : nullptr;
+            const sql_parser::AstNode* hi = lo ? lo->next_sibling : nullptr;
+            if (col && is_shard_key_ref(col, shard_key) && is_literal(lo) && is_literal(hi)) {
+                shards_.collect_int_range_shards(table_name,
+                                                 literal_to_int(lo), literal_to_int(hi),
+                                                 target_indices);
+            }
+        }
+    }
+
+    static bool is_compare_op(sql_parser::StringRef op) {
+        if (op.len == 1) return op.ptr[0] == '<' || op.ptr[0] == '>';
+        if (op.len == 2) return (op.ptr[0] == '<' || op.ptr[0] == '>') && op.ptr[1] == '=';
+        return false;
+    }
+
+    static int64_t literal_to_int(const sql_parser::AstNode* lit) {
+        if (!lit || !lit->value().ptr) return 0;
+        return std::strtoll(lit->value().ptr, nullptr, 10);
     }
 
     bool is_shard_key_ref(const sql_parser::AstNode* node, sql_parser::StringRef shard_key) const {

include/sql_engine/pg_connection_pool.h

@@ -0,0 +1,108 @@
+#ifndef SQL_ENGINE_PG_CONNECTION_POOL_H
+#define SQL_ENGINE_PG_CONNECTION_POOL_H
+
+#include "sql_engine/backend_config.h"
+#include <libpq-fe.h>
+#include <mutex>
+#include <unordered_map>
+#include <vector>
+#include <string>
+#include <stdexcept>
+#include <memory>
+
+#ifndef SQL_ENGINE_PG_STATEMENT_TIMEOUT_MS
+#define SQL_ENGINE_PG_STATEMENT_TIMEOUT_MS 30000
+#endif
+
+namespace sql_engine {
+
+class PgConnectionPool {
+public:
+    PgConnectionPool() = default;
+
+    ~PgConnectionPool() {
+        for (auto& kv : backends_) {
+            auto& be = *kv.second;
+            std::lock_guard<std::mutex> lk(be.mu);
+            for (PGconn* c : be.idle) {
+                if (c) PQfinish(c);
+            }
+        }
+    }
+
+    void add_backend(const BackendConfig& config) {
+        auto be = std::make_unique<Backend>();
+        be->config = config;
+        backends_[config.name] = std::move(be);
+    }
+
+    bool has_backend(const std::string& name) const {
+        return backends_.find(name) != backends_.end();
+    }
+
+    PGconn* checkout(const std::string& backend) {
+        Backend& be = get_backend(backend);
+        {
+            std::lock_guard<std::mutex> lk(be.mu);
+            if (!be.idle.empty()) {
+                PGconn* c = be.idle.back();
+                be.idle.pop_back();
+                if (c && PQstatus(c) == CONNECTION_OK) return c;
+                if (c) PQfinish(c);
+            }
+        }
+        return create_connection(be);
+    }
+
+    void checkin(const std::string& backend, PGconn* conn) {
+        if (!conn) return;
+        Backend& be = get_backend(backend);
+        std::lock_guard<std::mutex> lk(be.mu);
+        be.idle.push_back(conn);
+    }
+
+private:
+    struct Backend {
+        BackendConfig config;
+        std::mutex mu;
+        std::vector<PGconn*> idle;
+    };
+
+    std::unordered_map<std::string, std::unique_ptr<Backend>> backends_;
+
+    Backend& get_backend(const std::string& name) {
+        auto it = backends_.find(name);
+        if (it == backends_.end()) {
+            throw std::runtime_error("PgConnectionPool: unknown backend: " + name);
+        }
+        return *it->second;
+    }
+
+    static PGconn* create_connection(Backend& be) {
+        const BackendConfig& cfg = be.config;
+        std::string conninfo = "host=" + cfg.host
+            + " port=" + std::to_string(cfg.port)
+            + " user=" + cfg.user
+            + " password=" + cfg.password
+            + " dbname=" + cfg.database
+            + " connect_timeout=5"
+            + " options='-c statement_timeout="
+                + std::to_string(SQL_ENGINE_PG_STATEMENT_TIMEOUT_MS) + "'";
+        if (!cfg.ssl_mode.empty()) conninfo += " sslmode=" + cfg.ssl_mode;
+        if (!cfg.ssl_ca.empty()) conninfo += " sslrootcert=" + cfg.ssl_ca;
+        if (!cfg.ssl_cert.empty()) conninfo += " sslcert=" + cfg.ssl_cert;
+        if (!cfg.ssl_key.empty()) conninfo += " sslkey=" + cfg.ssl_key;
+
+        PGconn* c = PQconnectdb(conninfo.c_str());
+        if (PQstatus(c) != CONNECTION_OK) {
+            std::string err = PQerrorMessage(c);
+            PQfinish(c);
+            throw std::runtime_error("PgConnectionPool connect failed for " + cfg.name + ": " + err);
+        }
+        return c;
+    }
+};
+
+} // namespace sql_engine
+
+#endif

include/sql_engine/shard_map.h

@@ -3,6 +3,7 @@
 
 #include "sql_parser/common.h"
 #include <cstdint>
+#include <climits>
 #include <cstring>
 #include <string>
 #include <vector>
@@ -155,6 +156,32 @@ class ShardMap {
         return 0;
     }
 
+    RoutingStrategy routing_strategy(sql_parser::StringRef table_name) const {
+        const TableShardConfig* cfg = lookup(table_name);
+        return cfg ? cfg->strategy : RoutingStrategy::HASH;
+    }
+
+    // RANGE only. Inclusive [lo, hi]. HASH/LIST yield no indices (caller scatters).
+    void collect_int_range_shards(sql_parser::StringRef table_name,
+                                  int64_t lo, int64_t hi,
+                                  std::vector<size_t>& out) const {
+        const TableShardConfig* cfg = lookup(table_name);
+        if (!cfg || cfg->strategy != RoutingStrategy::RANGE || cfg->ranges.empty())
+            return;
+        if (lo > hi) return;
+        size_t n = cfg->shards.size();
+        const auto& ranges = cfg->ranges;
+        for (size_t i = 0; i < ranges.size(); ++i) {
+            int64_t seg_hi = (i + 1 == ranges.size())
+                ? INT64_MAX : ranges[i].upper_inclusive;
+            int64_t seg_lo = (i == 0) ? INT64_MIN
+                : (cfg->ranges[i - 1].upper_inclusive == INT64_MAX
+                   ? INT64_MAX : cfg->ranges[i - 1].upper_inclusive + 1);
+            if (seg_lo <= hi && seg_hi >= lo)
+                out.push_back(clamp_index(ranges[i].shard_index, n));
+        }
+    }
+
     bool same_routing(sql_parser::StringRef a, sql_parser::StringRef b) const {
         const TableShardConfig* ca = lookup(a);
         const TableShardConfig* cb = lookup(b);