PolarDB for PostgreSQL architecture
PolarDB for PostgreSQL is an enterprise-level database product from Alibaba Cloud. It uses a storage-compute decoupled architecture and is compatible with both PostgreSQL and Oracle. The storage and compute capabilities of PolarDB for PostgreSQL can be scaled horizontally. It provides enterprise-level database features such as high reliability, high availability (HA), and scalability. Additionally, PolarDB for PostgreSQL provides Massively Parallel Processing (MPP) capabilities to handle hybrid transactional and analytical processing (HTAP) workloads. It also offers innovative multi-model features for spatio-temporal data, vectors, search, and graphs to meet evolving enterprise data processing needs.
This topic describes the PolarDB for PostgreSQL architecture from the following perspectives:
Advantages of the PolarDB for PostgreSQL (Compatible with Oracle) cloud-native database
Overview of the PolarDB for PostgreSQL (Compatible with Oracle) architecture
PolarDB for PostgreSQL (Compatible with Oracle): A detailed look at the HTAP architecture
Problems with traditional databases
As business data volumes and complexity increase, traditional database systems face significant challenges:
Storage capacity is limited to a single machine.
Scaling reads with read-only instances increases costs because each instance requires its own copy of the storage.
Creating read-only instances takes longer as the data volume grows.
High replication delay.
PolarDB for PostgreSQLAdvantages of the cloud-native database
To address the challenges of traditional databases, Alibaba Cloud developed the PolarDB for PostgreSQL cloud-native database. It uses a proprietary architecture that decouples compute clusters from storage clusters. This design provides the following advantages:
Scalability: High elasticity with decoupled storage and compute.
Cost-effective: Lowers storage costs by sharing a single copy of data.
Ease of use: Supports one writer and multiple readers with transparent read/write splitting.
Reliability: Provides data redundancy with three copies and supports backups within seconds.
PolarDB for PostgreSQLOverview of the architecture
Overview of the storage-compute decoupled architecture
PolarDB for PostgreSQL uses a storage-compute decoupled design, allowing the storage and compute clusters to be scaled independently:
If compute resources are insufficient, you can scale the compute cluster separately.
If storage capacity is insufficient, you can scale the storage cluster separately.
With shared storage, the primary node and multiple read-only nodes share one copy of data. The primary node cannot flush dirty pages in the traditional way, which would cause the following problems:
A read-only node might read an outdated version of a page from storage that does not match the current state.
A read-only node might read a page that is newer than the data expected in its memory.
During a failover from the primary node to a read-only node, the pages in storage might be outdated when the read-only node takes over data updates. The read-only node must then read logs to recover the dirty pages.
The first problem requires multi-version page capabilities, and the second problem requires the primary database to control the speed of dirty page flushing.
Overview of the HTAP architecture
After implementing read/write splitting, a single compute node cannot fully utilize the high I/O bandwidth of the storage layer. It is also not possible to accelerate large queries by adding more compute resources. PolarDB for PostgreSQL introduced MPP distributed parallel execution based on shared storage to accelerate Online Analytical Processing (OLAP) queries in Online Transactional Processing (OLTP) scenarios.
PolarDB for PostgreSQL supports using a single dataset from an OLTP scenario in the following two compute engines:
Single-node execution engine: Handles high-concurrency OLTP workloads.
Distributed execution engine: Handles large-query OLAP workloads.
When using the same hardware resources, its performance is 90% of a traditional MPP data warehouse. It also provides SQL-level elasticity. If compute resources are insufficient, you can add more CPUs for OLAP analysis queries at any time without redistributing data.
PolarDB for PostgreSQL: A detailed look at the storage-compute decoupled architecture
Challenges of shared storage
With shared storage, the database architecture shifts from the traditional shared-nothing model to a shared-storage model. This shift introduces the following challenges:
Data consistency: The model changes from N compute instances + N storage instances to N compute instances + 1 storage instance.
Read/write splitting: Achieving low-latency replication with the new architecture.
High availability: Handling recovery and failover.
I/O model: Optimizing the I/O model from Buffer-IO to Direct-IO.
Architectural principles
The architectural principles of PolarDB for PostgreSQL based on shared storage are as follows:
The primary node is a read-write (RW) node, and the read-only nodes are read-only (RO).
Only the primary node can write to the shared storage layer. Therefore, the primary node and read-only nodes view consistent data on the disk.
The memory state of read-only nodes is synchronized with the primary node by replaying Write-Ahead Logging (WAL).
The primary node writes WAL logs to shared storage and replicates only the WAL metadata to read-only nodes.
Read-only nodes read the WAL from shared storage and replay it.
Data consistency
Memory state synchronization in traditional databases
In a traditional shared-nothing database, the primary and read-only nodes each have their own memory and storage. To synchronize them, WAL logs are copied from the primary node to the read-only nodes and replayed sequentially. This is the basic principle of a replicated state machine.
Memory state synchronization on shared storage
After decoupling storage and compute, the pages read from shared storage are consistent. The memory state is synchronized by reading the latest WAL from shared storage and replaying it, as shown in the following figure:
The primary node writes version 200 to shared storage by flushing dirty pages.
The read-only node replays logs based on version 100 to obtain version 200.
Past pages on shared storage
In the preceding process, if a page that was replayed from logs on a read-only node is evicted, reading that page again from storage might retrieve an old version. This is called a past page. The process is shown in the following figure:
At time T1, the primary node writes a log with LSN=200, updating the content of page P1 from 500 to 600.
At this time, the content of page P1 on the read-only node is 500.
At time T2, the primary node sends the metadata of log 200 to the read-only node. The read-only node learns that a new log exists.
At time T3, when reading page P1 on the read-only node, the node must read page P1 and the log with LSN=200. It then performs a replay to obtain the latest content of P1, which is 600.
At time T4, the read-only node evicts the latest replayed page P1 because of an insufficient BufferPool.
The primary node has not flushed the latest content of page P1 (600) to shared storage.
At time T5, a read operation for P1 is initiated again on the read-only node. Because P1 has been evicted from memory, it is read from shared storage. At this point, the content of a past page is read.
Solution for past pages
When a read-only node reads a page, it must find the corresponding base page and the starting log to replay them sequentially. The process is shown in the following figure:
Maintain the log metadata for each page in the memory of the read-only node.
When reading a page, apply logs one by one as needed until the desired page version is reached.
When applying logs, read from shared storage using the log's metadata.
Based on the preceding analysis, it is necessary to maintain an inverted index from each page to its logs. Because the memory of a read-only node is limited, this index must be persisted. PolarDB for PostgreSQL designed a persistent index structure called LogIndex, which is essentially a persistent hash data structure.
The read-only node receives WAL metadata from the primary node through the WAL receiver.
The WAL metadata records which pages were modified by the log entry.
This WAL metadata is inserted into the LogIndex, where the key is the Page ID and the value is the LSN.
A single WAL log entry might update multiple pages, such as during an index split, resulting in multiple records in the LogIndex.
Simultaneously, the page is marked as outdated in the BufferPool so that the corresponding log is replayed from the LogIndex the next time it is read.
When memory usage reaches a certain threshold, the LogIndex asynchronously flushes the in-memory hash to disk.
LogIndex solves the dependency on past pages during dirty page flushing. It also transforms the replay on read-only nodes into a lazy replay, where only the log metadata needs to be replayed.
Future pages on shared storage
In a storage-compute decoupled architecture, dirty page flushing also faces the problem of future pages. The process is shown in the following figure:
At time T1, the primary node updates P1 twice, generating two log entries. At this point, the content of page P1 on both the primary and read-only nodes is 500.
At time T2, the log with LSN=200 is sent to the read-only node.
At time T3, the read-only node replays the log with LSN=200 and obtains the content of P1 as 600. The read-only node's log replay has reached LSN 200. The subsequent log with LSN=300 is not yet available to it.
At time T4, the primary node flushes dirty pages, writing the latest content of P1 (700) to shared storage. At the same time, the read-only node's BufferPool evicts page P1.
At time T5, the read-only node reads page P1 again. Because P1 is not in the BufferPool, it reads the latest P1 from shared storage. However, because the read-only node has not replayed the log with LSN=300, it reads a future page that is ahead of its state.
The problem with future pages is that having some pages as future pages while others are normal leads to data inconsistency. For example, if an index splits into two pages and one read retrieves a normal page while the other retrieves a future page, the B+Tree index structure will be corrupted.
Solution for future pages
Future pages occur because the primary node's dirty page flushing speed exceeds the replay speed of a read-only node, even with fast lazy replay. Therefore, the solution is to control the progress of the primary node's dirty page flushing so that it does not exceed the replay position of the slowest read-only node. The process is shown in the following figure:
The read-only node has replayed logs up to position T4.
When the primary node flushes dirty pages, it sorts all dirty pages by LSN and only flushes those at or before T4. Dirty pages after T4 are not flushed.
The LSN position T4 is called the consistency point.
Low-latency replication
Problems with traditional streaming replication
Synchronization link: The log synchronization path has high I/O and a large network transmission volume.
Page replay: Reading and modifying the buffer is slow (I/O-intensive and CPU-intensive).
Data Definition Language (DDL) replay: Modifying a file requires a lock. This locking process is easily blocked, which slows down DDL operations.
Snapshot update: High concurrency on read-only nodes slows down transaction snapshot updates.
The process is as follows:
The primary node writes WAL logs to the local file system.
The WAL sender process reads and sends the logs.
The WAL receiver process on the read-only node receives and writes the logs to its local file system.
The replay process reads the WAL logs, reads the corresponding pages into the BufferPool, and replays them in memory.
The primary node flushes dirty pages to shared storage.
As this process shows, the entire link is very long. This results in high latency on read-only nodes, which affects load balancing for read/write splitting in user applications.
Optimization 1: Replicate only metadata
Because the underlying layer is shared storage, read-only nodes can directly read the required WAL data from it. Therefore, the primary node only needs to replicate the WAL log metadata (without the payload) to the read-only nodes. This reduces network traffic and I/O on the critical path. The process is shown in the following figure:
The optimized process is as follows:
A WAL record consists of a Header, PageID, and Payload.
Because read-only nodes can directly read WAL files from shared storage, the primary node sends (replicates) only the WAL metadata to the read-only nodes, including the Header and PageID.
On the read-only node, the complete WAL file is read directly from shared storage using the WAL metadata.
This optimization significantly reduces the network traffic between the primary and read-only nodes. The following figure shows that network traffic is reduced by 98%.
Optimization 2: Page replay optimization
In a traditional database, the log replay process reads many pages, applies logs one by one, and then writes them to disk. This process is on the critical path for user read I/O. With storage-compute decoupling, if a page is not in the BufferPool on a read-only node, no I/O is generated. Instead, only the LogIndex is recorded.
You can offload the following I/O operations from the replay process to the session process:
Data page I/O overhead.
Log apply overhead.
Multi-version replay of pages based on LogIndex.
As shown in the following figure, when applying the metadata of a WAL entry in the replay process on a read-only node:
If the corresponding page is not in memory, only the LogIndex is recorded.
If the corresponding page is in memory, it is marked as outdated, the LogIndex is recorded, and the replay process is complete.
When a user session process reads a page, it reads the correct page into the BufferPool and replays the corresponding logs using the LogIndex.
As shown, the main I/O operations are offloaded from a single replay process to multiple user processes.
This optimization significantly reduces replay latency, making it 30 times faster than other cloud-native databases.
Optimization 3: DDL lock replay optimization
When a DDL statement, such as drop table, is executed on the primary node, an exclusive lock must be placed on the table on all nodes. This ensures that the table file is not deleted by the primary node while being read on a read-only node, because there is only one copy of the file on shared storage. The exclusive lock is placed on the table on all read-only nodes by replicating the DDL lock through WAL, where it is then replayed. When the replay process replays the DDL lock, locking the table can be blocked for a long time. Therefore, you can optimize the critical path of the replay process by offloading the DDL lock to other processes.
This optimization ensures that the replay process remains smooth and is not blocked by waiting for DDL operations, which would obstruct the critical replay path.
After these three optimizations, the replication delay is greatly reduced, providing the following advantages:
Read/write splitting: Better load balancing, closer to the Oracle RAC user experience.
High availability: Accelerates the HA process.
Stability: Minimizes the number of future pages, allowing for fewer or no page snapshot writes.
Recovery optimization
Background
Database recovery from scenarios such as out-of-memory (OOM) errors or crashes takes a long time. This is primarily due to slow log replay, a problem that is more prominent in the Direct-IO model with shared storage.
Lazy Recovery
As described earlier, LogIndex enables lazy replay on read-only nodes. The recovery process after a primary node restarts is also essentially replaying logs. Therefore, lazy replay can be used to accelerate the recovery process:
Read WAL logs sequentially starting from the checkpoint.
After the LogIndex logs are replayed, the replay is considered complete.
Recovery is complete, and the service becomes available.
The actual replay is offloaded to the session processes that connect after the restart.
After optimization (replaying 500 MB of logs), the result is shown in the following figure:
Persistent BufferPool
The preceding solution optimizes the restart speed during recovery. However, after the restart, session processes must read WAL logs to replay the desired pages. This means there will be a brief period of slow response after recovery. An optimization is to not destroy the BufferPool when the database restarts. As shown in the following figure, the BufferPool is not destroyed during a crash and restart.
The shared memory in the kernel is divided into two parts:
Global structures, such as ProcArray.
The BufferPool structure. The BufferPool is allocated using named shared memory and remains valid after a process restart. Global structures need to be re-initialized after a process restart.
However, not all pages in the BufferPool can be reused. For example, if a process places an X lock on a page and then crashes, there is no process to release this X lock. Therefore, after a crash and restart, all pages in the BufferPool must be traversed to remove those that cannot be reused. Additionally, the recycling of the BufferPool depends on Kubernetes. This optimization ensures stable performance before and after a restart.
PolarDB for PostgreSQL A detailed look at the HTAP architecture
After implementing read/write splitting in PolarDB for PostgreSQL, the underlying storage pool theoretically provides unlimited I/O throughput. However, a single compute node has limited resources and cannot fully utilize the high I/O bandwidth of the storage layer or accelerate large queries by adding more compute resources. To address this, PolarDB for PostgreSQL introduced MPP distributed parallel execution based on shared storage to accelerate OLAP queries in OLTP scenarios.
HTAP architectural principles
In PolarDB for PostgreSQL, the underlying storage is shared across different nodes. Therefore, it cannot scan tables directly like a traditional MPP system. PolarDB for PostgreSQL added support for MPP distributed parallel execution to its original single-node execution engine and optimized it for shared storage. MPP based on shared storage is an industry first. The principles are as follows:
The Shuffle operator hides the data distribution.
The ParallelScan operator abstracts the shared storage.
As shown in the figure:
Table A and Table B are joined and then aggregated.
The table in shared storage remains a single table and is not physically partitioned.
Four types of scan operators were redesigned to scan the table on shared storage in chunks, forming a virtual partition.
Distributed optimizer
Based on the community's GPORCA optimizer, we extended it with Transformation Rules that are aware of shared storage characteristics. This allows the optimizer to explore the unique plan space available with shared storage. For example, a table in PolarDB for PostgreSQL can be scanned either fully or by region, which is a fundamental difference from traditional MPP. As shown in the following figure, the gray part at the top is the adaptation layer between the PolarDB for PostgreSQL kernel and the GPORCA optimizer. The bottom part is the ORCA kernel, and the gray module within it represents the extensions made for shared storage features.
Operator parallelization
In PolarDB for PostgreSQL, four types of operators need to be parallelized. The following describes the parallelization of a representative operator, SeqScan.
To maximize the use of the storage's high I/O bandwidth, sequential scans are logically divided into 4 MB chunks. This spreads the I/O across different disks as much as possible, allowing all disks to provide read services simultaneously. Another advantage is that each read-only node scans only a portion of the table file. The total size of the table that can be cached is the sum of the BufferPools of all read-only nodes.
In the chart below:
Adding read-only nodes linearly improves scan performance by 30 times.
When the buffer is enabled, the scan time drops from 37 minutes to 3.75 seconds.
Eliminating data skew
Skew is an inherent problem in traditional MPP systems:
In PolarDB for PostgreSQL, large objects are associated with a TOAST table through a heap table. Splitting either table cannot achieve balance.
In addition, there is jitter in transactions, buffers, network, and I/O load across different read-only nodes.
These two factors can lead to long-tail processes during distributed execution.
The coordinator node is internally divided into a DataThread and a ControlThread.
The DataThread is responsible for collecting and summarizing tuples.
The ControlThread is responsible for controlling the scan progress of each scan operator.
Faster-working processes can scan more logical data segments.
Buffer affinity needs to be considered during the process.
Although the allocation is dynamic, buffer affinity should be maintained. In addition, the context of each operator is stored in the worker's private memory. The coordinator does not store specific table information.
The table below shows that when large objects are present, static chunking causes data skew, while dynamic scanning still achieves linear improvement.
SQL-level scalability
The data sharing feature also supports the high elasticity required by cloud-native environments. External dependencies for various modules in the coordinator's chain are stored in shared storage. At the same time, runtime parameters needed by workers are synchronized from the coordinator through a control link. This design makes the coordinator and workers stateless.
Therefore:
Any read-only node that a SQL connection is made to can become a coordinator node. This solves the single-point-of-failure problem for the coordinator.
It supports different SQL statements using different numbers of CPUs for execution. This allows for flexible configuration of CPU cores for different business SQL queries.
Transactional consistency
Data consistency across multiple compute nodes is achieved through a wait-for-replay mechanism and a global snapshot mechanism. The wait-for-replay mechanism ensures that all workers can see the required data versions, while the global snapshot mechanism ensures that a unified version is selected.
TPC-H performance: Acceleration ratio
We conducted tests using a 1 TB TPC-H benchmark. First, we compared the performance of the new distributed parallel execution in PolarDB for PostgreSQL with single-node parallel execution. As a result, three SQL queries were accelerated by 60 times, and 19 SQL queries were accelerated by more than 10 times.
In addition, we tested the performance of the distributed execution engine as we added more CPUs. As you can see, performance scales linearly from 16 to 128 cores. When looking at the 22 SQL queries individually, the performance of each query improves linearly as more CPUs are added.
TPC-H performance: Comparison with traditional MPP data warehouses
Compared to a traditional MPP data warehouse, when using the same 16 nodes, the performance of PolarDB for PostgreSQL is 90% of the traditional MPP data warehouse.
As mentioned earlier, the distributed engine of PolarDB for PostgreSQL is scalable. Data does not need to be fully redistributed. When the degree of parallelism (DOP) is 8, the performance is 5.6 times that of a traditional MPP data warehouse.
Accelerating index building with distributed execution
OLTP applications create many indexes. Analysis shows that 80% of the index building process is spent on sorting and constructing index pages, and 20% is spent on writing index pages. We use distributed parallelism to accelerate the sorting process and use pipelined batch writing.
This optimization can improve index creation speed by 4 to 5 times.
Accelerating multi-model databases with distributed parallel execution: Spatio-temporal database
PolarDB for PostgreSQL is a multi-model database that supports spatio-temporal data. Spatio-temporal databases are both compute-intensive and I/O-intensive and can be accelerated using distributed execution. PolarDB for PostgreSQL introduced the feature of scanning shared RTREE indexes for shared storage.
Data volume: 400 million records, 500 GB.
Specifications: 5 read-only nodes, each with 16 CPU cores and 128 GB of memory.
Performance:
Scales linearly with the number of CPUs.
With 80 CPU cores, performance improves by 71 times.
