PointGet 的一生

   期间，看过某些模块，如：[PD 如何调度 Region](https://asktug.com/t/topic/242808)，也只是窥探数据库的局部功能，总有一种 **“根本不了解数据库逻辑概念和代码是怎样关联的！”** 的感觉。因此借着 **“点查”** 避开大量复杂优化器代码的机会，尽己所能的串联一下点查在 TiDB 和 TiKV 间的执行流程。       
   此后，越发觉得 **“产研”** 及 **“交付”** 价值的不同，比如：不能说看代码就很牛，在文档丰富、产品成熟的前提下，**学习文档才是较快速、较全面、较有价值的方法**，更多思考见总结部分。

// Run runs the server.func (s *Server) Run() error {    go s.startNetworkListener(s.listener, false, errChan)    go s.startNetworkListener(s.socket, true, errChan)        ......}
func (s *Server) startNetworkListener(listener net.Listener, isUnixSocket bool, errChan chan error) {    for {        conn, err := listener.Accept()        ......        go s.onConn(clientConn)        }}

func (s *session) ParseSQL(ctx context.Context, sql string, params ...parser.ParseParam) ([]ast.StmtNode, []error, error) {    ......    p := parserPool.Get().(*parser.Parser)    defer parserPool.Put(p)    p.SetSQLMode(s.sessionVars.SQLMode)    p.SetParserConfig(s.sessionVars.BuildParserConfig())    tmp, warn, err := p.ParseSQL(sql, params...)    res := make([]ast.StmtNode, len(tmp))    copy(res, tmp)    return res, warn, err}

   **2. 其次**，在 Compile 内部调用 [func Preprocess(...)](https://github.com/pingcap/tidb/blob/27348d67951c5d9e409c84ca095f0e5d3332c1fd/planner/core/preprocess.go#L114-L130)，进行 Preprocess 完成前置检查，如：语义检查。具体实现流程为，通过 AST 的 [Accept 方法](https://github.com/pingcap/tidb/blob/27348d67951c5d9e409c84ca095f0e5d3332c1fd/parser/ast/ast.go#L40), 构造一个 Vistor 实现对 AST 的遍历。 每个 Visitor 接口包含 Enter、Leave 方法，[并在 Enter 或 Leave 时](https://github.com/pingcap/tidb/blob/27348d67951c5d9e409c84ca095f0e5d3332c1fd/planner/core/preprocess.go#L192)，依据 SQL 类型进行判断。本例 point get 会跳到 [func (n \*SelectStmt) Accept(v Visitor)](https://github.com/pingcap/tidb/blob/27348d67951c5d9e409c84ca095f0e5d3332c1fd/parser/ast/dml.go#L1391-L1503) 中，不断分支处理完成遍历。详情参考 [TiDB 源码阅读之 Compiler --> 进度 10min 左右](https://www.bilibili.com/video/BV1m4411g7Yy?from=search\&seid=10358982920062599002\&spm_id_from=333.337.0.0)。        
   **3. 最后**，进入 Optimizer 处理，本例中因为是点查会越过大量优化器处理过程，直接进入 [func TryFastPlan(...)](https://github.com/pingcap/tidb/blob/27348d67951c5d9e409c84ca095f0e5d3332c1fd/planner/optimize.go#L131) 进行简单的 “权限检查” 及 “数据库名检查”。 

func TryFastPlan(ctx sessionctx.Context, node ast.Node) (p Plan) {    ......    case *ast.SelectStmt:        if fp := tryPointGetPlan(ctx, x, isForUpdateReadSelectLock(x.LockInfo)); fp != nil {            if checkFastPlanPrivilege(ctx, fp.dbName, fp.TblInfo.Name.L, mysql.SelectPriv) != nil {                return nil            }            if tidbutil.IsMemDB(fp.dbName) {                return nil            }            if fp.IsTableDual {                return            }            p = fp            return        }    }    return nil}

   **2. 其次**，ExecuteStmt 返回的 RecordSet 包含了 Executor 需要的所有信息，此时可以把 RecordSet 当成返回的结果集，但该执行流从未触发过 Next() 方法，即：没有真正的获取数据。        
   **3. 最后**，实际上 Next（） 触发由 的 writeResultset 触发,详细流程如下：**func (cc \*clientConn) handleStmt(...) --> func (cc \*clientConn) writeResultset(...) --> func (cc \*clientConn) writeChunks(...) --> func (trs \*tidbResultSet) Next(...) --> func (a \*recordSet) Next(...) --> func Next(...) --> func (e \*PointGetExecutor) Next(...) --> func (e \*PointGetExecutor) Next(...) --> func (e \*PointGetExecutor) getAndLock(...) --> func (e \*PointGetExecutor) get(...)** 层层调用，直至 Executor 完成所有数据的获取。        
   **4. 另外**，值得一提的是在 func (a \*ExecStmt) Exec(...) --> Build Executor 时，因为点查使用“主键”或“唯一索引”标定一行，**不存在重复数据读**，所以在 `autoCommit` 情况下直接取 `MaxUint64` 作为事务的 StartTS（该事务只有一个点查），即：无穷大 +∞。同时，还会赋予 `PriorityHigh` 优先级进行处理，详情见 [force-priority](https://docs.pingcap.com/zh/tidb/v5.2/tidb-configuration-file/#force-priority)。 

func (a *ExecStmt) buildExecutor() (Executor, error) {        ......         } else {            // Do not sync transaction for Execute statement, because the real optimization work is done in            // "ExecuteExec.Build".            useMaxTS, err := plannercore.IsPointGetWithPKOrUniqueKeyByAutoCommit(ctx, a.Plan)            if useMaxTS {                if err := ctx.InitTxnWithStartTS(math.MaxUint64); err != nil {                    return nil, err                }            }            if stmtPri := stmtCtx.Priority; stmtPri == mysql.NoPriority {                switch {                case useMaxTS:                    stmtCtx.Priority = kv.PriorityHigh                case a.LowerPriority:                    stmtCtx.Priority = kv.PriorityLow                }            }        }    }    b := newExecutorBuilder(ctx, a.InfoSchema, a.Ti, a.SnapshotTS, a.IsStaleness, a.ReplicaReadScope)    e := b.build(a.Plan)    return e, nil}

   **2. 其次**，DEBUG 到 TiKV Client 内部细节会发现 TiKV Client 遵照本地缓存是否存在数据，如果不存在构造请求头，并在 for{} 循环中调用 GetRegionCache() 查询 PD Client 的 Region Cache 获取所要查询 Region 在 TiKV 中的位置，向 TiKV 发送请求获取数据。之所以使用 for 循环，是因为 Region Cache 信息从 PD 获取，所以并不一定是最新的、最准确的 Region Location 信息，包含一些错误重拾操作,如：EpochNotMatch 等等，详情见: [Region Cache 缓存和清理逻辑解释](https://docs.google.com/document/d/1BHkeN8W2iWSdgWLK3U16QGwq9K7DA6Le7zxVhLPaQgc/edit) 或 [TiDB 源码阅读系列文章（十八）tikv-client（上）](https://pingcap.com/zh/blog/tidb-source-code-reading-18)。 

func (s *KVSnapshot) get(ctx context.Context, bo *retry.Backoffer, k []byte) ([]byte, error) {    // Check the cached values first.    s.mu.RLock()    if s.mu.cached != nil {        if value, ok := s.mu.cached[string(k)]; ok {            atomic.AddInt64(&s.mu.hitCnt, 1)            s.mu.RUnlock()            return value, nil        }    }    s.mu.RUnlock()    ......    s.mu.RLock()    req := tikvrpc.NewReplicaReadRequest(tikvrpc.CmdGet,        &kvrpcpb.GetRequest{            Key:     k,            Version: s.version,        }, s.mu.replicaRead, &s.replicaReadSeed, kvrpcpb.Context{            Priority:         s.priority.ToPB(),            NotFillCache:     s.notFillCache,            TaskId:           s.mu.taskID,            ResourceGroupTag: s.resourceGroupTag,        })    s.mu.RUnlock()
    for {        loc, err := s.store.GetRegionCache().LocateKey(bo, k)        resp, _, _, err := cli.SendReqCtx(bo, req, loc.Region, client.ReadTimeoutShort, tikvrpc.TiKV, "", ops...)        regionErr, err := resp.GetRegionError()        val := cmdGetResp.GetValue()        return val, nil    }}

func (e *PointGetExecutor) Next(ctx context.Context, req *chunk.Chunk) error {    if e.idxInfo != nil {        if isCommonHandleRead(e.tblInfo, e.idxInfo) {            handleBytes, err := EncodeUniqueIndexValuesForKey(e.ctx, e.tblInfo, e.idxInfo, e.idxVals)            e.handle, err = kv.NewCommonHandle(handleBytes)        } else {            e.idxKey, err = EncodeUniqueIndexKey(e.ctx, e.tblInfo, e.idxInfo, e.idxVals, tblID)            e.handleVal, err = e.get(ctx, e.idxKey)        }    }    ......    return nil} 

   **2. 其次**，从 Grpc 的请求中解析出 key 等相关信息，作为参数传递调用存储引擎方法 [storage.get(...)](https://github.com/tikv/tikv/blob/005efd56d3405b72a26bf623b5895cf3d9da0a6f/src/server/service/kv.rs#L1302-L1305) 进行实际的调用处理,包含构造 snapshot、获取 value 等。        
   **5. 最后**，在 v(value) 异步结果获取后，构造 Grpc 返回给请求端。 

fn future_get<E: Engine, L: LockManager>(    storage: &Storage<E, L>,    mut req: GetRequest,) -> impl Future<Output = ServerResult<GetResponse>> {
    let v = storage.get(        req.take_context(),        Key::from_raw(req.get_key()),        req.get_version().into(),    );
    async move {        let mut resp = GetResponse::default();            match v {                Ok((val, stats)) => {                    match val {                        Some(val) => resp.set_value(val),                        None => resp.set_not_found(true),                    }                }                Err(e) => resp.set_error(extract_key_error(&e)),            }        Ok(resp)    }}

   **2. 其次**，观察函数逻辑，先会对请求的优先级进行判断，然后从 read\_pool 线程池中 spawn 一个 handle，并为该 handle 指定解析出来的 “优先级” 在存储层执行请求，关于 SQL 优先级详见 --> [force-priority 说明](https://docs.pingcap.com/zh/tidb/stable/tidb-configuration-file#force-priority)。        
   **3. 再次**，prepare\_snap\_ctx(...) 会进行内存锁冲突检查，并基于 key、start\_ts 等信息构造 snapshot context。        
   **4. 然后**，通过传入 snap\_ctx ,在 [`let snapshot = Self::with_tls_engine(|engine| Self::snapshot(engine, snap_ctx)).await?;`](https://github.com/tikv/tikv/blob/005efd56d3405b72a26bf623b5895cf3d9da0a6f/src/storage/mod.rs#L566) 步构造出存储引擎的 snapshot。        
   **5. 最后**，并在 [`let result = snap_store.get(&key, &mut statistics)});`](https://github.com/tikv/tikv/blob/005efd56d3405b72a26bf623b5895cf3d9da0a6f/src/storage/mod.rs#L581-L588) 中获取 result 结果。 

pub fn get(        &self,        mut ctx: Context,        key: Key,        start_ts: TimeStamp,    ) -> impl Future<Output = Result<(Option<Value>, KvGetStatistics)>> {
        let priority = ctx.get_priority();
        let res = self.read_pool.spawn_handle(            async move {
                let snap_ctx = prepare_snap_ctx(                    &ctx,                    iter::once(&key),                    start_ts,                    &bypass_locks,                    &concurrency_manager,                    CMD,                )?;                let snapshot = Self::with_tls_engine(|engine| Self::snapshot(engine, snap_ctx)).await?;                {                    let snap_store = SnapshotStore::new(                        snapshot,                        start_ts,                        ctx.get_isolation_level(),                        !ctx.get_not_fill_cache(),                        bypass_locks,                        access_locks,                        false,                    );                    let result = snap_store.get(&key, &mut statistics)});
                    Ok((                        result?,                        KvGetStatistics {                            stats: statistics,                            perf_stats: delta,                            latency_stats,                        },                    ))                }            }            .in_resource_metering_tag(resource_tag),            priority,            thread_rng().next_u64(),        );    }

   **2. 其次**，**从 3.2.2 第 5 步构造 snap\_store.get(...) 部分深入**，对于点查会构造一个 point\_getter ，再 get 对应 value,大概逻辑是通过事务的隔离级别分支处理（现阶段所有请求均是 SI 隔离界别）。在 SI 分支中，针对该 User Key 扫一遍 lock CF ,详情参见代码 --> [`impl<S: Snapshot> PointGetter<S>`](https://github.com/tikv/tikv/blob/005efd56d3405b72a26bf623b5895cf3d9da0a6f/src/storage/mvcc/reader/point_getter.rs#L202-L210)； 因为自动提交的点查使用 max\_ts，所以这一步会返回空，意味着查询最新的已提交的 Default CF 数据即可。        
   **3. 最后**，排除锁信息后，进入 load\_data(user\_key) 会起一个 loop ，构造一个包含 cursor 的WriteRef 不断的扫 Write CF。因为事务提交后，会在 Write CF 中写一个 key 为 {user\_key}{commit\_ts}，value 为 {type}{start\_ts} 的记录，所以扫 PUT 类型的 Write CF 意味着可以判断出：查询的数据是否存在 Write CF 中，因为 [MVCC 数据读取](https://pingcap.com/zh/blog/tikv-source-code-reading-13) 指出小于 64 字节的数据会直接内嵌在 Lock Info 或 Write Info，否则调用 load\_data\_from\_default\_cf(...) 从 Default CF 中获取查询的结果值。 

fn load_data(&mut self, user_key: &Key) -> Result<Option<Value>> {
        loop {            let write = WriteRef::parse(self.write_cursor.value(&mut self.statistics.write))?;
            match write.write_type {                WriteType::Put => {                    match write.short_value {                        Some(value) => {                            // Value is carried in `write`.                            self.statistics.processed_size += user_key.len() + value.len();                            println!("-short-->{}<---",String::from_utf8_lossy(value));                            return Ok(Some(value.to_vec()));                        }                        None => {                            let start_ts = write.start_ts;                            let value = self.load_data_from_default_cf(start_ts, user_key)?;                            println!("-default-->{:?}<---",String::from_utf8_lossy(&value));                            self.statistics.processed_size += user_key.len() + value.len();                            return Ok(Some(value));                        }                    }                }            }
            if !self.write_cursor.next(&mut self.statistics.write) {                return Ok(None);            }        }    }