fix: close P0 sharding holes from the #56 review

- Detect qualified SET t.id as a shard-key assignment and move the row.
- Do not fall back to in-place UPDATE when a move SELECT finds no rows.
- Empty IN (subquery) rewrites to FALSE instead of per-shard subquery.
- Copy subquery string values into the arena (no ResultSet UAF).
- Do not mutate the cached logical AGGREGATE when redistributing.
- HASH routes integer-looking strings like ints.
- Semi-join prune is INNER only; colocated joins emit LEFT/RIGHT/FULL.
- Planner-time SELECTs go through TxnRoutingExecutor.

Author: renecannao

include/sql_engine/distributed_planner.h

@@ -159,8 +159,11 @@ class DistributedPlanner {
                     agg_child = agg_child->left;
                 }
                 if (agg_child && agg_child->type == PlanNodeType::AGGREGATE) {
-                    push_agg_exprs_from_project(node, agg_child);
-                    PlanNode* dist_agg = distribute_aggregate(agg_child);
+                    PlanNode* agg_copy = make_plan_node(arena_, PlanNodeType::AGGREGATE);
+                    agg_copy->aggregate = agg_child->aggregate;
+                    agg_copy->left = agg_child->left;
+                    push_agg_exprs_from_project(node, agg_copy);
+                    PlanNode* dist_agg = distribute_aggregate(agg_copy);
                     if (dist_agg && (dist_agg->type == PlanNodeType::MERGE_AGGREGATE ||
                                      dist_agg->type == PlanNodeType::AGGREGATE)) {
                         PlanNode* top = dist_agg;
@@ -1193,7 +1196,8 @@ class DistributedPlanner {
         PlanNode* current = nullptr;
         for (const auto& shard : shard_list) {
             sql_parser::StringRef sql = qb_.build_select_join(
-                left_table, right_table, join_node->join.condition, where_expr);
+                left_table, right_table, join_node->join.condition, where_expr,
+                join_node->join.join_type);
             PlanNode* rs = make_remote_scan(shard.backend_name.c_str(), sql, left_table);
             if (!current) {
                 current = rs;
@@ -1310,6 +1314,7 @@ class DistributedPlanner {
                                  const TableInfo* right_table) {
         if (!join_node || !remote_executor_ || !join_node->join.condition)
             return nullptr;
+        if (join_node->join.join_type != JOIN_INNER) return nullptr;
         if (!left_table || !right_table) return nullptr;
 
         bool ls = shards_.is_sharded(left_table->table_name);
@@ -1659,7 +1664,7 @@ class DistributedPlanner {
                            sql_parser::StringRef shard_key) const {
         if (!set_columns || !shard_key.ptr) return false;
         for (uint16_t i = 0; i < set_count; ++i) {
-            if (is_column_ref(set_columns[i], shard_key)) return true;
+            if (is_shard_key_ref(set_columns[i], shard_key)) return true;
         }
         return false;
     }
@@ -1844,7 +1849,7 @@ class DistributedPlanner {
         auto resolve = make_resolver(catalog_, table, src.values);
         for (uint16_t i = 0; i < set_count; ++i) {
             if (!set_cols[i]) continue;
-            const ColumnInfo* col = catalog_.get_column(table, set_cols[i]->value());
+            const ColumnInfo* col = catalog_.get_column(table, set_col_name(set_cols[i]));
             if (!col) continue;
             Value nv = value_null();
             if (functions_) {
@@ -1950,9 +1955,7 @@ class DistributedPlanner {
         }
 
         if (moves.empty()) {
-            sql_parser::StringRef sql = qb_.build_update(
-                table, up.set_columns, up.set_exprs, up.set_count, where_expr);
-            return make_remote_scan(pruned[0].backend_name.c_str(), sql, table);
+            return make_noop_update(table, pruned[0].backend_name.c_str());
         }
 
         bool any_move = false;
@@ -1999,6 +2002,31 @@ class DistributedPlanner {
         return current ? current : plan;
     }
 
+    static sql_parser::StringRef set_col_name(const sql_parser::AstNode* node) {
+        if (!node) return sql_parser::StringRef{nullptr, 0};
+        if (node->type == sql_parser::NodeType::NODE_QUALIFIED_NAME) {
+            const sql_parser::AstNode* c = node->first_child;
+            if (c && c->next_sibling) return c->next_sibling->value();
+        }
+        return node->value();
+    }
+
+    PlanNode* make_noop_update(const TableInfo* table, const char* backend) {
+        sql_parser::StringBuilder sb(arena_, 64);
+        sb.append("UPDATE ");
+        if (table) sb.append(table->table_name.ptr, table->table_name.len);
+        sb.append(" SET ");
+        if (table && table->column_count > 0) {
+            sb.append(table->columns[0].name.ptr, table->columns[0].name.len);
+            sb.append(" = ");
+            sb.append(table->columns[0].name.ptr, table->columns[0].name.len);
+        } else {
+            sb.append("id = id");
+        }
+        sb.append(" WHERE 1 = 0");
+        return make_remote_scan(backend, sb.finish(), table);
+    }
+
     bool is_column_ref(const sql_parser::AstNode* node, sql_parser::StringRef col_name) const {
         if (!node) return false;
         if (node->type == sql_parser::NodeType::NODE_COLUMN_REF ||
@@ -2175,7 +2203,7 @@ class DistributedPlanner {
 
         for (const auto& row : rs.rows) {
             if (row.column_count > 0) {
-                result.push_back(row.get(0));
+                result.push_back(copy_value_arena(row.get(0)));
             }
         }
         return result;
@@ -2249,8 +2277,9 @@ class DistributedPlanner {
                 lit = sql_parser::make_node(arena_, sql_parser::NodeType::NODE_LITERAL_INT,
                                              sql_parser::StringRef{s, static_cast<uint32_t>(n)});
             } else if (v.tag == Value::TAG_STRING && v.str_val.ptr) {
+                Value owned = copy_value_arena(v);
                 lit = sql_parser::make_node(arena_, sql_parser::NodeType::NODE_LITERAL_STRING,
-                                             v.str_val);
+                                             owned.str_val);
             } else if (v.tag == Value::TAG_DOUBLE) {
                 char buf[64];
                 int n = snprintf(buf, sizeof(buf), "%g", v.double_val);
@@ -2269,6 +2298,19 @@ class DistributedPlanner {
         return new_in;
     }
 
+    sql_parser::AstNode* make_false_pred() {
+        sql_parser::AstNode* eq = sql_parser::make_node(
+            arena_, sql_parser::NodeType::NODE_BINARY_OP,
+            sql_parser::StringRef{"=", 1});
+        eq->add_child(sql_parser::make_node(
+            arena_, sql_parser::NodeType::NODE_LITERAL_INT,
+            sql_parser::StringRef{"0", 1}));
+        eq->add_child(sql_parser::make_node(
+            arena_, sql_parser::NodeType::NODE_LITERAL_INT,
+            sql_parser::StringRef{"1", 1}));
+        return eq;
+    }
+
     // Rewrite a WHERE expression by replacing the first IN (subquery) with IN (literals).
     // Returns the rewritten expression, or the original if no rewrite needed.
     const sql_parser::AstNode* rewrite_where_subquery(
@@ -2286,7 +2328,7 @@ class DistributedPlanner {
                     if (!values.empty()) {
                         return build_in_list_from_values(where_expr, values);
                     }
-                    return where_expr;
+                    return make_false_pred();
                 }
             }
         }

include/sql_engine/remote_query_builder.h

@@ -98,12 +98,16 @@ class RemoteQueryBuilder {
             const TableInfo* left,
             const TableInfo* right,
             const sql_parser::AstNode* on_expr,
-            const sql_parser::AstNode* where_expr)
+            const sql_parser::AstNode* where_expr,
+            uint8_t join_type = 0)
     {
         sql_parser::StringBuilder sb(arena_, 512);
         sb.append("SELECT * FROM ");
         if (left) sb.append(left->table_name.ptr, left->table_name.len);
-        sb.append(" JOIN ");
+        if (join_type == 1) sb.append(" LEFT JOIN ");
+        else if (join_type == 2) sb.append(" RIGHT JOIN ");
+        else if (join_type == 3) sb.append(" FULL JOIN ");
+        else sb.append(" JOIN ");
         if (right) sb.append(right->table_name.ptr, right->table_name.len);
         if (on_expr) {
             sb.append(" ON ");

include/sql_engine/session.h

@@ -256,8 +256,9 @@ class Session {
 
         // If sharding is configured, distribute DML to remote backends.
         if (shard_map_ && remote_executor_) {
+            routing_exec_.bind(remote_executor_, &txn_mgr_);
             DistributedPlanner<D> dp(*shard_map_, catalog_, parser_.arena(),
-                                      remote_executor_, &functions_);
+                                      &routing_exec_, &functions_);
             PlanNode* dist_plan = dp.distribute_dml(plan);
 
             if (dp.last_error()) {
@@ -278,6 +279,7 @@ class Session {
                 result.success = true;
                 result.affected_rows = 0;
                 for_each_remote_scan(dist_plan, [&](const PlanNode* rs) {
+                    if (!result.success) return;
                     sql_parser::StringRef s{rs->remote_scan.remote_sql,
                                             rs->remote_scan.remote_sql_len};
                     DmlResult shard_result;
@@ -377,8 +379,9 @@ class Session {
 
     PlanNode* maybe_distribute(PlanNode* plan, sql_parser::Arena& arena) {
         if (!plan || !shard_map_ || !remote_executor_) return plan;
+        routing_exec_.bind(remote_executor_, &txn_mgr_);
         DistributedPlanner<D> dplanner(*shard_map_, catalog_, arena,
-                                       remote_executor_, &functions_);
+                                       &routing_exec_, &functions_);
         PlanNode* dist = dplanner.distribute(plan);
         if (dplanner.last_error()) return nullptr;
         return dist;

include/sql_engine/shard_map.h

@@ -9,6 +9,7 @@
 #include <vector>
 #include <unordered_map>
 #include <algorithm>
+#include <cstdlib>
 
 namespace sql_engine {
 
@@ -154,9 +155,13 @@ class ShardMap {
         if (!cfg || cfg->shards.empty()) return false;
         size_t n = cfg->shards.size();
         switch (cfg->strategy) {
-            case RoutingStrategy::HASH:
+            case RoutingStrategy::HASH: {
+                int64_t as_int = 0;
+                if (parse_full_int(val, val_len, as_int))
+                    return try_shard_index_for_int(table_name, as_int, out);
                 out = fnv1a_bytes(reinterpret_cast<const uint8_t*>(val), val_len) % n;
                 return true;
+            }
             case RoutingStrategy::RANGE:
                 return false;
             case RoutingStrategy::LIST:
@@ -241,8 +246,12 @@ class ShardMap {
         if (!cfg || cfg->shards.empty()) return 0;
         size_t n = cfg->shards.size();
         switch (cfg->strategy) {
-            case RoutingStrategy::HASH:
+            case RoutingStrategy::HASH: {
+                int64_t as_int = 0;
+                if (parse_full_int(val, val_len, as_int))
+                    return shard_index_for_int(table_name, as_int);
                 return fnv1a_bytes(reinterpret_cast<const uint8_t*>(val), val_len) % n;
+            }
             case RoutingStrategy::RANGE:
                 // RANGE is integer-keyed only. Fall back to scatter-friendly
                 // shard 0 rather than producing a misleading single-shard
@@ -357,6 +366,18 @@ class ShardMap {
         return idx < n ? idx : (n == 0 ? 0 : n - 1);
     }
 
+    static bool parse_full_int(const char* val, uint32_t val_len, int64_t& out) {
+        if (!val || val_len == 0 || val_len > 20) return false;
+        char buf[24];
+        std::memcpy(buf, val, val_len);
+        buf[val_len] = '\0';
+        char* end = nullptr;
+        long long n = std::strtoll(buf, &end, 10);
+        if (!end || end != buf + val_len) return false;
+        out = static_cast<int64_t>(n);
+        return true;
+    }
+
     static void split_keys(const std::string& spec, std::vector<std::string>& out) {
         size_t start = 0;
         while (start <= spec.size()) {

tests/test_distributed_dml.cpp

@@ -730,6 +730,36 @@ TEST_F(DistributedDmlTest, UpdateShardKeyMovesRow) {
               "Carol");
 }
 
+TEST_F(DistributedDmlTest, UpdateQualifiedShardKeyMovesRow) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 17)");
+    const char* src = backend_for_id(3);
+    const char* dst = backend_for_id(9);
+    ASSERT_STRNE(src, dst);
+
+    auto result = execute_distributed_dml("UPDATE users SET users.id = 9 WHERE id = 3");
+    EXPECT_TRUE(result.success) << result.error_message;
+    EXPECT_EQ(row_count_on(src, "users"), 0u);
+    EXPECT_EQ(row_count_on(dst, "users"), 1u);
+}
+
+TEST_F(DistributedDmlTest, UpdateShardKeyNoMatchingRowIsNoop) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 17)");
+    const char* home = backend_for_id(3);
+    auto result = execute_distributed_dml("UPDATE users SET id = 9 WHERE id = 99");
+    EXPECT_TRUE(result.success) << result.error_message;
+    EXPECT_EQ(row_count_on(home, "users"), 1u);
+    EXPECT_EQ(mock_executor.total_row_count("users"), 1u);
+}
+
+TEST_F(DistributedDmlTest, InsertStringIntHashesLikeInt) {
+    auto ins = execute_distributed_dml(
+        "INSERT INTO users (id, name, age) VALUES ('3', 'Carol', 17)");
+    EXPECT_TRUE(ins.success) << ins.error_message;
+    EXPECT_EQ(row_count_on(backend_for_id(3), "users"), 1u);
+    auto got = execute_distributed_select("SELECT name FROM users WHERE id = 3");
+    ASSERT_EQ(got.row_count(), 1u);
+}
+
 TEST_F(DistributedDmlTest, UpdateShardKeySameShard) {
     int64_t a = 3;
     int64_t b = a;

tests/test_distributed_planner.cpp

@@ -1200,6 +1200,51 @@ TEST_F(DistributedPlannerTest, ColocatedJoinPushedToShards) {
     }
 }
 
+TEST_F(DistributedPlannerTest, ColocatedLeftJoinEmitsLeftJoin) {
+    shard_map.add_table(TableShardConfig{
+        "orders", "user_id",
+        {ShardInfo{"shard_1"}, ShardInfo{"shard_2"}, ShardInfo{"shard_3"}}
+    });
+
+    Parser<Dialect::MySQL> parser;
+    const char* sql = "SELECT * FROM users LEFT JOIN orders ON users.id = orders.user_id";
+    auto pr = parser.parse(sql, std::strlen(sql));
+    ASSERT_EQ(pr.status, ParseResult::OK);
+    PlanBuilder<Dialect::MySQL> builder(catalog, parser.arena());
+    PlanNode* plan = builder.build(pr.ast);
+    DistributedPlanner<Dialect::MySQL> dp(shard_map, catalog, parser.arena());
+    PlanNode* dist = dp.distribute(plan);
+    ASSERT_NE(dist, nullptr);
+
+    std::vector<PlanNode*> remotes;
+    find_nodes(dist, PlanNodeType::REMOTE_SCAN, remotes);
+    ASSERT_FALSE(remotes.empty());
+    for (auto* rs : remotes) {
+        std::string remote(rs->remote_scan.remote_sql, rs->remote_scan.remote_sql_len);
+        EXPECT_NE(remote.find("LEFT JOIN"), std::string::npos) << remote;
+    }
+}
+
+TEST_F(DistributedPlannerTest, SemiJoinSkipsLeftJoin) {
+    Parser<Dialect::MySQL> parser;
+    const char* sql = "SELECT * FROM users LEFT JOIN orders ON users.id = orders.user_id";
+    auto pr = parser.parse(sql, std::strlen(sql));
+    ASSERT_EQ(pr.status, ParseResult::OK);
+    PlanBuilder<Dialect::MySQL> builder(catalog, parser.arena());
+    PlanNode* plan = builder.build(pr.ast);
+    DistributedPlanner<Dialect::MySQL> dp(shard_map, catalog, parser.arena(),
+                                          &mock_executor, &functions);
+    PlanNode* dist = dp.distribute(plan);
+    ASSERT_NE(dist, nullptr);
+    std::vector<PlanNode*> remotes;
+    find_nodes(dist, PlanNodeType::REMOTE_SCAN, remotes);
+    for (auto* rs : remotes) {
+        std::string remote(rs->remote_scan.remote_sql, rs->remote_scan.remote_sql_len);
+        if (remote.find("users") != std::string::npos)
+            EXPECT_EQ(remote.find(" IN "), std::string::npos) << remote;
+    }
+}
+
 TEST_F(DistributedPlannerTest, CompositeColocatedJoinPushedToShards) {
     catalog.add_table("", "kv", {
         {"tenant_id", SqlType::make_int(), false},

tests/test_shard_map.cpp

@@ -10,6 +10,7 @@
 
 #include <set>
 #include <climits>
+#include <string>
 
 using namespace sql_engine;
 using sql_parser::StringRef;
@@ -47,6 +48,19 @@ TEST(ShardMapHashTest, IsDeterministic) {
     }
 }
 
+TEST(ShardMapHashTest, StringIntegerRoutesLikeInt) {
+    ShardMap map;
+    map.add_table(make_two_shards(RoutingStrategy::HASH));
+    for (int i = -20; i < 20; ++i) {
+        std::string s = std::to_string(i);
+        size_t a = map.shard_index_for_int(sref("users"), i);
+        size_t b = 99;
+        ASSERT_TRUE(map.try_shard_index_for_string(
+            sref("users"), s.c_str(), static_cast<uint32_t>(s.size()), b));
+        EXPECT_EQ(a, b) << "for " << i;
+    }
+}
+
 TEST(ShardMapHashTest, DistributesAcrossBothShards) {
     // FNV-1a with modulo 2 should hit both shards for the small range
     // 0..63 — the test is loose because we don't want to over-specify