国测 vexdb
-- cn: 50000 -- dn1: 50005 -- dn2: 50010 vsql -d postgres -p 50000 -r SELECT * FROM PG_DIST_NODE; CREATE TABLE t1(c1 INT, c2 TEXT); SELECT create_distributed_table('t1', 'c1'); -- 创建8个分片，dn1和dn2，每个dn有4个分片，即4个表，表名为 t1_$oid SELECT * FROM pg_dist_shard; -- min和max是hash之后的值 SELECT * FROM pg_dist_shard_placement; SELECT * FROM pg_dist_partition; INSERT INTO t1 VALUES(1, 'aa'), (3, 'bb'), (7, 'cc'), (10, 'dd'); vsql -d postgres -p 50005 -r SELECT oid,relname FROM pg_class WHERE relname like '%t1%'; vsql -d postgres -p 50010 -r SELECT oid,relname FROM pg_class WHERE relname like '%t1%'; # cn SELECT create_distributed_table('t1', 'c1'); # --> shard 1 (dn1) SELECT worker_apply_shard_ddl_command (12042, 'CREATE TABLE public.t1 (c1 integer, c2 text) WITH (orientation=row, compression=no, fillfactor=80)'); # dn1 worker_apply_shard_ddl_command ddlnode = ParseTreeNode(sql) RelayEventExtendNames(ddlnode, sharid) # 在语法解析树中，把t1名称，修改为t1_$shardid ProcessUtilityParseTree SELECT worker_apply_shard_ddl_command (12042, 'ALTER TABLE public.t1 OWNER TO shenkun'); # --> shard 2 (dn2) SELECT worker_apply_shard_ddl_command (12043, 'CREATE TABLE public.t1 (c1 integer, c2 text) WITH (orientation=row, compression=no, fillfactor=80)'); SELECT worker_apply_shard_ddl_command (12043, 'ALTER TABLE public.t1 OWNER TO shenkun'); # --> shared 3 (dn1) ... # --> 一共8个分片，每个分片2条SQL！ # cn INSERT INTO t1 VALUES(1,'aa'),(3,'bb'),(7,'cc'),(10,'dd'); exec_simple_query pg_parse_query pg_analyze_and_rewrite pg_plan_query [planner hook] distributed_planner CreateDistributedPlan CreateModifyPlan RouterInsertJob (deferredPruning=true) # 返回 CustomScan PortalStart ExecutorStart [DVecExecutorStart] DVecExecutorStart BeginCustomScan → DVecBeginScan DVecBeginModifyScan RegenerateTaskListForInsert BuildRoutesForInsert FindShardInterval ActiveShardPlacementList RebuildQueryStrings DeparseTaskQuery AppendShardIdToName # 生成改名后SQL PortalRun ExecutorRun [ExecutorRun_hook] DVecExecutorRun ExecCustomScan DVecExecScan AdaptiveExecutor RunDistributedExecution ... SendRemoteCommand StartRemoteTransactionBegin BeginTransactiionCommand # 生成：BEGIN ISOLATION LEVEL READ COMMITTED; AssignDistributedTransactionIdCommand # 生成：SELECT assign_distributed_transaction_id(0,20,'...'); SendRemoteCommand # 一次发送2条事务SQL SendRemoteCommand PQsendQuery # INTO t1_$shardid .. CoordinatedRemoteTransactionsPrepare .. SendRemoteCommand # 发送 REPARE TRANSACTION .. SendRemoteCommand # 发送 COMMIT PREPARED .. # --> shard 1 (dn1) BEGIN ISOLATION LEVEL READ COMMITTED; SELECT assign_distributed_transaction_id(0,20,'...'); INSERT INTO public.t1_12042 AS dvec_table_alias (c1, c2) VALUES (1,'aa'::text) REPARE TRANSACTION 'dvecx_...'; COMMIT PREPARED 'dvecx_...'; # 执行分布式事务 INSERT INTO public.t1_12042 AS dvec_table_alias (c1, c2) VALUES (1,'aa'::text) # --> shard 2 (dn2) INSERT INTO public.t1_12043 AS dvec_table_alias (c1, c2) VALUES (10,'dd'::text) # --> shared 3 (d1) ... # --> 一共8个分片，其中4个分片，各收到1条INSERT语句 # 多行在同一分片，INSERT语句合并 # cn SELECT * FROM t1 WHERE c1 > 10; exec_simple_query pg_parse_query pg_analyze_and_rewrite pg_plan_query [planner hook] distributed_planner CreateDistributedPlan GetRouterPlanType (一造 multi-shard fan-out(非 router 计划)) MultiLogicalPlanCreate (logical planner:建逻辑计划) CreatePhysicalDistributedPlan (physical planner) ├─ 为命中的每个分片建一个 Task(本表 8 个分片 → 8 个 Task) UpdateTaskQueryString DeparseTaskQuery deparse_shard_query AppendShardIdToName # t1 → t1_<shardId> # SELECT * + c1,c2 ; WHERE c1>10 → (c1 OPERATOR(pg_catalog.>) 10) # 包成 CustomScan 返回 PortalStart ExecutorStart [DVecExecutorStart] DVecExecutorStart BeginCustomScan → DVecBeginScan PortalRun ExecutorRun [ExecutorRun_hook] DVecExecutorRun ExecCustomScan DVecExecScan AdaptiveExecutor RunDistributedExecution AssignTasksToConnectionsOrWorkerPool # 8 个 Task 分给 DN1/DN2 的连接 ConnectionStateMachine StartRemoteTransactionBegin *发送: BEGIN + assign(每条 DN 连接首次用,只发一次) BeginTransactionCommand # 生成: BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED AssignDistributedTransactionIdCommand # 生成: SELECT assign_distributed_transaction_id(...) SendRemoteCommand → PQsendQuery *两条合并成一条发出 # 发送: SELECT c1,c2 FROM t1_<shardId> WHERE (c1>10) # DN1 上 4 条 (12042/12044/12046/12048), DN2 上 4 条 (COMMIT) RemoteTransactionCommit SendRemoteCommand → PQsendQuery # 只读 → 普通 COMMIT(无 PREPARE/COMMIT PREPARED,无 2PC) # COMMIT # --> shard 1 (dn1) SELECT c1, c2 FROM public.t1_12042 t1 WHERE (c2 OPERATOR(pg_catalog.>) 'cc'::text) SELECT * FROM t1 WHERE c2 > 'cc'; 密态： cn CREATE TABLE t1 (c1 ..) INSERT INTO t1 VALUES (..) SELECT c1,c2 FROM t1 WHERE .. dn CREATE TABLE t1_$shardid (c1 ..) INSERT INTO t1_$shardid VALUES (..) # values的值不变 SELECT c1,c2 FROM t1_$shardid WHERE .. # WHERE子句不变