Hologres:EXPLAIN and EXPLAIN ANALYZE

Introduction to execution plans

In Hologres, the query optimizer (QO) generates an execution plan for each SQL statement. The query engine (QE) generates a final execution plan based on the execution plan generated by the QO. The QE then executes the SQL statement and returns a result. An execution plan contains information such as SQL statistics, operators, and operator execution durations. A good execution plan helps you execute an SQL statement with fewer resources and obtain results faster. Execution plans are vital to routine data development. They help you identify issues in SQL statements and optimize SQL statements.

Hologres is compatible with PostgreSQL. You can query execution plans of SQL statements by executing the EXPLAIN and EXPLAIN ANALYZE statements.

• EXPLAIN: This statement returns the SQL execution plan that is estimated by the QO based on the characteristics of SQL statements, rather than the actual execution plan. This estimated execution plan provides reference for SQL statement execution.

• EXPLAIN ANALYZE: This statement returns the actual SQL execution plan. Compared with the estimated execution plan returned by the EXPLAIN statement, the actual execution plan returned by the EXPLAIN ANALYZE statement contains actual execution information such as the actually executed operators and the accurate execution duration of each operator. Based on the execution duration of each operator, you can perform SQL statement optimization.

EXPLAIN

• Syntax

  You can execute the following EXPLAIN statement to query the execution plan that is estimated by the QO:

  EXPLAIN <sql>;

• Example

  In this example, a TPC-H query statement is used.

  Note

  The return value in this example cannot be used as the result published by TPC-H.

  EXPLAIN SELECT
          l_returnflag,
          l_linestatus,
          sum(l_quantity) AS sum_qty,
          sum(l_extendedprice) AS sum_base_price,
          sum(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
          sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
          avg(l_quantity) AS avg_qty,
          avg(l_extendedprice) AS avg_price,
          avg(l_discount) AS avg_disc,
          count(*) AS count_order
  FROM
          lineitem
  WHERE
          l_shipdate <= date '1998-12-01' - interval '120' day
  GROUP BY
          l_returnflag,
          l_linestatus
  ORDER BY
          l_returnflag,
          l_linestatus;

• Returned result

  QUERY PLAN
  Sort  (cost=0.00..7795.30 rows=3 width=80)
    Sort Key: l_returnflag, l_linestatus
    ->  Gather  (cost=0.00..7795.27 rows=3 width=80)
          ->  Project  (cost=0.00..7795.27 rows=3 width=80)
                ->  Project  (cost=0.00..7794.27 rows=3 width=104)
                      ->  Final HashAggregate  (cost=0.00..7793.27 rows=3 width=76)
                            Group Key: l_returnflag, l_linestatus
                            ->  Redistribution  (cost=0.00..7792.95 rows=1881 width=76)
                                  Hash Key: l_returnflag, l_linestatus
                                  ->  Partial HashAggregate  (cost=0.00..7792.89 rows=1881 width=76)
                                        Group Key: l_returnflag, l_linestatus
                                        ->  Local Gather  (cost=0.00..7791.81 rows=44412 width=76)
                                              ->  Decode  (cost=0.00..7791.80 rows=44412 width=76)
                                                    ->  Partial HashAggregate  (cost=0.00..7791.70 rows=44412 width=76)
                                                          Group Key: l_returnflag, l_linestatus
                                                          ->  Project  (cost=0.00..3550.73 rows=584421302 width=33)
                                                                ->  Project  (cost=0.00..2585.43 rows=584421302 width=33)
                                                                      ->  Index Scan using Clustering_index on lineitem  (cost=0.00..261.36 rows=584421302 width=25)
                                                                            Segment Filter: (l_shipdate <= '1998-08-03 00:00:00+08'::timestamp with time zone)
                                                                            Cluster Filter: (l_shipdate <= '1998-08-03 00:00:00+08'::timestamp with time zone)
  

• Result description

  You need to view the execution plan from bottom to top. Each arrow (->) indicates a node. Each node contains information such as the operator and the estimated number of returned rows. The following table describes the parameters of an operator.

  Parameter	Description
  cost	The estimated execution duration of an operator. The cost value of a parent node includes the cost values of its child nodes. The cost parameter contains the estimated start cost and estimated total cost that are separated by two periods (..). Estimated start cost: the cost before the output phase starts. Estimated total cost: the estimated total cost of the operator execution. In the preceding returned result, the estimated start cost of the Final HashAggregate node is 0.00, and the estimated total cost of the node is 7793.27.
  rows	The number of rows that are returned by an operator, which is estimated based on table statistics. The value of the rows parameter for the Index Scan node is 1000 by default. Note In most cases, if rows=1000 is returned, the statistics of the table are invalid, and no estimation is performed based on the table statistics. You can execute the analyze <tablename> statement to update the statistics of the table.
  width	The estimated average length of columns returned by an operator. A large value indicates a long column length. Unit: bytes.

EXPLAIN ANALYZE

• Syntax

  You can execute the following EXPLAIN ANALYZE statement to query the actual execution plan of an SQL statement and the execution duration of each operator. This helps you diagnose SQL performance issues.

  EXPLAIN ANALYZE <sql>;

• Example

  In this example, a TPC-H query statement is used.

  EXPLAIN ANALYZE SELECT
          l_returnflag,
          l_linestatus,
          sum(l_quantity) AS sum_qty,
          sum(l_extendedprice) AS sum_base_price,
          sum(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
          sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
          avg(l_quantity) AS avg_qty,
          avg(l_extendedprice) AS avg_price,
          avg(l_discount) AS avg_disc,
          count(*) AS count_order
  FROM
          lineitem
  WHERE
          l_shipdate <= date '1998-12-01' - interval '120' day
  GROUP BY
          l_returnflag,
          l_linestatus
  ORDER BY
          l_returnflag,
          l_linestatus;

• Returned result

  QUERY PLAN
  Sort  (cost=0.00..7795.30 rows=3 width=80)
    Sort Key: l_returnflag, l_linestatus
  [id=21 dop=1 time=2427/2427/2427ms rows=4(4/4/4) mem=3/3/3KB open=2427/2427/2427ms get_next=0/0/0ms]
    ->  Gather  (cost=0.00..7795.27 rows=3 width=80)
        [20:1 id=100003 dop=1 time=2426/2426/2426ms rows=4(4/4/4) mem=1/1/1KB open=0/0/0ms get_next=2426/2426/2426ms]
          ->  Project  (cost=0.00..7795.27 rows=3 width=80)
              [id=19 dop=20 time=2427/2426/2425ms rows=4(1/0/0) mem=87/87/87KB open=2427/2425/2425ms get_next=1/0/0ms]
                ->  Project  (cost=0.00..7794.27 rows=0 width=104)
                      ->  Final HashAggregate  (cost=0.00..7793.27 rows=3 width=76)
                            Group Key: l_returnflag, l_linestatus
                          [id=16 dop=20 time=2427/2425/2424ms rows=4(1/0/0) mem=574/570/569KB open=2427/2425/2424ms get_next=1/0/0ms]
                            ->  Redistribution  (cost=0.00..7792.95 rows=1881 width=76)
                                  Hash Key: l_returnflag, l_linestatus
                                [20:20 id=100002 dop=20 time=2427/2424/2423ms rows=80(20/4/0) mem=3528/1172/584B open=1/0/0ms get_next=2426/2424/2423ms]
                                  ->  Partial HashAggregate  (cost=0.00..7792.89 rows=1881 width=76)
                                        Group Key: l_returnflag, l_linestatus
                                      [id=12 dop=20 time=2428/2357/2256ms rows=80(4/4/4) mem=574/574/574KB open=2428/2357/2256ms get_next=1/0/0ms]
                                        ->  Local Gather  (cost=0.00..7791.81 rows=44412 width=76)
                                            [id=11 dop=20 time=2427/2356/2255ms rows=936(52/46/44) mem=7/6/6KB open=0/0/0ms get_next=2427/2356/2255ms pull_dop=9/9/9]
                                              ->  Decode  (cost=0.00..7791.80 rows=44412 width=76)
                                                  [id=8 dop=234 time=2435/1484/5ms rows=936(4/4/4) mem=0/0/0B open=2435/1484/5ms get_next=4/0/0ms]
                                                    ->  Partial HashAggregate  (cost=0.00..7791.70 rows=44412 width=76)
                                                          Group Key: l_returnflag, l_linestatus
                                                        [id=5 dop=234 time=2435/1484/3ms rows=936(4/4/4) mem=313/312/168KB open=2435/1484/3ms get_next=0/0/0ms]
                                                          ->  Project  (cost=0.00..3550.73 rows=584421302 width=33)
                                                              [id=4 dop=234 time=2145/1281/2ms rows=585075720(4222846/2500323/3500) mem=142/141/69KB open=10/1/0ms get_next=2145/1280/2ms]
                                                                ->  Project  (cost=0.00..2585.43 rows=584421302 width=33)
                                                                    [id=3 dop=234 time=582/322/2ms rows=585075720(4222846/2500323/3500) mem=142/142/69KB open=10/1/0ms get_next=582/320/2ms]
                                                                      ->  Index Scan using Clustering_index on lineitem  (cost=0.00..261.36 rows=584421302 width=25)
                                                                            Segment Filter: (l_shipdate <= '1998-08-03 00:00:00+08'::timestamp with time zone)
                                                                            Cluster Filter: (l_shipdate <= '1998-08-03 00:00:00+08'::timestamp with time zone)
                                                                          [id=2 dop=234 time=259/125/1ms rows=585075720(4222846/2500323/3500) mem=1418/886/81KB open=10/1/0ms get_next=253/124/0ms]
  
  ADVICE: 
  [node id : 1000xxx] distribution key miss match!  table lineitem defined distribution keys : l_orderkey; request distribution columns : l_returnflag, l_linestatus; 
  shuffle data skew in different shards!  max rows is 20, min rows is 0
  
  Query id:[300200511xxxx]
  ======================cost======================
  Total cost:[2505] ms
  Optimizer cost:[47] ms
  Init gangs cost:[4] ms
  Build gang desc table cost:[2] ms
  Start query cost:[18] ms
  - Wait schema cost:[0] ms
  - Lock query cost:[0] ms
  - Create dataset reader cost:[0] ms
  - Create split reader cost:[0] ms
  Get the first block cost:[2434] ms
  Get result cost:[2434] ms
  ====================resource====================
  Memory: 921(244/230/217) MB,  straggler worker id: 72969760xxx
  CPU time: 149772(38159/37443/36736) ms, straggler worker id: 72969760xxx
  Physical read bytes: 3345(839/836/834) MB, straggler worker id: 72969760xxx
  Read bytes: 41787(10451/10446/10444) MB, straggler worker id: 72969760xxx
  DAG instance count: 41(11/10/10), straggler worker id: 72969760xxx
  Fragment instance count: 275(70/68/67), straggler worker id: 72969760xxx

• Result description

  The output of the EXPLAIN ANALYZE statement provides the actual execution path of an SQL statement. The output is displayed in a tree structure and contains detailed execution information of each operator at each phase. The output of the EXPLAIN ANALYZE statement contains the following parts: query plan, advice, cost, and resource.

Query Plan

The query plan part provides detailed execution information of each operator. Similar to the EXPLAIN statement, you also need to view the query plan from bottom to top. Each arrow (->) indicates a node.

An SQL statement may involve multiple operators. For more information about operators, see Operators.

Advice

The advice part provides optimization suggestions that are automatically generated by the system based on the output of the EXPLAIN ANALYZE statement. The following suggestions are available:

• Table xxx misses bitmap index: We recommend that you configure a distribution key, a clustering key, or a bitmap index for the table.

• Table xxx Miss Stats! please run 'analyze xxx';: No statistics are available for the table.

• shuffle data xxx in different shards! max rows is 20, min rows is 0: Data screw may exist.

Cost

The cost part provides the total execution duration of an SQL statement and the execution duration of each phase. You can locate performance bottlenecks based on the execution duration of each phase.

Total cost: the total execution duration of an SQL statement in the unit of milliseconds. It consists of the following parts:

• Optimizer cost: the duration that is taken by the QO to generate an execution plan. Unit: milliseconds.

• Build gang desc table cost: the duration that is taken to convert the execution plan that is generated by the QO into the data that is in the format required by the QE. Unit: milliseconds.

• Init gangs cost: the duration that is taken to preprocess the execution plan that is generated by the QO and send the query request to the QE to trigger the start query phase. Unit: milliseconds.

• Start query cost: the duration of the initialization phase, which starts when the Init gangs step completes and ends when the query operation starts. The initialization phase involves operations such as lock acquiring and schema version alignment and consists of the following parts:

  • Wait schema cost: the duration that is taken to align the storage engine (SE) version and FE node version with the schema version. If the table schema changes, both the FE node version and SE version need to be updated. If the FE node version and SE version are inconsistent with the schema version, schema latency may occur. If a large number of DDL statements are executed on the partitioned parent table, the processing speed of SE is slow and the latency is high. As a result, data write and read are slow. In this case, you can optimize the frequency of executing DDL statements.

  • Lock query cost: the duration that is taken for a query to acquire a lock. If the duration is long, the query is waiting for a lock.

  • Create dataset reader cost: the duration that is taken to create an index data reader. If the duration is long, the cache may not be hit.

  • Create split reader cost: the duration that is taken to open a file. If the duration is long, the metadata of the file does not hit the cache. If this occurs, the I/O overhead is high.

• Get result cost: the duration from the time when the start query phase ends to the time when all results are returned. The Get result cost contains the Get the first block cost. Unit: milliseconds.

  • Get the first block cost: the duration from the time when the start query phase ends to the time when the first record batch is returned. In some scenarios, the Get the first block cost is very close to or consistent with the Get result cost. For example, if a hash aggregate operator is used in the first step of a query plan, full data of downstream operators is required to create a hash table for an aggregate operation. For common queries with filter conditions, data is calculated and returned in real time. In this scenario, the Get the first block cost greatly differs from the Get result cost. The difference depends on the data amount.

Resource

The resource part provides the amounts of resources that are consumed during the query execution, in the format of total amount of consumed resources(maximum amount of consumed resources/average amount of consumed resources/minimum amount of consumed resources).

Hologres is a distributed engine. A Hologres instance has multiple worker nodes. The computing results of worker nodes are merged and the merging result is returned to the client. Therefore, the resource consumption information is displayed in the format of total(max worker/avg worker/min worker).

• total: the total amount of resources that are consumed by a SQL query statement.

• max: the maximum amount of resources that are consumed by a worker node.

• avg: the average amount of resources consumed per worker node, which is calculated by using the following formula: Total amount of consumed resources/Number of worker nodes.

• min: the minimum amount of resources that are consumed by a worker node.

The following table describes the metrics in the resource part.

Operators

SCAN

• seq scan

  Seq Scan is used to read data from a table in sequence. A full table scan is performed. The on keyword after Seq Scan is followed by the name of the scanned table.

  Example: Execute the following statement to check the execution plan of a query on a common internal table. Seq Scan is returned in the execution plan of the query statement.

  EXPLAIN SELECT * FROM public.holo_lineitem_100g;

  The following figure shows the returned result.

  

  • Query data from a partitioned table

    If you query data from a partitioned table, the Seq Scan on Partitioned Table keywords are displayed in the execution plan of the query. You can obtain the number of partitions that are scanned in the query statement from the Partitions selected keywords.

    Example: Execute the following statement to check the execution plan of a query on a partitioned parent table with one partition scanned.

    EXPLAIN SELECT * FROM public.hologres_parent;

    The following figure shows the returned result.

    

  • Query data by using a foreign table

    If you query data by using a foreign table, the Foreign Table Type keywords are displayed in the execution plan of the query to specify the source of the foreign table. The value of Foreign Table Type can be MaxCompute, OSS, or Hologres.

    Example: Execute the following statement to check the execution plan of a query by using a MaxCompute foreign table.

    EXPLAIN SELECT * FROM public.odps_lineitem_100;

    The following figure shows the returned result.

    

• Index Scan and Index Seek

  You can use indexes to accelerate queries on a table. Hologres uses different indexes at the underlying layer based on the storage mode of the table. Based on the column-oriented storage mode and row-oriented storage mode, indexes are classified into clustering_index and Index Seek (also named pk_index).

  • Clustering_index: the type of indexes that are applicable to column-oriented tables, such as the segment key or clustering key. If a query on a column-oriented table hits indexes, this type of indexes are used. Seq Scan Using Clustering_index is usually used with Filter. Filter is a child node and lists the hit indexes. Filter can be a clustering filter, segment filter, or bitmap filter. For more information, see Column-oriented storage.

    • Example 1: A query hits indexes.

      BEGIN;
      CREATE TABLE column_test (
       "id" bigint not null ,
       "name" text not null ,
       "age" bigint not null 
      );
      CALL set_table_property('column_test', 'orientation', 'column');
      CALL set_table_property('column_test', 'distribution_key', 'id');
      CALL set_table_property('column_test', 'clustering_key', 'id');
      COMMIT;
      
      INSERT INTO column_test VALUES(1,'tom',10),(2,'tony',11),(3,'tony',12);
      
      EXPLAIN SELECT * FROM column_test WHERE id>2;

      The following figure shows the returned result.

      

    • Example 2: A query does not hit any index, and indexes of the clustering_index type are not used.

      EXPLAIN SELECT * FROM column_test WHERE age>10;

      The following figure shows the returned result.

      

  • Index Seek (also named pk_index): the type of indexes that are applicable to row-oriented tables, such as the primary key. In most cases, fixed plans are used for point queries on row-oriented tables that are configured with primary keys. If queries on row-oriented tables that are configured with primary keys do not use fixes plans, indexes of the Index Seek type are used. For more information, see Row-oriented storage.

    Example: Query data from a row-oriented table.

    BEGIN;
    CREATE TABLE row_test_1 (
        id bigint not null,
        name text not null,
        class text ,
    PRIMARY KEY (id)
    );
    CALL set_table_property('row_test_1', 'orientation', 'row');
    CALL set_table_property('row_test_1', 'clustering_key', 'name');
    COMMIT;
    INSERT INTO row_test_1 VALUES ('1','qqq','3'),('2','aaa','4'),('3','zzz','5');
    
    BEGIN;
    CREATE TABLE row_test_2 (
        id bigint not null,
        name text not null,
        class text ,
    PRIMARY KEY (id)
    );
    CALL set_table_property('row_test_2', 'orientation', 'row');
    CALL set_table_property('row_test_2', 'clustering_key', 'name');
    COMMIT;
    INSERT INTO row_test_2 VALUES ('1','qqq','3'),('2','aaa','4'),('3','zzz','5');
    
    --pk_index
    EXPLAIN SELECT * FROM (SELECT id FROM row_test_1 WHERE id = 1) t1 JOIN row_test_2 t2 ON t1.id = t2.id;
    

    The following figure shows the returned result.

    

Filter

Filter is used to filter data based on SQL conditions. In most cases, Filter is a child node of Seq Scan and is executed with the Seq Scan node. Filter specifies whether data is filtered and whether a filter condition hits indexes. This section describes various types of Filters.

• Filter

  If an execution plan contains only the Filter keyword, the filter condition does not hit any index. In this case, you need to check the table indexes and configure indexes again to accelerate the SQL query.

  Note

  If an execution plan contains One-Time Filter: false, the output is empty.

  Example:

  BEGIN;
  CREATE TABLE clustering_index_test (
   "id" bigint not null ,
   "name" text not null ,
   "age" bigint not null 
  );
  CALL set_table_property('clustering_index_test', 'orientation', 'column');
  CALL set_table_property('clustering_index_test', 'distribution_key', 'id');
  CALL set_table_property('clustering_index_test', 'clustering_key', 'age');
  COMMIT;
  
  INSERT INTO clustering_index_test VALUES (1,'tom',10),(2,'tony',11),(3,'tony',12);
  
  EXPLAIN SELECT * FROM clustering_index_test WHERE id>2;

  The following figure shows the returned result.

  

• Segment Filter

  Segment Filter indicates that a query hits the segment key. Segment Filter is used with index_scan. For more information, see Event time column (segment key).

• Cluster Filter

  Cluster Filter indicates that a query hits the clustering key. For more information, see Clustering key.

• Bitmap Filter

  Bitmap Filter indicates that a query hits the bitmap index. For more information, see Bitmap index.

• Join Filter

  Join Filter indicates that data needs to be filtered after the join operation.

Decode

Decode is used to encode or decode data to accelerate the computing of text data.

Local Gather and Gather

In Hologres, data is stored as files in shards. Local Gather is used to merge data in multiple files into one shard. Gather is used to summarize data in multiple shards and return the result.

Example:

EXPLAIN SELECT * FROM public.lineitem;

The following figure shows the returned execution plan. Data is scanned, merged into one shard by using the Local Gather operator, and then summarized by using the Gather operator.

Redistribution

Redistribution is used to hash data or randomly distribute data to one or more shards.

• The redistribution operator is commonly used in the following scenarios:

  • JOIN, COUNT DISTINCT (a JOIN method), and GROUP BY clauses: If no distribution key is configured for the table or the distribution key setting is invalid, data is shuffled among multiple shards during the query. If redistribution occurs when you join multiple tables, the local join feature is not used. As a result, the query performance is poor.

  • Keys such as the join key or group by key that are involved in expressions: If you use a CAST function to convert data types, the local join feature is not used. In this case, redistribution occurs.

• Examples:

  • Example 1: Join two tables. The distribution key is invalid and redistribution occurs.

    BEGIN;
    CREATE TABLE tbl1(
    a int not null,
    b text not null
    );
    CALL set_table_property('tbl1', 'distribution_key', 'a');
    CREATE TABLE tbl2(
    c int not null,
    d text not null
    );
    CALL set_table_property('tbl2', 'distribution_key', 'd');
    COMMIT;
    
    EXPLAIN SELECT * FROM tbl1  JOIN tbl2 ON tbl1.a=tbl2.c;

    The following figure shows the returned execution plan. The redistribution operator is contained, indicating that the distribution key is invalid. In the SQL statement, tbl1.a=tbl2.c is configured to join tables. However, the distribution key of the tbl1 table is a and that of the tbl2 table is d. As a result, data is shuffled during the join operation.

    

    Optimization suggestion: If an SQL statement contains the redistribution operator, we recommend that you check the setting of the distribution key. For more information about redistribution scenarios and setting of the distribution key, see Distribution key.

  • Example 2: The join key is involved in an expression. The data type is changed. As a result, the local join feature is not used, and redistribution occurs.

    Optimization suggestion: Do not use expressions.

Join

The definition of the Join operator in the execution plan is the same as that in a standard database. The join operation can be classified into hash join, nested loop join, and merge join.

• Hash Join

  Hash join is a way of joining two or more tables. During a hash join, a hash table is built in the memory based on a to-be-joined table, which is usually a small table. The data of the to-be-joined columns is hash calculated and then inserted into the hash table. Data in other to-be-joined tables is read by row, hash calculated, and matched with data in the hash table. Matched data is returned. The following table describes the categories of hash joins.

  Category	Description
  Hash Left Join	When multiple tables are joined, all rows in the left table that meet the join conditions are returned and then matched against the right table. If no data is matched, null is returned.
  Hash Right Join	When multiple tables are joined, all rows in the right table and the rows in the left table that meet the join conditions are returned. If the rows in the right table do not match data in the left table, null is returned for the left table.
  Hash Inner Join	When multiple tables are joined, only the rows that meet the join conditions are returned.
  Hash Full Join	When multiple tables are joined, all rows in the left table and the right table are returned. If the data in one table does not match data in the other table, null is returned for the table whose data cannot be matched.
  Hash Anti Join	Only unmatched data is returned. This type of join is mostly used for queries with the NOT EXISTS clause.
  Hash Semi Join	Rows are returned if a data record is matched. The returned rows do not contain duplicate data. This type of join is usually used for queries with the EXISTS clause.

  When you view execution plans of hash join operations, you also need to focus on the child nodes.

  • hash cond: the join condition. Example: hash cond(tmp.a=tmp1.b).

  • hash key: the key that is used for hash calculation in multiple shards. In most cases, the key indicates the key of GROUP BY.

  During a hash join, you need to check whether the table with a small amount of data is used to create a hash table. You can use one of the following methods to check whether the small table is used to create a hash table:

  • In the execution plan, the table with the hash keyword is the table that is used to create a hash table.

  • In the execution plan, the bottom table is the table that is used to create a hash table.

  Optimization suggestion:

  • Update statistics

    The core idea is to use the small table to create a hash table. If a large table is used to create a hash table in the memory, more resources are consumed. In most cases, this issue occurs because statistics of the table are not updated and the QO uses a large table to create a hash table.

    In this example, the hash_join_test_2 (also named tbl2) table contains 1,000,000 rows of data, and the hash_join_test_1 (also named tbl1) table contains 10,000 rows of data. However, the table statistics are not updated and indicate that the tbl2 table contains 1,000 rows of data. As a result, the tbl2 table is considered as the small table and is used to create a hash table. The query efficiency is low.

    BEGIN ;
    CREATE TABLE public.hash_join_test_1 (
        a integer not null,
        b text not null
    );
    CALL set_table_property('public.hash_join_test_1', 'distribution_key', 'a');
    CREATE TABLE public.hash_join_test_2 (
        c integer not null,
        d text not null
    );
    CALL set_table_property('public.hash_join_test_2', 'distribution_key', 'c');
    COMMIT ;
    
    INSERT INTO hash_join_test_1 SELECT i, i+1 FROM generate_series(1, 10000) AS s(i);
    INSERT INTO hash_join_test_2 SELECT i, i::text FROM generate_series(10, 1000000) AS s(i);
    
    EXPLAIN SELECT * FROM hash_join_test_1 tbl1  JOIN hash_join_test_2 tbl2 ON tbl1.a=tbl2.c;

    The following figure shows the execution plan. The large table hash_join_test_2 is used to create a hash table.

    If the table statistics are not updated, you can manually execute the analyze <tablename> statement to update the statistics. Sample statements:

    ANALYZE hash_join_test_1;
    ANALYZE hash_join_test_2;

    The following figure shows the execution plan after the table statistics are updated. The small table hash_join_test_1 is used to create a hash table, and the number of rows estimated by the QO is correct.

  • Adjust the join order

    In most cases, you can resolve join issues by updating table statistics. However, based on the default mechanism, if the SQL statements are complex and five or more tables are joined, the Hologres QO selects an optimal execution plan based on the SQL statements.The selection process is time-consuming. You can execute the following statement to configure the Grand Unified Configuration (GUC) parameter and control the join order to accelerate the selection process of the QO:

    SET optimizer_join_order = '<value>'; 

    The following table describes valid values of the GUC parameter.

    Valid value	Description
    exhaustive (default value)	The join order is determined by using an algorithm, and an optimal execution plan is generated. This may increase the QO overhead during multi-table join.
    query	The execution plan is generated based on SQL statements. The QO does not make any changes. This value is applicable and helps decrease the QO overhead only if the multiple tables to be joined contain no more than hundreds of millions of rows of data. We recommend that you do not configure this GUC parameter at the database level. Otherwise, the performance of other join operations deteriorates.
    greedy	The join order is generated by using the greedy algorithm. In this mode, the QO overhead is moderate.

• Nested loop join and Materialize

  In a nested loop join of multiple tables, data is read from one table into an outer table. Each data record of the outer table is traversed into an inner table. Then, the inner and outer tables are joined in nested loops. This process is equivalent to the process of calculating the Cartesian product. In the execution plan, the first inner table usually has a Materialize operator.

  Optimization suggestion:

  • The principle of nested loop is that the inner table is driven by the outer table. Each row returned by the outer table must match a row in the inner table. Therefore, the returned result set cannot be too large. Otherwise, large amounts of resources are consumed. We recommend that you use the table that returns a small result as the outer table.

  • Non-equivalent joins usually generate nested loop joins. We recommend that you prevent non-equivalent joins in SQL statements.

  • The following code provides an example of a nested loop join.

    BEGIN;
    CREATE TABLE public.nestedloop_test_1 (
        a integer not null,
        b integer not null
    );
    CALL set_table_property('public.nestedloop_test_1', 'distribution_key', 'a');
    CREATE TABLE public.nestedloop_test_2 (
        c integer not null,
        d text not null
    );
    CALL set_table_property('public.nestedloop_test_2', 'distribution_key', 'c');
    COMMIT;
    
    INSERT INTO nestedloop_test_1 SELECT i, i+1 FROM generate_series(1, 10000) AS s(i);
    INSERT INTO nestedloop_test_2 SELECT i, i::text FROM generate_series(10, 1000000) AS s(i);
    
    EXPLAIN SELECT * FROM nestedloop_test_1 tbl1,nestedloop_test_2 tbl2 WHERE tbl1.a>tbl2.c;

    The following figure shows the execution plan. The Materialize and Nested Loop operators are displayed. This indicates that the SQL statement uses the nested loop join.

• Cross join

  In Hologres V3.0 and later, a cross join is used as an optimized implementation of a nested loop join in scenarios such as non-equivalent joins that involve small tables. In a nested loop join, a row of data is extracted from the outer loop, all data in the inner loop is traversed, and then the subquery status of the inner loop is reset. In a cross join, all data in small tables is loaded to the memory, and the data is joined with the data that is read from large tables in streaming mode. This significantly improves computing performance. However, cross joins consume more memory resources than nested loop joins.

  The following figure shows the execution plan. The Cross Join operator is displayed. This indicates that the SQL statement uses the cross join.

  

  If you want to disable cross joins, execute the following SQL statements to disable it.

  -- Disable cross joins at the session level.
  SET hg_experimental_enable_cross_join_rewrite = off;
  
  -- Disable cross joins at the database level. This configuration takes effect for new connections.
  ALTER database <database name> hg_experimental_enable_cross_join_rewrite = off;

Broadcast

Broadcast is used to distribute data to each shard. In most cases, broadcast join is used to join small tables with large tables. When an SQL statement is delivered, the QO compares the costs of redistribution and broadcast, and then generates an execution plan based on the algorithm.

Optimization suggestion:

• If the table that you want to query is small and the instance contains a small number of shards, such as 5 shards, broadcast is recommended.

  In this example, two tables are joined. The data amount of the broadcast_test_ 1 table and that of the broadcast_test_ 2 table are significantly different.

  BEGIN;
  CREATE TABLE broadcast_test_1 (
      f1 int, 
      f2 int);
  CALL set_table_property('broadcast_test_1','distribution_key','f2');
  CREATE TABLE broadcast_test_2 (
      f1 int,
      f2 int);
  COMMIT;
  
  INSERT INTO broadcast_test_1 SELECT i AS f1, i AS f2 FROM generate_series(1, 30)i;
  INSERT INTO broadcast_test_2 SELECT i AS f1, i AS f2 FROM generate_series(1, 30000)i;
  
  ANALYZE broadcast_test_1;
  ANALYZE broadcast_test_2;
  
  EXPLAIN SELECT * FROM broadcast_test_1 t1, broadcast_test_2 t2 WHERE t1.f1=t2.f1;

  The following figure shows the returned result.

  

• If the tables to be joined are not small tables but the broadcast operator is used, the table statistics may not be updated as expected. For example, the statistics indicate that the table contains 1,000 rows of data, but actually the table contains 1,000,000 rows of data. In this case, execute the analyze <tablename> statement to update the table statistics.

Shard prune and Shards selected

• Shard prune

  The method that is used to obtain a shard. Valid values:

  • lazaily: The system labels shards based on the shard IDs in the node and uses the specific shards during subsequent computing operations.

  • eagerly: The system selects the hit shards.

  The QO automatically selects a method based on the execution plan.

• Shards selected

  The number of selected shards. The value 1 out of 20 indicates that one shard is selected from 20 shards.

ExecuteExternalSQL

As described in Hologres architecture, the query engines of Hologres are classified into Hologres Query Engine (HQE), PostgreSQL Query Engine (PQE), and Seahawks Query Engine (SQE). PQE is the native PostgreSQL engine. Some operators and functions that are not supported by HQE are executed by PQE. PQE is less efficient than HQE. If an execution plan contains the ExecuteExternalSQL operator, PQE is used.

• Example 1: An SQL statement uses PQE.

  CREATE TABLE pqe_test(a text);
  INSERT INTO pqe_test VALUES ('2023-01-28 16:25:19.082698+08');
  EXPLAIN SELECT a::timestamp FROM pqe_test;

  In the following execution plan, ExecuteExternalSQL is displayed, and PQE is used to process the ::timestamp operator.

  

• Example 2: The ::timestamp operator is changed to to_timestamp. HQE is used.

  EXPLAIN SELECT to_timestamp(a,'YYYY-MM-DD HH24:MI:SS') FROM pqe_test;

  In the following execution plan, ExecuteExternalSQL is not displayed, and PQE is not used.

Optimization suggestion: Use the execution plan to find the function or operator that is processed by PQE, and rewrite the SQL statement to enable HQE to process the function or operator. This helps improve the query efficiency. For more information about how to rewrite operators, see Optimize query performance.

EXPLAIN SELECT l_shipdate FROM public.lineitem ORDER BY l_shipdate;

EXPLAIN SELECT * FROM public.lineitem limit 1;

References

You can view execution plans in a visualized manner in HoloWeb. For more information, see View execution plans.