test: broaden coverage for the #56 P0 fixes

Add cases for qualified non-key UPDATE, same-value and multi-row moves,
string/int HASH including negatives and leading zeros, empty IN-subquery
DML, string IN-list delete, INSERT...SELECT empty source, unknown-table
SELECT/UPDATE/DELETE, plan-cache COUNT/SUM twice, composite qualified
move, RIGHT JOIN emit, LIST OR prune, and partial vs full composite HASH
prune.

Author: renecannao

tests/test_distributed_dml.cpp

@@ -986,3 +986,178 @@ TEST_F(DistributedDmlTest, PlanCacheSeesUpdatedShardMap) {
             EXPECT_EQ(mock_executor.get_executed_sqls(s).size(), 0u) << s;
     }
 }
+
+TEST_F(DistributedDmlTest, UpdateQualifiedNonKeyDoesNotMove) {
+    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 users.age = 40 WHERE id = 3");
+    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, UpdateShardKeyToSameValue) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 17)");
+    auto result = execute_distributed_dml("UPDATE users SET id = 3 WHERE id = 3");
+    EXPECT_TRUE(result.success) << result.error_message;
+    EXPECT_EQ(row_count_on(backend_for_id(3), "users"), 1u);
+    EXPECT_EQ(mock_executor.total_row_count("users"), 1u);
+}
+
+TEST_F(DistributedDmlTest, UpdateMovesThenPointSelectAndDelete) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 17)");
+    ASSERT_STRNE(backend_for_id(3), backend_for_id(9));
+    EXPECT_TRUE(execute_distributed_dml("UPDATE users SET id = 9 WHERE id = 3").success);
+    EXPECT_EQ(execute_distributed_select("SELECT name FROM users WHERE id = 3").row_count(), 0u);
+    EXPECT_EQ(execute_distributed_select("SELECT name FROM users WHERE id = 9").row_count(), 1u);
+    EXPECT_TRUE(execute_distributed_dml("DELETE FROM users WHERE id = 9").success);
+    EXPECT_EQ(mock_executor.total_row_count("users"), 0u);
+}
+
+TEST_F(DistributedDmlTest, UpdateMovesMultipleRowsToOneShard) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'A', 1)");
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (4, 'B', 2)");
+    EXPECT_TRUE(execute_distributed_dml("UPDATE users SET id = 9 WHERE id IN (3, 4)").success);
+    EXPECT_EQ(row_count_on(backend_for_id(9), "users"), 2u);
+    EXPECT_EQ(mock_executor.total_row_count("users"), 2u);
+}
+
+TEST_F(DistributedDmlTest, InsertNegativeStringInt) {
+    auto ins = execute_distributed_dml(
+        "INSERT INTO users (id, name, age) VALUES ('-7', 'Neg', 1)");
+    EXPECT_TRUE(ins.success) << ins.error_message;
+    EXPECT_EQ(row_count_on(backend_for_id(-7), "users"), 1u);
+    EXPECT_EQ(execute_distributed_select("SELECT name FROM users WHERE id = -7").row_count(), 1u);
+}
+
+TEST_F(DistributedDmlTest, InsertLeadingZeroStringInt) {
+    auto ins = execute_distributed_dml(
+        "INSERT INTO users (id, name, age) VALUES ('03', 'Zed', 1)");
+    EXPECT_TRUE(ins.success) << ins.error_message;
+    EXPECT_EQ(row_count_on(backend_for_id(3), "users"), 1u);
+    EXPECT_EQ(execute_distributed_select("SELECT name FROM users WHERE id = 3").row_count(), 1u);
+}
+
+TEST_F(DistributedDmlTest, EmptyInSubqueryDeletesNothing) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 17)");
+    mock_executor.clear_sql_logs();
+    auto result = execute_distributed_dml(
+        "DELETE FROM users WHERE id IN (SELECT user_id FROM orders)");
+    EXPECT_TRUE(result.success) << result.error_message;
+    EXPECT_EQ(mock_executor.total_row_count("users"), 1u);
+}
+
+TEST_F(DistributedDmlTest, EmptyInSubqueryUpdateTouchesNothing) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 17)");
+    auto result = execute_distributed_dml(
+        "UPDATE users SET age = 99 WHERE id IN (SELECT user_id FROM orders)");
+    EXPECT_TRUE(result.success) << result.error_message;
+    auto got = execute_distributed_select("SELECT age FROM users WHERE id = 3");
+    ASSERT_EQ(got.row_count(), 1u);
+    EXPECT_EQ(got.rows[0].get(0).int_val, 17);
+}
+
+TEST_F(DistributedDmlTest, InSubqueryStringNames) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 17)");
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (4, 'Dave', 18)");
+    auto result = execute_distributed_dml(
+        "DELETE FROM users WHERE name IN (SELECT name FROM users WHERE id = 3)");
+    EXPECT_TRUE(result.success) << result.error_message;
+    EXPECT_EQ(mock_executor.total_row_count("users"), 1u);
+    EXPECT_EQ(execute_distributed_select("SELECT name FROM users WHERE id = 4").row_count(), 1u);
+}
+
+TEST_F(DistributedDmlTest, InsertSelectEmptySourceIsNoop) {
+    auto result = execute_distributed_dml(
+        "INSERT INTO users (id, name, age) SELECT order_id, 'x', 1 FROM orders WHERE order_id = 999");
+    EXPECT_TRUE(result.success) << result.error_message;
+    EXPECT_EQ(mock_executor.total_row_count("users"), 0u);
+}
+
+TEST_F(DistributedDmlTest, UnknownTableSelectIsEmpty) {
+    catalog.add_table("", "ghost", {{"id", SqlType::make_int(), false}});
+    Parser<Dialect::MySQL> parser;
+    auto pr = parser.parse("SELECT * FROM ghost", 19);
+    ASSERT_EQ(pr.status, ParseResult::OK);
+    PlanBuilder<Dialect::MySQL> builder(catalog, parser.arena());
+    PlanNode* plan = builder.build(pr.ast);
+    DistributedPlanner<Dialect::MySQL> dist(shard_map, catalog, parser.arena(),
+                                             &mock_executor, &functions);
+    EXPECT_EQ(dist.distribute(plan), nullptr);
+    ASSERT_NE(dist.last_error(), nullptr);
+}
+
+TEST_F(DistributedDmlTest, UpdateUnknownTableErrors) {
+    catalog.add_table("", "ghost", {{"id", SqlType::make_int(), false}});
+    auto result = execute_distributed_dml("UPDATE ghost SET id = 1");
+    EXPECT_FALSE(result.success);
+    EXPECT_NE(result.error_message.find("shard map"), std::string::npos);
+}
+
+TEST_F(DistributedDmlTest, DeleteUnknownTableErrors) {
+    catalog.add_table("", "ghost", {{"id", SqlType::make_int(), false}});
+    auto result = execute_distributed_dml("DELETE FROM ghost");
+    EXPECT_FALSE(result.success);
+    EXPECT_NE(result.error_message.find("shard map"), std::string::npos);
+}
+
+TEST_F(DistributedDmlTest, PlanCacheCountTwice) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 17)");
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (4, 'Dave', 18)");
+    LocalTransactionManager txn(data_arena);
+    Session<Dialect::MySQL> session(catalog, txn);
+    session.set_remote_executor(&mock_executor);
+    session.set_shard_map(&shard_map);
+    const char* sql = "SELECT COUNT(*) FROM users";
+    auto a = session.execute_query(sql);
+    auto b = session.execute_query(sql);
+    ASSERT_EQ(a.row_count(), 1u);
+    ASSERT_EQ(b.row_count(), 1u);
+    EXPECT_EQ(a.rows[0].get(0).tag, b.rows[0].get(0).tag);
+    EXPECT_EQ(a.rows[0].get(0).to_int64(), 2);
+    EXPECT_EQ(b.rows[0].get(0).to_int64(), 2);
+    EXPECT_EQ(session.plan_cache_size(), 1u);
+}
+
+TEST_F(DistributedDmlTest, PlanCacheSumAndGroupByTwice) {
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (3, 'Carol', 17)");
+    execute_distributed_dml("INSERT INTO users (id, name, age) VALUES (4, 'Dave', 18)");
+    LocalTransactionManager txn(data_arena);
+    Session<Dialect::MySQL> session(catalog, txn);
+    session.set_remote_executor(&mock_executor);
+    session.set_shard_map(&shard_map);
+    const char* sql = "SELECT SUM(age) FROM users";
+    auto a = session.execute_query(sql);
+    auto b = session.execute_query(sql);
+    ASSERT_EQ(a.row_count(), 1u);
+    ASSERT_EQ(b.row_count(), 1u);
+    EXPECT_EQ(a.rows[0].get(0).to_int64(), 35);
+    EXPECT_EQ(b.rows[0].get(0).to_int64(), 35);
+}
+
+TEST_F(DistributedDmlTest, CompositeQualifiedShardKeyMove) {
+    catalog.add_table("", "kv", {
+        {"tenant_id", SqlType::make_int(), false},
+        {"id",        SqlType::make_int(), false},
+        {"name",      SqlType::make_varchar(255), true},
+    });
+    TableShardConfig cfg;
+    cfg.table_name = "kv";
+    cfg.shard_key = "tenant_id+id";
+    cfg.shards = {{"shard0"}, {"shard1"}, {"shard2"}};
+    shard_map.add_table(cfg);
+    mock_executor.add_table_to_all("kv", {
+        {"tenant_id", SqlType::make_int(), false},
+        {"id",        SqlType::make_int(), false},
+        {"name",      SqlType::make_varchar(255), true},
+    });
+    EXPECT_TRUE(execute_distributed_dml(
+        "INSERT INTO kv (tenant_id, id, name) VALUES (1, 3, 'A')").success);
+    auto result = execute_distributed_dml(
+        "UPDATE kv SET kv.id = 9 WHERE tenant_id = 1 AND id = 3");
+    EXPECT_TRUE(result.success) << result.error_message;
+    EXPECT_EQ(execute_distributed_select(
+        "SELECT name FROM kv WHERE tenant_id = 1 AND id = 9").row_count(), 1u);
+    EXPECT_EQ(execute_distributed_select(
+        "SELECT name FROM kv WHERE tenant_id = 1 AND id = 3").row_count(), 0u);
+}

tests/test_distributed_planner.cpp

@@ -1358,6 +1358,86 @@ TEST_F(DistributedPlannerTest, CompositeRangePrunesOnFirstKey) {
     EXPECT_EQ(count_remotes("SELECT * FROM users WHERE id BETWEEN 6 AND 10"), 1u);
 }
 
+TEST_F(DistributedPlannerTest, ColocatedRightJoinEmitsRightJoin) {
+    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 RIGHT 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);
+    std::vector<PlanNode*> remotes;
+    find_nodes(dist, PlanNodeType::REMOTE_SCAN, remotes);
+    ASSERT_FALSE(remotes.empty());
+    bool saw_right = false;
+    for (auto* rs : remotes) {
+        std::string remote(rs->remote_scan.remote_sql, rs->remote_scan.remote_sql_len);
+        if (remote.find("RIGHT JOIN") != std::string::npos) saw_right = true;
+    }
+    EXPECT_TRUE(saw_right);
+}
+
+TEST_F(DistributedPlannerTest, OrEqualityPrunesUnionOfShards) {
+    TableShardConfig cfg;
+    cfg.table_name = "users";
+    cfg.shard_key = "id";
+    cfg.shards = {{"shard_1"}, {"shard_2"}, {"shard_3"}};
+    cfg.strategy = RoutingStrategy::LIST;
+    cfg.list = {{true, 1, "", 0}, {true, 6, "", 1}, {true, 15, "", 2}};
+    shard_map.add_table(cfg);
+
+    Parser<Dialect::MySQL> parser;
+    auto pr = parser.parse("SELECT * FROM users WHERE id = 1 OR id = 6", 42);
+    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);
+    std::vector<PlanNode*> remotes;
+    find_nodes(dist, PlanNodeType::REMOTE_SCAN, remotes);
+    EXPECT_EQ(remotes.size(), 2u);
+}
+
+TEST_F(DistributedPlannerTest, CompositeHashPartialWhereScatters) {
+    TableShardConfig cfg;
+    cfg.table_name = "users";
+    cfg.shard_key = "id+name";
+    cfg.shards = {{"shard_1"}, {"shard_2"}, {"shard_3"}};
+    shard_map.add_table(cfg);
+
+    Parser<Dialect::MySQL> parser;
+    auto pr = parser.parse("SELECT * FROM users WHERE id = 3", 32);
+    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);
+    std::vector<PlanNode*> remotes;
+    find_nodes(dist, PlanNodeType::REMOTE_SCAN, remotes);
+    EXPECT_EQ(remotes.size(), 3u);
+}
+
+TEST_F(DistributedPlannerTest, CompositeHashBothKeysPrune) {
+    TableShardConfig cfg;
+    cfg.table_name = "users";
+    cfg.shard_key = "id+age";
+    cfg.shards = {{"shard_1"}, {"shard_2"}, {"shard_3"}};
+    shard_map.add_table(cfg);
+
+    Parser<Dialect::MySQL> parser;
+    auto pr = parser.parse("SELECT * FROM users WHERE id = 3 AND age = 17", 45);
+    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);
+    std::vector<PlanNode*> remotes;
+    find_nodes(dist, PlanNodeType::REMOTE_SCAN, remotes);
+    EXPECT_EQ(remotes.size(), 1u);
+}
+
 TEST_F(DistributedPlannerTest, SemiJoinPrunesProbeShards) {
     Parser<Dialect::MySQL> parser;
     const char* sql = "SELECT * FROM users JOIN orders ON users.id = orders.user_id";

tests/test_shard_map.cpp

@@ -48,6 +48,62 @@ TEST(ShardMapHashTest, IsDeterministic) {
     }
 }
 
+TEST(ShardMapHashTest, LeadingZerosRouteLikeInt) {
+    ShardMap map;
+    map.add_table(make_two_shards(RoutingStrategy::HASH));
+    size_t a = map.shard_index_for_int(sref("users"), 3);
+    size_t b = 99;
+    ASSERT_TRUE(map.try_shard_index_for_string(sref("users"), "03", 2, b));
+    EXPECT_EQ(a, b);
+}
+
+TEST(ShardMapHashTest, NonNumericStringDoesNotCoerce) {
+    TableShardConfig cfg;
+    cfg.table_name = "users";
+    cfg.shard_key = "id";
+    cfg.shards = {ShardInfo{"a"}, ShardInfo{"b"}, ShardInfo{"c"}, ShardInfo{"d"}};
+    ShardMap map;
+    map.add_table(cfg);
+    size_t as_int = map.shard_index_for_int(sref("users"), 3);
+    bool differs = false;
+    const char* samples[] = {"3x", "3.0", " 3", "3 "};
+    for (const char* s : samples) {
+        size_t idx = 99;
+        ASSERT_TRUE(map.try_shard_index_for_string(
+            sref("users"), s, static_cast<uint32_t>(std::strlen(s)), idx));
+        if (idx != as_int) differs = true;
+    }
+    EXPECT_TRUE(differs);
+}
+
+TEST(ShardMapHashTest, EmptyStringIsUnparsed) {
+    ShardMap map;
+    map.add_table(make_two_shards(RoutingStrategy::HASH));
+    size_t idx = 99;
+    EXPECT_TRUE(map.try_shard_index_for_string(sref("users"), "", 0, idx));
+    EXPECT_LT(idx, 2u);
+}
+
+TEST(ShardMapListTest, StringIntegerDoesNotCoerceToIntKey) {
+    TableShardConfig cfg = make_two_shards(RoutingStrategy::LIST);
+    cfg.list = {ShardListEntry{true, 3, "", 1}};
+    ShardMap map;
+    map.add_table(cfg);
+    size_t idx = 99;
+    EXPECT_FALSE(map.try_shard_index_for_string(sref("users"), "3", 1, idx));
+    EXPECT_TRUE(map.try_shard_index_for_int(sref("users"), 3, idx));
+    EXPECT_EQ(idx, 1u);
+}
+
+TEST(ShardMapRangeTest, StringIntegerIsUnroutable) {
+    TableShardConfig cfg = make_two_shards(RoutingStrategy::RANGE);
+    cfg.ranges = {ShardRange{5, 0}, ShardRange{100, 1}};
+    ShardMap map;
+    map.add_table(cfg);
+    size_t idx = 99;
+    EXPECT_FALSE(map.try_shard_index_for_string(sref("users"), "3", 1, idx));
+}
+
 TEST(ShardMapHashTest, StringIntegerRoutesLikeInt) {
     ShardMap map;
     map.add_table(make_two_shards(RoutingStrategy::HASH));