diff --git a/include/sql_engine/distributed_planner.h b/include/sql_engine/distributed_planner.h index 1a8491f..d224065 100644 --- a/include/sql_engine/distributed_planner.h +++ b/include/sql_engine/distributed_planner.h @@ -357,7 +357,13 @@ class DistributedPlanner { if (keys.empty()) return all_shards; std::vector target_indices; - if (keys.size() > 1) { + if (keys.size() > 1 && + shards_.routing_strategy(table->table_name) == RoutingStrategy::RANGE) { + sql_parser::StringRef first{keys[0].c_str(), + static_cast(keys[0].size())}; + extract_shard_targets(where_expr, first, table->table_name, + all_shards.size(), target_indices); + } else if (keys.size() > 1) { extract_composite_targets(where_expr, keys, table->table_name, target_indices); } else { sql_parser::StringRef shard_key{keys[0].c_str(), @@ -1203,6 +1209,157 @@ class DistributedPlanner { return current ? current : join_node; } + bool column_on_table(const sql_parser::AstNode* node, const TableInfo* table, + const TableInfo* other) const { + if (!node || !table) return false; + if (node->type == sql_parser::NodeType::NODE_QUALIFIED_NAME) { + const sql_parser::AstNode* t = node->first_child; + if (!t) return false; + sql_parser::StringRef tn = t->value(); + if (table->table_name.equals_ci(tn.ptr, tn.len)) return true; + if (table->alias.ptr && table->alias.equals_ci(tn.ptr, tn.len)) return true; + return false; + } + if (node->type == sql_parser::NodeType::NODE_COLUMN_REF || + node->type == sql_parser::NodeType::NODE_IDENTIFIER) { + if (!catalog_.get_column(table, node->value())) return false; + if (other && catalog_.get_column(other, node->value())) return false; + return true; + } + return false; + } + + const sql_parser::AstNode* probe_key_in_join(const sql_parser::AstNode* cond, + const TableInfo* probe, + const TableInfo* build) const { + if (!cond || !probe || !build) return nullptr; + const auto& keys = shards_.get_shard_keys(probe->table_name); + if (keys.size() != 1) return nullptr; + sql_parser::StringRef sk{keys[0].c_str(), static_cast(keys[0].size())}; + std::vector> eqs; + collect_eq_pairs(cond, eqs); + for (const auto& eq : eqs) { + if (is_shard_key_ref(eq.first, sk) && column_on_table(eq.second, build, probe)) + return eq.first; + if (is_shard_key_ref(eq.second, sk) && column_on_table(eq.first, build, probe)) + return eq.second; + } + return nullptr; + } + + sql_parser::AstNode* make_in_list_on_column(const sql_parser::AstNode* col, + const std::vector& values) { + if (!col || values.empty()) return nullptr; + sql_parser::AstNode* stub = sql_parser::make_node( + arena_, sql_parser::NodeType::NODE_IN_LIST, + sql_parser::StringRef{nullptr, 0}); + sql_parser::AstNode* col_copy = sql_parser::make_node( + arena_, col->type, col->value(), col->flags); + col_copy->first_child = col->first_child; + stub->add_child(col_copy); + return build_in_list_from_values(stub, values); + } + + sql_parser::AstNode* and_preds(const sql_parser::AstNode* a, + const sql_parser::AstNode* b) { + if (!a) return const_cast(b); + if (!b) return const_cast(a); + sql_parser::AstNode* n = sql_parser::make_node( + arena_, sql_parser::NodeType::NODE_BINARY_OP, + sql_parser::StringRef{"AND", 3}); + n->add_child(const_cast(a)); + n->add_child(const_cast(b)); + return n; + } + + std::vector collect_build_join_keys(const TableInfo* build, + const sql_parser::AstNode* where_expr, + const sql_parser::AstNode* join_eq_other) { + std::vector out; + if (!build || !join_eq_other || !remote_executor_) return out; + const sql_parser::AstNode* proj[1] = {join_eq_other}; + const auto& shards = shards_.get_shards(build->table_name); + if (shards.empty()) return out; + std::vector targets = shards; + if (shards_.is_sharded(build->table_name) && shards.size() > 1) + return out; + sql_parser::StringRef sql = qb_.build_select( + build, where_expr, proj, 1, nullptr, 0, + nullptr, nullptr, 0, -1, true); + ResultSet rs = remote_executor_->execute(shards[0].backend_name.c_str(), sql); + for (const auto& row : rs.rows) { + if (row.column_count > 0 && value_is_routable(row.get(0))) + out.push_back(copy_value_arena(row.get(0))); + } + return out; + } + + const sql_parser::AstNode* other_eq_side(const sql_parser::AstNode* cond, + const sql_parser::AstNode* probe_key) const { + std::vector> eqs; + collect_eq_pairs(cond, eqs); + for (const auto& eq : eqs) { + if (eq.first == probe_key) return eq.second; + if (eq.second == probe_key) return eq.first; + } + return nullptr; + } + + PlanNode* try_semijoin_prune(PlanNode* join_node, + const TableInfo* left_table, + const TableInfo* right_table) { + if (!join_node || !remote_executor_ || !join_node->join.condition) + return nullptr; + if (!left_table || !right_table) return nullptr; + + bool ls = shards_.is_sharded(left_table->table_name); + bool rs = shards_.is_sharded(right_table->table_name); + if (ls == rs) return nullptr; + + const TableInfo* probe = ls ? left_table : right_table; + const TableInfo* build = ls ? right_table : left_table; + bool probe_is_left = ls; + const sql_parser::AstNode* probe_key = + probe_key_in_join(join_node->join.condition, probe, build); + if (!probe_key) return nullptr; + const sql_parser::AstNode* build_col = + other_eq_side(join_node->join.condition, probe_key); + if (!build_col) return nullptr; + + ScanContext bctx = extract_scan_context( + probe_is_left ? join_node->right : join_node->left); + std::vector keys = collect_build_join_keys(build, bctx.where_expr, build_col); + if (keys.empty()) return nullptr; + + sql_parser::AstNode* in_list = make_in_list_on_column(probe_key, keys); + if (!in_list) return nullptr; + + ScanContext pctx = extract_scan_context( + probe_is_left ? join_node->left : join_node->right); + if (!pctx.scan) return nullptr; + const sql_parser::AstNode* probe_where = and_preds(pctx.where_expr, in_list); + PlanNode* probe_dist = distribute_scan(pctx.scan, probe_where, + nullptr, nullptr, nullptr, false); + + PlanNode* build_dist = nullptr; + if (bctx.scan && !shards_.is_sharded(build->table_name)) { + sql_parser::StringRef sql = qb_.build_select( + build, bctx.where_expr, nullptr, 0, nullptr, 0, + nullptr, nullptr, 0, -1, false); + build_dist = make_remote_scan( + shards_.get_backend(build->table_name), sql, build); + } else { + build_dist = distribute_node(probe_is_left ? join_node->right : join_node->left); + } + if (!probe_dist || !build_dist) return nullptr; + + PlanNode* result = make_plan_node(arena_, PlanNodeType::JOIN); + result->join = join_node->join; + result->left = probe_is_left ? probe_dist : build_dist; + result->right = probe_is_left ? build_dist : probe_dist; + return result; + } + PlanNode* distribute_join(PlanNode* join_node) { const TableInfo* left_table = find_table(join_node->left); const TableInfo* right_table = find_table(join_node->right); @@ -1217,6 +1374,9 @@ class DistributedPlanner { return distribute_colocated_join(join_node, left_table, right_table); } + if (PlanNode* sj = try_semijoin_prune(join_node, left_table, right_table)) + return sj; + PlanNode* left_dist = nullptr; PlanNode* right_dist = nullptr; diff --git a/include/sql_engine/shard_map.h b/include/sql_engine/shard_map.h index 907a651..894a5cd 100644 --- a/include/sql_engine/shard_map.h +++ b/include/sql_engine/shard_map.h @@ -177,7 +177,7 @@ class ShardMap { size_t& out) const { const TableShardConfig* cfg = lookup(table_name); if (!cfg || cfg->shards.empty() || !parts || n == 0) return false; - if (n == 1) { + if (n == 1 || cfg->strategy == RoutingStrategy::RANGE) { return parts[0].is_int ? try_shard_index_for_int(table_name, parts[0].int_val, out) : try_shard_index_for_string(table_name, parts[0].str, diff --git a/scripts/run_pg_sharding_demo.sh b/scripts/run_pg_sharding_demo.sh new file mode 100755 index 0000000..bbe2550 --- /dev/null +++ b/scripts/run_pg_sharding_demo.sh @@ -0,0 +1,81 @@ +#!/bin/bash +# RANGE + LIST + 2PC against the two PostgreSQL shards. +set -e + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +PROJECT_DIR="$(dirname "$SCRIPT_DIR")" +cd "$PROJECT_DIR" + +if ! docker exec parsersql-pg-shard1 pg_isready -Upostgres &>/dev/null 2>&1; then + echo "ERROR: PG shards not running. Start them with: ./scripts/start_pg_sharding_demo.sh" + exit 1 +fi + +if [ ! -f ./sqlengine ]; then + echo "Building sqlengine..." + make build-sqlengine +fi + +PG1='pgsql://postgres:test@127.0.0.1:16432/testdb?name=pg1' +PG2='pgsql://postgres:test@127.0.0.1:16433/testdb?name=pg2' +TXN_LOG="${TMPDIR:-/tmp}/parsersql-pg-demo.txn" + +run_sql() { + local desc="$1" + local sql="$2" + echo "----------------------------------------------" + echo "QUERY: $desc" + echo "SQL: $sql" + echo "" + echo "$sql" | ./sqlengine \ + --backend "$PG1" \ + --backend "$PG2" \ + --shard "users:id:range:5=pg1,10=pg2" \ + --shard "regions:name:list:us-east=pg1,us-west=pg2" \ + --shard "orders:id:range:105=pg1,110=pg2" \ + --txn-log "$TXN_LOG" \ + 2>&1 + echo "" +} + +echo "==============================================" +echo " PostgreSQL LIST + RANGE + 2PC demo" +echo "==============================================" +echo " pg1 :16432 users 1-5 / us-east" +echo " pg2 :16433 users 6-10 / us-west" +echo "" + +run_sql "RANGE point lookup" \ + "SELECT name FROM users WHERE id = 3" + +run_sql "RANGE BETWEEN prune" \ + "SELECT name FROM users WHERE id BETWEEN 6 AND 10" + +run_sql "LIST point lookup" \ + "SELECT tz FROM regions WHERE name = 'us-west'" + +run_sql "Scatter scan" \ + "SELECT COUNT(*) FROM users" + +echo "==============================================" +echo " 2PC write across both shards (one engine)" +echo "==============================================" +{ + echo "BEGIN" + echo "INSERT INTO users (id, name, age) VALUES (0, 'Zero', 1)" + echo "INSERT INTO users (id, name, age) VALUES (11, 'Eleven', 2)" + echo "COMMIT" +} | ./sqlengine \ + --backend "$PG1" \ + --backend "$PG2" \ + --shard "users:id:range:5=pg1,10=pg2" \ + --shard "regions:name:list:us-east=pg1,us-west=pg2" \ + --shard "orders:id:range:105=pg1,110=pg2" \ + --txn-log "$TXN_LOG" \ + 2>&1 +echo "" + +run_sql "Read back 2PC inserts" \ + "SELECT id, name FROM users WHERE id IN (0, 11)" + +echo "Demo complete. Stop: docker rm -f parsersql-pg-shard1 parsersql-pg-shard2" diff --git a/scripts/start_pg_sharding_demo.sh b/scripts/start_pg_sharding_demo.sh new file mode 100755 index 0000000..f1a1578 --- /dev/null +++ b/scripts/start_pg_sharding_demo.sh @@ -0,0 +1,98 @@ +#!/bin/bash +# Two PostgreSQL shards for LIST + RANGE + 2PC. Ports 16432/16433 +# (15432 is the unit-test backend; 13306 is the MySQL sharding demo). +set -e + +echo "=== Starting 2-shard PostgreSQL demo ===" + +docker rm -f parsersql-pg-shard1 parsersql-pg-shard2 2>/dev/null || true + +docker run -d --name parsersql-pg-shard1 \ + -p 16432:5432 \ + -e POSTGRES_PASSWORD=test \ + -e POSTGRES_DB=testdb \ + postgres:16 \ + -c max_prepared_transactions=16 + +docker run -d --name parsersql-pg-shard2 \ + -p 16433:5432 \ + -e POSTGRES_PASSWORD=test \ + -e POSTGRES_DB=testdb \ + postgres:16 \ + -c max_prepared_transactions=16 + +echo "Waiting for PG shard 1..." +until docker exec parsersql-pg-shard1 pg_isready -Upostgres &>/dev/null 2>&1; do sleep 1; done +echo "PG shard 1 ready" + +echo "Waiting for PG shard 2..." +until docker exec parsersql-pg-shard2 pg_isready -Upostgres &>/dev/null 2>&1; do sleep 1; done +echo "PG shard 2 ready" + +echo "Loading RANGE users 1-5 + LIST region us-east on shard 1..." +docker exec -i parsersql-pg-shard1 psql -Upostgres testdb <<'SQL' +DROP TABLE IF EXISTS orders; +DROP TABLE IF EXISTS users; +DROP TABLE IF EXISTS regions; + +CREATE TABLE users ( + id INT PRIMARY KEY, + name VARCHAR(255) NOT NULL, + age INT +); +CREATE TABLE regions ( + name VARCHAR(64) PRIMARY KEY, + tz VARCHAR(32) +); +CREATE TABLE orders ( + id INT PRIMARY KEY, + user_id INT, + total NUMERIC(10,2) +); + +INSERT INTO users VALUES + (1, 'Alice', 30), + (2, 'Bob', 25), + (3, 'Carol', 35), + (4, 'Dave', 28), + (5, 'Eve', 32); +INSERT INTO regions VALUES ('us-east', 'EST'); +INSERT INTO orders VALUES (101, 1, 150.00), (102, 3, 50.00); +SQL + +echo "Loading RANGE users 6-10 + LIST region us-west on shard 2..." +docker exec -i parsersql-pg-shard2 psql -Upostgres testdb <<'SQL' +DROP TABLE IF EXISTS orders; +DROP TABLE IF EXISTS users; +DROP TABLE IF EXISTS regions; + +CREATE TABLE users ( + id INT PRIMARY KEY, + name VARCHAR(255) NOT NULL, + age INT +); +CREATE TABLE regions ( + name VARCHAR(64) PRIMARY KEY, + tz VARCHAR(32) +); +CREATE TABLE orders ( + id INT PRIMARY KEY, + user_id INT, + total NUMERIC(10,2) +); + +INSERT INTO users VALUES + (6, 'Frank', 40), + (7, 'Grace', 22), + (8, 'Hank', 31), + (9, 'Ivy', 27), + (10, 'Jack', 36); +INSERT INTO regions VALUES ('us-west', 'PST'); +INSERT INTO orders VALUES (106, 6, 80.00), (107, 8, 120.00); +SQL + +echo "PostgreSQL shards ready:" +echo " pg1 127.0.0.1:16432 users 1-5, region us-east" +echo " pg2 127.0.0.1:16433 users 6-10, region us-west" +echo "Run: ./scripts/run_pg_sharding_demo.sh" +echo "Stop: docker rm -f parsersql-pg-shard1 parsersql-pg-shard2" diff --git a/src/sql_engine/tool_config_parser.cpp b/src/sql_engine/tool_config_parser.cpp index e9c4738..8cf06b8 100644 --- a/src/sql_engine/tool_config_parser.cpp +++ b/src/sql_engine/tool_config_parser.cpp @@ -208,10 +208,6 @@ ParsedShard parse_shard_spec(const std::string& spec) { return ps; } } else if (strategy_token == "range") { - if (ps.config.shard_key.find('+') != std::string::npos) { - ps.error = "composite shard keys require HASH strategy: " + spec; - return ps; - } ps.config.strategy = RoutingStrategy::RANGE; for (auto& entry : split_csv(body)) { std::string upper_str, backend; diff --git a/tests/test_distributed_planner.cpp b/tests/test_distributed_planner.cpp index ca52c72..f5f1c88 100644 --- a/tests/test_distributed_planner.cpp +++ b/tests/test_distributed_planner.cpp @@ -1288,6 +1288,57 @@ TEST_F(DistributedPlannerTest, IncompleteCompositeJoinStaysLocal) { EXPECT_FALSE(joins.empty()) << "join on a partial composite key must gather"; } +TEST_F(DistributedPlannerTest, CompositeRangePrunesOnFirstKey) { + TableShardConfig cfg; + cfg.table_name = "users"; + cfg.shard_key = "id+name"; + cfg.shards = {{"shard_1"}, {"shard_2"}, {"shard_3"}}; + cfg.strategy = RoutingStrategy::RANGE; + cfg.ranges = {{5, 0}, {10, 1}, {100000, 2}}; + shard_map.add_table(cfg); + + auto count_remotes = [&](const char* sql) { + Parser parser; + auto pr = parser.parse(sql, std::strlen(sql)); + PlanBuilder builder(catalog, parser.arena()); + PlanNode* plan = builder.build(pr.ast); + DistributedPlanner dp(shard_map, catalog, parser.arena()); + PlanNode* dist = dp.distribute(plan); + std::vector remotes; + find_nodes(dist, PlanNodeType::REMOTE_SCAN, remotes); + return remotes.size(); + }; + + EXPECT_EQ(count_remotes("SELECT * FROM users WHERE id <= 5"), 1u); + EXPECT_EQ(count_remotes("SELECT * FROM users WHERE id BETWEEN 6 AND 10"), 1u); +} + +TEST_F(DistributedPlannerTest, SemiJoinPrunesProbeShards) { + Parser parser; + const char* sql = "SELECT * FROM users JOIN orders ON users.id = orders.user_id"; + auto pr = parser.parse(sql, std::strlen(sql)); + ASSERT_EQ(pr.status, ParseResult::OK); + PlanBuilder builder(catalog, parser.arena()); + PlanNode* plan = builder.build(pr.ast); + ASSERT_NE(plan, nullptr); + + DistributedPlanner dp(shard_map, catalog, parser.arena(), + &mock_executor, &functions); + PlanNode* dist = dp.distribute(plan); + ASSERT_NE(dist, nullptr); + + std::vector remotes; + find_nodes(dist, PlanNodeType::REMOTE_SCAN, remotes); + bool saw_users_in = false; + 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 && + remote.find("IN") != std::string::npos) + saw_users_in = true; + } + EXPECT_TRUE(saw_users_in) << "semi-join should push IN on the sharded probe"; +} + TEST_F(DistributedPlannerTest, UnknownTableErrors) { catalog.add_table("", "ghost", { {"id", SqlType::make_int(), false}, diff --git a/tests/test_distributed_real.cpp b/tests/test_distributed_real.cpp index ef904f0..9bf397b 100644 --- a/tests/test_distributed_real.cpp +++ b/tests/test_distributed_real.cpp @@ -17,6 +17,7 @@ #include "sql_engine/function_registry.h" #include "sql_engine/session.h" #include "sql_engine/local_txn.h" +#include "sql_engine/distributed_txn.h" #include "sql_parser/parser.h" #include @@ -365,4 +366,92 @@ TEST_F(LiveShardedWriteTest, InsertThenPointSelect) { session.execute_statement("DELETE FROM shard_write_t WHERE id = 9"); } +bool pg_port_available(uint16_t port) { + std::string conninfo = std::string("host=127.0.0.1 port=") + std::to_string(port) + + " user=postgres password=test dbname=testdb connect_timeout=2"; + PGconn* conn = PQconnectdb(conninfo.c_str()); + bool ok = (PQstatus(conn) == CONNECTION_OK); + PQfinish(conn); + return ok; +} + +TEST(LivePgShardedWriteTest, RangeListAndTwoPhaseCommit) { + if (!pg_port_available(16432) || !pg_port_available(16433)) { + GTEST_SKIP() << "Need PG on 16432 and 16433 (start_pg_sharding_demo.sh)"; + } + + MultiRemoteExecutor exec; + BackendConfig p1; + p1.name = "pg1"; + p1.host = "127.0.0.1"; + p1.port = 16432; + p1.user = "postgres"; + p1.password = "test"; + p1.database = "testdb"; + p1.dialect = Dialect::PostgreSQL; + BackendConfig p2 = p1; + p2.name = "pg2"; + p2.port = 16433; + exec.add_backend(p1); + exec.add_backend(p2); + + InMemoryCatalog catalog; + catalog.add_table("", "users", { + {"id", SqlType::make_int(), false}, + {"name", SqlType::make_varchar(255), true}, + {"age", SqlType::make_int(), true}, + }); + catalog.add_table("", "regions", { + {"name", SqlType::make_varchar(64), false}, + {"tz", SqlType::make_varchar(32), true}, + }); + + ShardMap shards; + TableShardConfig users; + users.table_name = "users"; + users.shard_key = "id"; + users.shards = {{"pg1"}, {"pg2"}}; + users.strategy = RoutingStrategy::RANGE; + users.ranges = {{5, 0}, {100000, 1}}; + shards.add_table(users); + TableShardConfig regions; + regions.table_name = "regions"; + regions.shard_key = "name"; + regions.shards = {{"pg1"}, {"pg2"}}; + regions.strategy = RoutingStrategy::LIST; + regions.list = { + {false, 0, "us-east", 0}, + {false, 0, "us-west", 1}, + }; + shards.add_table(regions); + + Arena txn_arena{65536, 1048576}; + DistributedTransactionManager txn( + exec, DistributedTransactionManager::BackendDialect::POSTGRESQL); + Session session(catalog, txn); + session.set_remote_executor(&exec); + session.set_shard_map(&shards); + + auto east = session.execute_query("SELECT tz FROM regions WHERE name = 'us-east'"); + auto west = session.execute_query("SELECT tz FROM regions WHERE name = 'us-west'"); + ASSERT_EQ(east.row_count(), 1u); + ASSERT_EQ(west.row_count(), 1u); + + EXPECT_TRUE(session.begin()); + auto i0 = session.execute_statement( + "INSERT INTO users (id, name, age) VALUES (0, 'Zero', 1)"); + auto i11 = session.execute_statement( + "INSERT INTO users (id, name, age) VALUES (11, 'Eleven', 2)"); + EXPECT_TRUE(i0.success) << i0.error_message; + EXPECT_TRUE(i11.success) << i11.error_message; + EXPECT_TRUE(session.commit()); + + auto back = session.execute_query("SELECT name FROM users WHERE id IN (0, 11)"); + EXPECT_EQ(back.row_count(), 2u); + + session.execute_statement("DELETE FROM users WHERE id = 0"); + session.execute_statement("DELETE FROM users WHERE id = 11"); + exec.disconnect_all(); +} + } // namespace diff --git a/tests/test_shard_map.cpp b/tests/test_shard_map.cpp index 31dd0a9..a656e82 100644 --- a/tests/test_shard_map.cpp +++ b/tests/test_shard_map.cpp @@ -278,3 +278,22 @@ TEST(ShardMapTest, CompositeHashIsDeterministicAndDiffersFromSingle) { EXPECT_LT(ia, 3u); EXPECT_LT(ib, 3u); } + +TEST(ShardMapTest, CompositeRangeRoutesOnFirstKey) { + TableShardConfig cfg; + cfg.table_name = "kv"; + cfg.shard_key = "tenant_id+id"; + cfg.shards = {ShardInfo{"s0"}, ShardInfo{"s1"}}; + cfg.strategy = RoutingStrategy::RANGE; + cfg.ranges = {ShardRange{5, 0}, ShardRange{100, 1}}; + ShardMap map; + map.add_table(cfg); + + ShardKeyPart low[] = {{true, 3, nullptr, 0}, {true, 99, nullptr, 0}}; + ShardKeyPart high[] = {{true, 9, nullptr, 0}, {true, 1, nullptr, 0}}; + size_t il = 99, ih = 99; + ASSERT_TRUE(map.try_shard_index_for_parts(sref("kv"), low, 2, il)); + ASSERT_TRUE(map.try_shard_index_for_parts(sref("kv"), high, 2, ih)); + EXPECT_EQ(il, 0u); + EXPECT_EQ(ih, 1u); +} diff --git a/tests/test_ssl_config.cpp b/tests/test_ssl_config.cpp index b60a56d..6516e7c 100644 --- a/tests/test_ssl_config.cpp +++ b/tests/test_ssl_config.cpp @@ -237,8 +237,10 @@ TEST(SSLConfigTest, ParseShardSpecCompositeHash) { ASSERT_EQ(ps.config.shards.size(), 3u); } -TEST(SSLConfigTest, ParseShardSpecRejectsCompositeRange) { +TEST(SSLConfigTest, ParseShardSpecCompositeRangeUsesFirstKey) { auto ps = parse_shard_spec("kv:tenant_id+id:range:5=s0,10=s1"); - EXPECT_FALSE(ps.ok); - EXPECT_NE(ps.error.find("HASH"), std::string::npos); + ASSERT_TRUE(ps.ok); + EXPECT_EQ(ps.config.strategy, RoutingStrategy::RANGE); + EXPECT_EQ(ps.config.shard_key, "tenant_id+id"); + ASSERT_EQ(ps.config.ranges.size(), 2u); } diff --git a/tools/sqlengine.cpp b/tools/sqlengine.cpp index ea6adc1..d14ebb4 100644 --- a/tools/sqlengine.cpp +++ b/tools/sqlengine.cpp @@ -356,8 +356,14 @@ int main(int argc, char* argv[]) { } if (remote_exec) { + bool all_pg = !backends.empty(); + for (const auto& b : backends) { + if (b.dialect != Dialect::PostgreSQL) { all_pg = false; break; } + } dtxn.reset(new DistributedTransactionManager( - *remote_exec, DistributedTransactionManager::BackendDialect::MYSQL)); + *remote_exec, + all_pg ? DistributedTransactionManager::BackendDialect::POSTGRESQL + : DistributedTransactionManager::BackendDialect::MYSQL)); if (!txn_log_path.empty()) { if (!txn_log.open(txn_log_path)) { std::cerr << "Error: cannot open txn log " << txn_log_path << std::endl;