Author: Shirly

This article will briefly introduce the basic principles of TiKV Coprocessor, for those who want to understand the process of TiKV data reading and execution, but also for those who want to contribute code to the module. Before reading this article, it is recommended that you understand the overall architecture of TiDB and read three articles about TiDB technology: storage, computing, and scheduling.

What is a Coprocessor

Those familiar with the overall TiDB framework may remember that TiDB is stateless and data is stored in the TiKV layer. When TiDB receives a query request from the client, it obtains specific data information from TiKV. The simplest process for a read request is as follows:

The first thing to be sure of is that this approach does solve the problem, but what about performance? Let’s break it down:

  1. TiKV returns all data, network overhead is too high.
  2. TiDB needs to compute all the data and consumes a lot of CPU, whereas TiKV doesn’t do any computation and is very idle.

After seeing the above questions, you, as smart as you, may easily think, can TiKV do a calculation for the part of the data it is responsible for, and then return it to TiDB?

Why not?

The module that TiKV reads data and calculates is defined as Coprocessor. This concept is inspired by HBase. At present, the implementation of TiDB is similar to the Endpoint of Coprocessor in HBase, and can also be analogous to MySQL stored procedures.

With Coprocessor, how does a read request get sent to TiKV from a macro perspective? Take the following requests for example:

As shown, the above query statements are processed in TiDB as follows:

  1. TiDB receives the query statement, analyzes the statement, calculates the physical execution plan, and organizes the Coprocessor request called TiKV.
  2. TiDB distributes the Coprocessor request to all relevant TiKV based on the distribution of data.
  3. After receiving the Coprocessor request, TiKV filters and aggregates the data according to the request operator and then returns it to TiDB.
  4. After TiDB receives the return results of all data, it performs secondary aggregation, calculates the final results and returns them to the client.

Overview of main functions and processing

Read requests processed by TiKV Coprocessor are classified into three types:

  • DAG: Performs physical operators to compute intermediate results for SQL, thereby reducing TiDB computation and network overhead. This is what the Coprocessor does in most scenarios.
  • Analyze: Analyze table data, statistics, sampling table data information, after persistence is used by TiDB optimizer.
  • CheckSum: verifies table data and verifies data consistency after data import.

When did TiKV distinguish between the three requests after receiving the Coprocessor request?

Request to TiKV layer, processing process is as follows:

  • The gRPC Server receives the request and distributes it to the Coprocessor Endpoint for processing.
  • After receiving the request, the Endpoint distributes the request to the corresponding thread pool based on the priority of the request.
  • All requests asynchronously fetch a Snapshot from the storage tier and then begin the actual processing phase.
  • According to different types of requests, different handlers are constructed to process data.

Currently, among the three interfaces supported by Coprocessor, the latter two interfaces are relatively simple, while DAG is the most complex and most commonly used, so this paper will focus on DAG class requests.

An overview of the DAG Request

DAG, as its name implies, is a directed acyclic graph composed of a series of operators known as Executors in the code.

At present, DAG request mainly realizes two calculation models:

  • Volcano model: each operator spit out as needed in line, deprecated after 3.0.
  • Vectorization calculation model: each operator batch processing data, after 3.0 began to spread.

On the current TiKV Master, the volcano model is in the transition stage of the vectorization model, so the two calculation models exist simultaneously. When TiKV receives the request, it will preferentially detect whether it can go to the quantization model. If some functions are not realized in the quantization model, it will go to the old calculation model. The specific processing logic flow is as follows:

Related code in the SRC/coprocessor/dag/mod. The rs.

Since the volcano model has been deprecated, we will focus only on vectorization.

An overview of the operator

In the vectorization model, all operators implement the BatchExecutor interface, which mainly defines a get_batch function:

pub trait BatchExecutor: Send {
   fn next_batch(&mut self, scan_rows: usize) -> BatchExecuteResult;
}

pub struct BatchExecuteResult {
   pub physical_columns: LazyBatchColumnVec,
   pub logical_rows: Vec<usize>,
   pub is_drained: Result<bool, Error>,
   ...
}
Copy the code

Parameter Description:

  • In next_batch, scan_rows is controlled by the upper layer. The number of scan_rows is doubled from 32 to 1024 because it is slow to scan too much data.

  • Return value In BatchExecuteResult, a single field is used to indicate that all data is processed because a batch of empty data is returned. For example, all data may be filtered.

At present, TiKV supports the following types of operators.

TableScan

  • Definition: scans table data based on the specified primary key range and returns filtered columns. It will only appear as the lowest level operator, getting data from the bottom KV.

  • Source path: components/tidb_query/SRC/batch/executors table_scan_executor. Rs

  • Select col from t

IndexScan

  • Definition: Returns scan index data based on the specified index and filters out some index columns. It will only appear as the lowest level operator, getting data from the bottom KV.

  • Source path: components/tidb_query/SRC/batch/executors index_scan_executor. Rs

  • Select index from t

Selection

  • Definition: The result of the underlying operator is filtered according to filter conditions, which consist of multiple expressions.

  • Source path: components/tidb_query/SRC/batch/executors selection_executor. Rs

  • Select col from t where a+b=10

Limit

  • Definition: returns a limited number of rows from the data spit out by the underlying operator.

  • Source path: components/tidb_query/SRC/batch/executors limit_executor. Rs

  • Select col from t limit 10

TopN

  • Definition: Retrieves the first several rows of data after sorting according to a given expression.

  • Source path: components/tidb_query/SRC/batch/executors top_n_executor. Rs

  • Select col from t order by a+1 limit 10

Aggregation

  • Definition: Groups and aggregates according to a given expression.

  • Source path: components/tidb_query/SRC/batch/executors / * _aggr_executor in rs

  • Select count(1) from t group by score + 1

Use a mixture of operators

To sum up, each operator can be arbitrarily combined in the following ways, as shown in the figure below:

Select count(1) from t where age>10

summary

For space reasons, this article only gives you an overview of the Coprocessor, so you need to get a sense of it. In the future, we will release more in-depth analysis of the source code related to this module. You are welcome to continue reading and give constructive suggestions for improvement.

pingcap.com/blog-cn/tik…