PolarDB:Technical architecture of IMCI

The PolarDB team uses a whitelist mechanism to achieve the compatibility goal. The whitelist mechanism is used by taking into account the following factors:

• The limitation of the available system resources, mainly memory resources.

  In most cases, columnar indexes are not created on all columns in all tables of a database. When a query statement needs to use a column that does not exist in the column store, the statement cannot be executed on the column store.

• The performance.

  The PolarDB team rewrites a column-oriented SQL execution engine that involves all physical execution operators and expression computing. The coverage of supported scenarios is inadequate compared with the native row store of MySQL. When an issued SQL statement contains some operator fragments or column types that are not supported by the IMCI execution engine, the execution engine must be able to identify and intercept the SQL statement, and switch back to the row store for execution.

The whitelist mechanism checks data types, operators, and expressions in SQL statements and other scenarios, such as scenarios in which multi-statements are not supported.

MySQL has been evolving for decades. MySQL supports various column types and rich SQL syntax. In IMCI, the initial focus is on optimizing the most common SQL performance issues in analytical query statements. Even if the applicable scenarios of IMCI are limited, the SQL syntax supported by IMCI is more compatible with MySQL than most OLAP systems. When an issued SQL statement cannot be executed on the column store, the SQL statement directly falls back to the native execution engine of MySQL. This way, 100% compatibility with MySQL is achieved.

The purpose of plan conversion is to convert the abstract syntax tree (AST) representation of a native logical execution plan in MySQL into a logical plan of IMCI. After a logical plan of IMCI is generated, a physical plan is generated after a series of optimizations.

Plan conversion is simple and direct. Users need to only traverse the execution plan tree and convert the AST that is optimized by MySQL into a tree structure with relation operators as its nodes in IMCI. In this process, implicit conversion of types is performed to ensure compatibility with the flexible type system of MySQL.

The plan conversion process generates an equivalent logical plan. The logical plan cannot be executed by the executor and must be converted into a physical plan for execution. The optimizer of IMCI uses simple logic. In addition to some basic execution plan optimizations, such as determining whether to use Hash Join or Nested Loop Join, the optimizer is mainly used to convert subqueries that are not supported by the IMCI executor into equivalent Join operations.

Based on the two execution engines for a row store and a column store, the optimizer provides more choices when an execution plan is selected. The optimizer can compare the cost of a row store execution plan with that of a column store execution plan and use the execution plan that has a lower cost.

In addition to the native row store-based serial execution feature of MySQL, PolarDB also supports the row store-based parallel query feature that can make full use of the multi-core computing power. Therefore, the PolarDB optimizer chooses between row store-based serial execution, row store-based parallel query, and IMCI. In the current iteration phase, the PolarDB optimizer complies with the following execution process:

1. The PolarDB optimizer parses an SQL statement and generates a logical plan. Then, the PolarDB optimizer calls the native optimizer of MySQL to perform optimizations such as join order adjustment. At the same time, the logical plan that is obtained is passed to the compilation module of IMCI execution plans. The module attempts to generate a column store execution plan. The SQL statement may be blocked by the whitelist and fall back to the row store for execution.

2. The PolarDB optimizer calculates the row-oriented execution cost based on the row store execution plan. If the cost exceeds a specific threshold, the PolarDB optimizer attempts to push the SQL statement down to the IMCI executor for execution based on an IMCI plan.

3. If the IMCI executor cannot execute the SQL statement, PolarDB attempts to compile a parallel query execution plan and execute the SQL statement. If the parallel query execution plan cannot be generated, the IMCI and parallel query features cannot execute the SQL statement, and the SQL statement falls back to the row store for execution.

The preceding strategy is based on the following judgment: In terms of execution performance, row store serial execution is inferior to row store parallel execution, which is inferior to IMCI. In terms of SQL compatibility, IMCI is inferior to row store parallel execution, which is inferior to row store serial execution. However, the actual situation is more complicated. For example, in some cases, a parallel index join based on row store indexes has a lower cost than a sort merge join based on a column store. IMCI may be selected to execute the SQL statement based on the current strategy.

The IMCI execution engine uses the classic volcano model and improves execution performance based on a column store and vectorization execution.

In terms of the volcano model, in the relational algebra that corresponds to the syntax tree generated by SQL, each operation is abstracted into an operator. The execution engine constructs the entire SQL into an operator tree. The query tree calls the Next() interface from top to bottom, and the data is pulled from bottom to top. The advantage of this method is that the computing model is simple and straightforward. This is achieved by abstracting different physical operators into iterators. Each operator only cares about its own internal logic. This reduces the coupling between operators and makes it easier to write a logically correct execution engine.

• In the IMCI execution engine, each operator also uses an iterator function to access data. The difference is that each call to the iterator returns a batch but not a row of data. Therefore, the execution engine can be regarded as a volcano model that uses a vectorization model.

  

• The capability of serial execution is limited by factors such as single-core computing efficiency, memory access latency, and I/O latency. Several key physical operators such as Scan, Join, and Agg in the IMCI executor support parallel execution. The physical operators need to support parallelism, and the IMCI optimizer also needs to support the generation of a parallel execution plan. When the optimizer determines the access mode of a table, the optimizer decides whether to enable parallel execution based on the amount of data to be accessed. If the optimizer determines to enable parallel execution, the optimizer refers to a series of state data to determine the degree of parallelism (DOP). The state data includes available CPU, memory, and I/O resources in the system, information about tasks that are scheduled and queued, statistics, query complexity, and parameters that can be configured by users. Based on these data, a recommended DOP value is calculated for the operators, and each operator internally uses the same DOP value. Users can also specify a DOP value by using hints.

  

Based on the preceding two optimization ideas, all physical execution operators are reimplemented, including Table Scan, Hash Join, Nested Loop Join, and Group By. In the following example, Hash Join is used to demonstrate the parallelization and vectorization acceleration of the executor. The following figure shows the execution process of Hash Join in IMCI.

Vectorization execution solves the problem of low single-core execution efficiency, while parallel execution breaks through the single-core computing bottleneck. The combination of the two makes the IMCI execution speed several orders of magnitude faster than the traditional row execution of MySQL.

In AP scenarios, SQL often contains many computing processes that involve one or more values, operators, or functions. This belongs to the category of expression computing. The evaluation of expressions is a computing-intensive task, so the computing efficiency of expressions is a key factor that affects overall performance.

The traditional expression computing system of MySQL uses the row-by-row operation method, which is generally called iterator model implementation. The iterator abstracts the entire table, and the expressions are implemented as a tree structure. However, this abstraction brings performance loss. This is because, during the iteration of the iterator, the acquisition of each row of data triggers multiple layers of function calls. The row-by-row acquisition of data brings too much I/O and is not friendly to the cache. MySQL uses a tree iterator model as MySQL is constrained by the access methods of storage engines. This makes it difficult to optimize the complex computing logic.

For a column store, since the data in each column is separately and sequentially stored, the expression computing in a specific column can be performed in batches. For each expression, its unit of input and output is batch. In the batch processing model, the computing process can be accelerated by using SIMD instructions.

The vectorized expression system has two key optimizations:

• The vectorized expression system makes full use of the advantages of a column store and replaces the iterator model with the batch processing model. The PolarDB team uses SIMD instructions to rewrite the expression kernel implementation of the most common data types. For example, all the basic mathematical operations (+, -, ×, /, and abs) of all numeric types including INT, DECIMAL, and DOUBLE are implemented by using corresponding SIMD instructions. With the help of the AVX512 instruction set, the performance of single-core operations is improved by several times.

  

• The expression implementation is similar to that of PostgreSQL. In SQL compilation and optimization stages, IMCI expressions are stored in a tree structure, which is similar to that of the existing row iterator model. However, before execution, the expression tree is traversed in a post-order and converted into a one-dimensional array for storage. In subsequent computing, users need to only traverse the one-dimensional array to complete the operation. The computing is more efficient because it eliminates the recursive process in the tree iterator model. In addition, this method provides a concise abstraction of the computing process and separates data from computing, which is suitable for parallel computing.

In the architecture of the plug-in storage engine framework of MySQL, the simplest solution to add column store support is to implement a separate storage engine, like a column store of Infobright and MariaDB. PolarDB uses the solution of implementing a column store as a secondary index of InnoDB. This is mainly based on the following considerations:

• InnoDB natively supports multiple indexes. INSERT, UPDATE, and DELETE operations apply to the primary index and all secondary indexes at row granularity and guarantee transactions. The solution of implementing a column store as a secondary index can reuse this transaction processing framework.

• The column store implemented as a secondary index can use the same data encoding format as other row store indexes. Memory copy is enough. Users do not need to consider information such as character sets and collations.

• A secondary index can be flexibly managed. Users can specify the columns in the index when users create a table. Users can also use DDL statements to add or delete columns in a secondary index in subsequent operations. For example, users can add INT, FLOAT, and DOUBLE columns that need to be analyzed to a columnar index. The TEXT and BLOB fields that are generally involved only in point queries but take up a lot of space can be kept in the row store.

• The crash recovery process can reuse the redo transaction log module of InnoDB to ensure seamless compatibility with existing implementations. This also facilitates the physical replication process of PolarDB. Column store indexes can be generated on independent RO nodes or standby nodes to provide analysis services.

• A secondary index has the same lifecycle as the primary index. Users can conveniently manage the secondary index.

  

As shown in the preceding figure, all the primary and secondary indexes in PolarDB are implemented as a B+ tree. A columnar index is an index by definition, but it is a virtual index used to capture the addition, deletion, and modification operations on the columns covered by the index.

For the preceding table, the primary index contains five columns (C1, C2, C3, C4, and C5) of data, and the secondary index contains two columns (C2 and C1) of data. In the common secondary index, C2 and C1 are encoded into one row and stored in the B+ tree. The column store index contains three columns (C2, C3, and C4) of data. In actual physical storage, the three columns are split and stored independently. Each column is converted into the column store format based on their writing order.

Another advantage of implementing a column store as a secondary index is that the engineering implementation of the executor is very simple. The term of covering index already exists in MySQL, which means that all columns required by a query are stored in a secondary index. The data in this secondary index can be directly used to meet the query requirements. Compared with the primary index, the use of a secondary index can greatly reduce the amount of data read and thus improve the query performance. When all columns required for a query are covered by a columnar index, the acceleration of the column store can improve query performance by dozens or even hundreds of times.

Each column in a columnar index is stored in an unordered format and uses the append write mode. Space reclamation is achieved by using label deletion and asynchronous compaction in the background. The specific implementation has the following key points:

• Records in a columnar index are organized based on row groups. Each row group contains 64,000 rows. Different columns in each row group are packaged to form a data pack.

• Each row group uses the append write mode, and the data pack that belongs to each column also uses the append write mode. For a columnar index, only an active row group is responsible for accepting new writes. When the row group is full, it freezes. All the data packs that are contained in the row group are compressed and stored on disks, and the statistics of each data block are recorded to facilitate filtering.

• Each new row written to the column-oriented row group is assigned a row ID for positioning. The system calculates the position for all columns that belong to a row by using the row ID. At the same time, the system maintains the mapping index from the primary keys to row IDs to support subsequent deletion and modification operations.

• Update operations are implemented based on label deletion. To perform delete operations, users can directly specify the bitmap. To perform update operations, users can calculate the original position based on the row ID, specify the deletion label, and then write the new data to an active row group.

• When the number of invalid records in a row group exceeds a specific threshold, asynchronous compaction is triggered in the background. On the one hand, the asynchronous compaction is used to reclaim space. On the other hand, the asynchronous compaction can make effective data storage more compact and improve the efficiency of analytical query orders.

  

This data organization method meets the requirement of batch scanning and filtering by columns during analytical queries. In addition, the impact on TP transaction operations is very small. For a write operation, users need to only append data to the memory by column. For a delete operation, users need to only specify a deletion label. An update operation involves label deletion and an append write. The column store supports transaction-level updates without reducing the OLTP performance.

Row-to-column operations are performed in the following cases:

• Case 1: DDL statements are used to create a columnar index on some columns if users have analytical requirements for an existing table. Users need to scan the data in the entire table to create a columnar index.

• Case 2: Row-to-column operations are performed on columns involved in a transaction operation.

In the case of full-table row-to-column conversion, users can use parallel scanning to scan primary keys of InnoDB and convert all the columns involved into the column store format in turn. This operation is very fast and only limited by the I/O throughput speed and available CPU resources on the server. This operation is an online DDL process and does not block the running of online services.

After a columnar index is created on a table, all update transactions simultaneously update the row-oriented and column-oriented data to ensure the transactional consistency of the row-oriented and column-oriented data. The following figure demonstrates the difference when the IMCI feature is disabled and enabled.

• When the IMCI feature is disabled, updates to all rows by the transaction are locked first. Then, the data page is modified. All locked records are unlocked at a time before the transaction is committed.

• After the IMCI feature is enabled, the transaction system creates an update cache for the column store. When all data pages are modified, the modification operations of the column store are recorded. Before the transaction ends and the locked records are unlocked, the update cache is applied to the column store system.

For general OLTP requests, the time spent in updating the memory data pages accounts for only a small part of the transaction operation process. Therefore, this method has very little impact on the latency of TP transactions. For large transactions that operate on a large number of rows, the updates to the columnar index are directly applied to the column store in real time, but the updates are not publicly visible before the transaction is committed. This ensures that the commit latency of large transactions increases within a very small time range. To further reduce the impact on TP performance, the updates to the columnar index can be asynchronously applied to the column store if AP queries do not have high requirements for data timeliness.

The column store provides the same level of transaction isolation as the row store does. For each write operation, each row in a row group records the ID of the transaction that modifies the row. For each label deletion operation in a delete bitmap, the transaction ID of the operation is also recorded. By writing and deleting transaction IDs, AP queries can obtain a globally consistent snapshot in a very lightweight manner.

As can be seen from the preceding storage format, all data packs in IMCI are unordered and use the append write mode. Therefore, it is impossible to filter out data that does not meet the requirements as accurately as normal ordered indexes of InnoDB. In IMCI, users can use statistics to filter data blocks to reduce the unit price of data access.

• When each active data pack finishes writing, the system performs computing in advance. Then, the system generates information, including the minimum value, the maximum value, the sum of values, the number of null values, and the total number of records, in the data pack. All the information is maintained in the metadata area of data packs and resides in the memory. Since data deletion operations are involved in frozen data packs, the update and maintenance of statistics are performed in the background.

• For query requests, data packs are classified into relevant, irrelevant, and possibly relevant data packs based on query conditions. This reduces actual data block access. For some aggregate query operations such as count and sum, users can perform simple operations on pre-computed statistical values. These data blocks do not even need to be decompressed.

The rough index scheme based on statistics is not very friendly for some queries that need to accurately locate some data. However, in a hybrid row-column store engine, columnar indexes need to only assist in accelerating queries that involve the scanning of a large amount of data. In this scenario, IMCI has significant advantages. For SQL queries that access only a small amount of data, the optimizer usually uses the cost model to calculate a more cost-effective solution based on a row store.

The hybrid row-column store solution of PolarDB supports both AP queries and TP queries in one cluster. However, many services have high OLTP loads, and sudden OLAP loads may interfere with the response latency of TP services. Therefore, it is necessary to support load isolation in HTAP databases. With the write-once-read-many architecture of PolarDB, users can easily isolate AP loads and TP loads. Based on the technical architecture of PolarDB, the following deployment modes are supported:

• Mode 1: The hybrid row-column store solution is enabled on read/write (RW) nodes. This mode supports lightweight AP queries. This mode is suitable when TP loads are mainly used and AP requests are relatively few. Users can also use this mode if users want to use PolarDB to query reports and the data comes from batch data import.

• Mode 2: RW nodes support OLTP loads, and an AP-type RO node is started to enable the hybrid row-column store solution to support queries. In this mode, CPU resources can be completely isolated, and all the memory of the AP-type RO node can be allocated to the column store and executor. However, since the same shared storage is used, I/O is affected.

• Mode 3: Both RW and RO nodes support OLTP loads, and the hybrid row-column store solution is enabled on a separate standby node to support AP queries. Since the standby node uses an independent shared storage cluster, this mode can also achieve the isolation of I/O resources, in addition to the isolation of CPU and memory resources that is achieved in Mode 2.

In addition to the preceding different deployment architectures that support different levels of resource isolation, in PolarDB, automatic degree of parallelism (Auto DOP) is supported in some large queries that need to be executed in parallel. This mechanism takes into account the current system loads and available CPU and memory resources, and limits the resources that can be used by a single query. This prevents a single query from consuming too many resources and affecting the processing of other requests.