Multi-node analysis

Updated at:

This topic introduces the multi-node parallel execution feature for columnar tables in PolarDB for MySQL. It covers the technical architecture, use cases, best practices, and performance test results.

Note

If you have questions about the multi-node parallel execution feature for columnar tables, you can get support by joining the DingTalk group. In the group, you can directly @ a specialist with your questions.

DingTalk group number: 24490017825

Background

Columnar tables are the OLAP solution for PolarDB. As data volume, query complexity, and the need to query external tables such as those on OSS increase, a single read-only node can no longer meet the performance demands of large-scale data. To address this, columnar table clusters provide multi-node parallel execution and resource scaling capabilities.

Technical architecture

The multi-node parallel execution feature for columnar tables uses a group of read-only nodes to run queries in parallel. As your query workload changes, you can quickly add or remove read-only nodes to balance query performance and computing costs.

image

Requirements

Your PolarDB for MySQL cluster must meet the following requirements:

  • product version: Enterprise Edition.

  • kernel version: MySQL 8.0.2, with a minor kernel version of 8.0.2.2.34 or later.

Use cases

  • Use the resource scaling of multi-node parallel execution to increase CPU and IOPS, reducing query latency.

  • Improve data caching capacity by having each node process only a subset of the data.

Best practices

When you run analytical queries on large datasets, the primary goals are to minimize data shuffle and reduce disk I/O. The columnar tables in PolarDB use partition keys and sort keys to achieve superior query performance.

Partition key

A partition key distributes data across multiple nodes. The core idea is to keep related data together for local computation.

How it works

For partitioned tables in PolarDB that use HASH or KEY partitioning for primary or secondary partitions, the multi-node execution environment for columnar tables uses a Share-Nothing architecture. This means each partition is processed by a single, dedicated node.

Core advantages

  • Improved data caching: Each node processes only its assigned partitions, allowing it to use local memory more effectively for data caching.

  • Optimized query performance: For JOIN operations and GROUP BY queries based on the partition key, data is processed locally on each node. This significantly reduces data shuffle.

Best practices

  • Choose core business association keys: Use the columns most frequently used in JOIN or GROUP BY clauses as the partition key, such as order_id or user_id.

  • Keep the number of partitions consistent: The tables involved in a JOIN operation must have the same number of HASH/KEY partitions. Otherwise, local computation cannot be achieved.

  • Use a large prime number for the partition count: Choose a sufficiently large prime number, such as 97 or 199, as the number of partitions. This helps ensure that data is distributed evenly across nodes and prevents data skew.

Sort key

A sort key organizes data within each physical partition. The core idea is to skip reading irrelevant data. For large datasets, you can filter data by using RANGE partitioning or by adding a sort key to a columnar table.

How it works

In columnar storage, data is organized into multiple data blocks. When you set a sort key, the data within each data block is physically sorted based on the specified column. When a query is run, the database uses metadata in the header of each data block, such as minimum and maximum values, to decide whether to read the block.

Core advantages

  • Efficient data pruning: When a WHERE clause contains a filter on the sort key, the query engine can directly skip (or prune) entire data blocks whose data ranges do not match the filter conditions, reducing the required disk I/O.

Best practices

  • Select frequently filtered columns: Use columns that are frequently filtered in WHERE clauses as the sort key. This is especially effective for columns used in range queries (>, <, BETWEEN) or high-cardinality equality queries.

  • Combine with RANGE partitioning: Combine sort keys with RANGE partitioning to achieve multi-level filtering.

Performance test results

The following results are from a multi-node test on a TPC-H 1 TB dataset using only columnar tables. The test was run on a three-node MPP cluster, with each node configured with 32 cores and 256 GB of memory.

Note

The TPC-H implementation in this topic is based on the TPC-H benchmark but does not fully comply with all TPC-H requirements. Therefore, the results cannot be compared with published TPC-H benchmark results.

Query

1 RO (seconds)

2 RO (seconds)

3 RO (seconds)

4 RO (seconds)

Q1

20.360

10.708

7.580

3.298

Q2

0.835

0.575

0.468

0.369

Q3

11.608

3.012

1.882

1.529

Q4

11.529

2.077

1.490

1.214

Q5

17.276

7.908

2.721

2.229

Q6

0.556

0.323

0.244

0.206

Q7

8.296

2.358

1.648

1.288

Q8

22.813

10.577

6.281

2.362

Q9

80.469

26.910

16.874

13.185

Q10

25.870

4.008

2.699

2.113

Q11

1.857

1.132

0.893

0.791

Q12

4.936

2.702

1.899

1.543

Q13

24.795

11.241

7.083

5.156

Q14

4.291

1.081

0.802

0.645

Q15

4.195

2.538

1.512

1.238

Q16

5.496

2.642

1.866

1.430

Q17

12.762

5.208

1.704

1.396

Q18

96.524

16.726

11.108

9.017

Q19

15.707

7.953

2.957

2.454

Q20

76.146

2.229

1.775

1.288

Q21

35.898

13.211

4.957

3.826

Q22

6.494

3.095

2.207

1.925

Total of 22 queries

488.713

138.214

80.650

58.502

FAQ

Determine if a SQL query uses MPP

The multi-node parallel execution (MPP) feature for PolarDB columnar tables can accelerate complex queries. You can determine if a query uses this feature by analyzing its EXPLAIN execution plan. The key indicator is the presence of an Exchange operator.

  1. Step 1: Force an MPP plan

    First, use a specific hint to confirm if the optimizer can rewrite your SQL statement into an MPP plan.

    Procedure
    Add EXPLAIN before your SQL statement and insert the /*+ SET_VAR(imci_plan_use_mpp=forced) */ hint. This hint forces the optimizer to attempt to generate an MPP execution plan.

    Example

    EXPLAIN SELECT /*+ SET_VAR(imci_plan_use_mpp=forced) */ COUNT(*) FROM nation;

    Interpreting the result
    After you run the command, if the returned execution plan includes an Exchange operator, this indicates that your SQL query can use multi-node parallel execution.

  2. Step 2: Check the actual execution plan

    After confirming the potential, check if the optimizer chooses to execute your SQL in MPP mode under normal conditions (without hints).

    Procedure
    Run EXPLAIN directly on your original SQL statement.

    Example

    EXPLAIN SELECT COUNT(*) FROM nation;

    Interpreting the result

    • If the execution plan includes an Exchange operator: This indicates that the optimizer, after a cost-based evaluation, chose MPP as the most efficient execution method for your query.

    • If the execution plan does not include an Exchange operator: This might mean one of the following:

      • The query is too simple or involves too little data, so the optimizer determined that single-node execution is faster.

      • The structure of the SQL query or the table, such as the partition key settings, prevents the use of MPP. You can return to Step 1 to check and optimize your SQL query or table design.