crdb 语句执行流程

connection manager

连接管理，接受用户连接。

• 相关文件
  • main.go
  • pkg/cli/cli.go
  • pkg/cli/start.go
  • pkg/server/server.go

• 函数调用堆栈

>main.main                      main.go
|>cli.Main                      pkg/cli/cli.go
| |>cli.runStart                pkg/cli/start.go
| | |>(s *Server) Start         pkg/server/server.go
| | | |>(s *Server) ServeConn   pkg/sql/pgwire/server.go

Postgres Wire Protocol

pgwire.v3Conn结构体接收客户端的连接，身份验证，按照pg通信协议解析语句。
v3Conn.serve() 实现读取SQL，执行SQL，返回结果集的循环。

• 相关文件

  • pkg/sql/pgwire/v3.go

• 函数调用堆栈

>(s *Server) ServeConn              pkg/sql/pgwire/server.go
|>(c *v3Conn) serve                 pkg/sql/pgwire/v3.go
| |>(b *readBuffer) readTypedMsg    pkg/sql/pgwire/encoding.go
| |>(c *v3Conn) handleSimpleQuery   pkg/sql/pgwire/v3.go

SQL Executor

sql.Executor具体负责解析SQL，执行SQL，返回结果集给pgwire.v3Conn。

入口函数是Executor.execRequest(), 接收一批SQL语句，将执行交给sql.Session进行处理。

Parsing

解析是讲SQL语句根据YACC语法规则解析成抽象语法树(AST)。sql.y文件是从PG直接拷贝过来，做了一定的裁剪，然后又新增了一些语法。

AST node都定义在sql/parser，分为两类， statements 和expressions.AST最后会经过planner转换为执行计划。

• 相关文件

  • pkg/sql/parser/sql.y
  • pkg/sql/executor.go
  • pkg/sql/parser/parse.go

• 函数调用堆栈

>pgwire.(*v3Conn).serve
| >pgwire.(*v3Conn).handleSimpleQuery
| |>sql.(*Executor).ExecuteStatements
| | |>sql.(*Executor).execRequest
| | | |>parser.Parse
| | | | |>parser.(*Parser).Parse
| | | | | |>parser.(*sqlParserImpl).Parse  //generated by goyacc from sql.y

Statement Execution

重点调用runTxnAttempt函数进入执行阶段，执行里面包含生成执行计划。

• 相关文件

  • pkg/sql/executor.go

• 函数调用堆栈

>sql.(*Executor).execRequest
| >sql.(*Executor).execParsed
| |>sql.runWithAutoRetry
| | |>sql.runTxnAttempt

Building execution plans

crdb的执行计划是planNode节点组成的树形结构，类似AST，但是包含了语义信息和运行状态。计划树是通过planner.makePlan()生成的，接收经过parsed的语句，返回planNode树的根节点，这个过程中会进行语义检查以及各种表达式变化。

planner查看AST根节点从而知道语句类型，根据不同的语句类型，调用不同的计划生成函数。比如SELECT语句的计划树是通过planner.SelectClause()生成。

SELECT计划树包含各种处理逻辑：scanNode进行全表扫描，filterNode进行过滤WHERE条件, sortNode进行ORDER BY操作。

最后计划树会经过简化和优化，比如去掉不必要的节点。

• 相关文件

  • pkg/sql/executor.go
  • pkg/sql/plan.go

• 函数调用堆栈

>sql.runTxnAttempt
| >sql.(*Executor).execSingleStatement
| | >sql.(*Executor).execStmtInOpenTxn
| | |>sql.(*Executor).execStmt
| | | |>sql.(*planner).makePlan

Expressions

parser.Expr是AST的子集，表示各种表达式, 出现在WHERE、LIMIT、ORDER BY条件中。所有的表达式都要经过planner.analyzeExpr处理。每个planNode负责调用analyzeExpr分析本节点所包含的表达式。

planner.analyzeExpr执行如下流程：

1. resolving names
2. normalization
3. type checking
4. 使用sql.subquery替换sub-query

Notable planNodes

每个planNode包行两个方法：

1. Start: 初始化
2. Next: 生成下一行数据

• renderNode的start方法调用堆栈

>sql.(*Executor).execClassic
| >sql.(*planner).startPlan
| | >sql.(*renderNode).Start

有两种执行方式：

1. executor.execClassic()
2. executor.execDistSQL()

• execClassic函数调用堆栈

>sql.runTxnAttempt
| >sql.(*Executor).execSingleStatement
| | >sql.(*Executor).execStmtInOpenTxn
| | |>sql.(*Executor).execStmt
| | | |>sql.(*Executor).execClassic

KV

crdb kv层处理请求的执行，请求和响应统一使用基于protocol buffer的API，消息格式定义在pkg/roachpb/api.proto。实际上，KV client发送的是批量请求BatchRequest s, 所有的请求都有一个Header，包含消息的目标以及事务信息。

The KV client interface

KV client包含启动事务的接口starting a (KV) transaction, Txn 是事务上下文结构体。

• Txn.Run()最终调用txn.db.sender.Send(..., batch):
• 请求最终会传递到需要执行的副本节点。
• Send() 最终传递到最底层接口。TxnCoordSender -> DistSender -> Node -> Stores -> Store -> Replica

TxnCoordSender

TxnCoordSender是最上层的client.Sender. TxnCoordSender负责处理事务状态，一个事务开启后，TxnCoordSender异步发送心跳信号给txn record, 保持连接, 同时会记录整个事务的所有写入的key。事务提交或回滚时，会清理write intents。事务中所有的语句都要经过TxnCoordSender, 然后将请求传递给DistSender。

DistSender

DistSender
这一层任务繁重，需要处理gateway节点和range节点之间的通信，接受BatchRequests，将请求路由到不同的range node，然后从不同的range node接收结果集进行组装。

1. DistSender.Send() 调用DistSender.divideAndSendBatchToRanges() 将请求按照range进行细分。
2. 每个细分的请求，调用DistSender.sendPartialBatchAsync() 将请求发到对应的range所在的node。
3. 将细分请求发送到对应range的正确的副本，是通过RPC请求进行的。
4. sendToReplicas函数用于初始化RPC请求，并发送请求。
5. RPC的发送隐藏在Transport interface, grpcTransport.SendNext() 进行gRPC的调用。
6. 从不同副本返回的响应合并成单独的BatchResponse，最终返回给Send()方法。

RPC server - Node and Stores

RPC服务是通过Node结构体实现的。Node将请求直接传递给其成员Stores进行处理。

Stores.Send()根据确定目标副本，并将请求路由到目标副本。

Store - intent resolution

每个store代表着一个物理磁盘设备. Store除了路由请求到目标副本外，还有个重要角色，处理write intents记录，通过这个达到处理R-W，W-W冲突的目的。

碰到WriteIntentError后，需要进行try to "resolve"
it，使用
intentResolver 解决冲突.

Replica - executing reads, proposing Raft commands

Replica代表一个range, Replica是Sender的最后一层,其它上层的Sender最终都是将请求发送给LeaseHolder的Replica, 读请求和写请求处理逻辑不同，读直接返回数据，写请求需要通过raft一致性协议进行。

Read request path

读请求首先会检查请求是否真正发送到了LeaseHolder上，是通过replica.redirectOnOrAcquireLease()函数进行检查，如果当前replica不是leaseholder，则把请求转给leaseholder，如果当前不存在leaseholder，则会发出申请当前节点为leaseholder的请求。

确定leaseholder后，如果所读的key正在被其它事务写，需要进行同步等待。同步通过
CommandQueue进行

• 函数调用堆栈

>storage.(*Stores).Send
|>storage.(*Store).Send
| |>storage.(*Replica).Send
| | |>storage.(*Replica).executeWriteBatch
| | | |>storage.(*Replica).tryExecuteWriteBatch
| | | | |>storage.(*Replica).redirectOnOrAcquireLease

>storage.(*Stores).Send
|>storage.(*Store).Send
| |>storage.(*Replica).Send
| | |>storage.(*Replica).executeReadOnlyBatch
| | | |>storage.evaluateBatch
| | | | |>storage.evaluateCommand
| | | | | |>storage.evalScan
| | | | | | |>engine.MVCCScan

Engine

Engine是crdb技术堆栈的最底层，负责物理存储。Engine是接口层，底部实现目前只有RocksDB, 是基于RocksDB C库的一个封装，使用了cgo与C进行交互。尽管RocksDB是crdb非常重要的组成部分，crdb并没有去修改RocksDB。

>storage.(*Stores).Send
|>storage.(*Store).Send
| |>storage.(*Replica).Send
| | |>storage.(*Replica).executeReadOnlyBatch
| | | |>storage.evaluateBatch
| | | | |>storage.evaluateCommand
| | | | | |>storage.evalScan
| | | | | | |>engine.MVCCScan
| | | | | | | |>engine.mvccScanInternal
| | | | | | | | |>engine.MVCCIterate
| | | | | | | | | |>engine.mvccGetInternal

Write request path

写请求比读更加复杂，开始于replica.addWriteCmd().这个函数会不断调用replica.tryAddWriteCmd. 其主要做如下事情：

1. 等待当前其它正在修改的请求结束，将自己放到CommandQueue。
2. 调用redirectOnOrAcquireLease检查leaseholder
3. 与timestamp cache比较，如果不合格，写的时间戳需要提升。
4. 将请求转换为raft command, 调用replica.propose，将代码权交给raft机制进行处理
5. 等待replica.propose返回
6. 调用返回，将写请求命令清除出CommandQueue。

Evaluation of requests and application of Raft commands

>storage.(*Replica).propose
| >storage.(*Replica).requestToProposal
| | >storage.(*Replica).evaluateProposal
| | | >storage.(*Replica).evaluateProposalInner
| | | | >storage.(*Replica).evaluateTxnWriteBatch
| | | | | >storage.evaluateBatch
| | | | | | >storage.evaluateCommand
| | | | | | | >storage.evalPut
| | | | | | | | >engine.MVCCPut

Raft command application