ApsaraDB for ClickHouse:Table engines

Table engines refer to the types of tables. In ApsaraDB for ClickHouse, table engines determine how data is stored and read, whether indexes are supported, and whether primary/secondary replication is supported. The following table describes the table engines supported by ApsaraDB for ClickHouse.

In this example, a primary key is used to show the feature difference. In the SQL syntax of ApsaraDB for ClickHouse, the primary key is used to remove duplicates to ensure that data is unique. In the MergeTree table engine, the primary key is used to accelerate queries. Even after the compaction is complete, data rows that have the same primary key still exist.

The following example shows how to use the MergeTree table engine:

1. Create a table named test_tbl. The primary key is (id, create_time). The stored data is sorted based on the primary key and is partitioned based on the value of create_time.

   CREATE TABLE test_tbl ON CLUSTER default (
     id UInt16,
     create_time Date,
     comment Nullable(String)
   ) ENGINE = MergeTree()
      PARTITION BY create_time
        ORDER BY  (id, create_time)
        PRIMARY KEY (id, create_time)
        SETTINGS index_granularity=8192;

2. Write data that has duplicate primary keys.

   insert into test_tbl values(1, '2019-12-13', null);
   insert into test_tbl values(1, '2019-12-13', null);
   insert into test_tbl values(2, '2019-12-14', null);
   insert into test_tbl values(3, '2019-12-15', null);
   insert into test_tbl values(3, '2019-12-15', null);

3. Query data.

   select * from test_tbl;

   The following query result is returned:

   ┌─id─┬─create_time─┬─comment──┐
   │  1 │ 2019-12-13  │   NULL   │
   │  1 │ 2019-12-13  │   NULL   │
   │  2 │ 2019-12-14  │   NULL   │
   │  3 │ 2019-12-15  │   NULL   │
   │  3 │ 2019-12-15  │   NULL   │
   └────┴─────────────┴──────────┘

4. Execute the OPTIMIZE statement to forcibly perform compaction in the background. Compaction is forcibly performed because the table engines in the MergeTree family use a structure that is similar to a log-structured merge-tree (LSM-tree) and the processing logic at the storage layer is not implemented until the compaction phase is started.

   optimize table test_tbl final;

5. Query data again.

   select * from test_tbl;

   The following query result is returned. Data that has duplicate primary keys still exists.

   ┌─id─┬─create_time─┬─comment──┐
   │  1 │ 2019-12-13  │   NULL   │
   │  1 │ 2019-12-13  │   NULL   │
   │  2 │ 2019-12-14  │   NULL   │
   │  3 │ 2019-12-15  │   NULL   │
   │  3 │ 2019-12-15  │   NULL   │
   └────┴─────────────┴──────────┘

Although the ReplacingMergeTree table engine can remove duplicates that have the same primary key, the table engine still has the following limits. Therefore, the ReplacingMergeTree table engine is used to ensure that duplicates are eventually removed for data, but cannot ensure that the primary keys are unique during the query process.

• In a distributed scenario, the data that has the same primary key may be distributed to different nodes. Duplicates cannot be removed for data across different shards.

• Before the execution of the OPTIMIZE statement is complete, duplicates that have the same primary key may not be removed. For example, duplicates are removed for some data, but duplicates that have the same primary key are not removed for other data.

• The OPTIMIZE statement is executed in the background, and the execution time cannot be predicted.

• In a scenario in which a large volume of data exists, a long period of time is required to manually execute the OPTIMIZE statement. In this case, requirements for real-time queries cannot be met.

The following example shows how to use the ReplacingMergeTree table engine:

1. Create a table named test_tbl_replacing.

   CREATE TABLE test_tbl_replacing (
     id UInt16,
     create_time Date,
     comment Nullable(String)
   ) ENGINE = ReplacingMergeTree()
      PARTITION BY create_time
        ORDER BY  (id, create_time)
        PRIMARY KEY (id, create_time)
        SETTINGS index_granularity=8192;

2. Write data that has duplicate primary keys.

   insert into test_tbl_replacing values(1, '2019-12-13', null);
   insert into test_tbl_replacing values(1, '2019-12-13', null);
   insert into test_tbl_replacing values(2, '2019-12-14', null);
   insert into test_tbl_replacing values(3, '2019-12-15', null);
   insert into test_tbl_replacing values(3, '2019-12-15', null);

3. Query data.

   select * from test_tbl_replacing;

   The following query result is returned:

   ┌─id─┬─create_time─┬─comment──┐
   │  1 │ 2019-12-13  │   NULL   │
   │  1 │ 2019-12-13  │   NULL   │
   │  2 │ 2019-12-14  │   NULL   │
   │  3 │ 2019-12-15  │   NULL   │
   │  3 │ 2019-12-15  │   NULL   │
   └────┴─────────────┴──────────┘ 

4. Execute the OPTIMIZE statement to forcibly perform compaction in the background. Compaction is forcibly performed because the table engines in the MergeTree family use a structure similar to an LSM tree and processing logic at the storage layer is not implemented until the compaction phase is started.

   optimize table test_tbl_replacing final;

5. Query data again.

   select * from test_tbl_replacing;

   The following query result is returned. Data that has duplicate primary keys is removed.

   ┌─id─┬─create_time─┬─comment──┐
   │  1 │ 2019-12-13  │   NULL   │
   │  2 │ 2019-12-14  │   NULL   │
   │  3 │ 2019-12-15  │   NULL   │
   └────┴─────────────┴──────────┘     

The CollapsingMergeTree table engine eliminates the limits on the features of the ReplacingMergeTree table engine. When you use the CollapsingMergeTree table engine, you must specify the Sign column in the CREATE TABLE statement. Rows are divided into two categories based on the value of the Sign column: state row and cancel row. Rows for which the value of the Sign column is 1 are called state rows. State rows are used to add states. Rows for which the value of the Sign column is -1 are called cancel rows. Cancel rows are used to delete states.

During compaction in the background, rows that have the same primary key and opposite Sign values are collapsed or deleted. If no compaction is performed, state rows and cancel rows exist at the same time. To collapse or delete rows that have the same primary key, perform the following operations at the business layer:

• Record the values of original state rows, or query the database to obtain the values of original state rows before you delete the states.

  The reason is that you must write cancel rows when you delete states. Cancel rows must contain the data that has the same primary keys as the primary keys of original state rows, except the Sign column.

• When you execute aggregate functions, such as count() and sum(col), data redundancy may exist. To obtain valid results, modify SQL statements at the business layer. Change count() to sum(Sign) and sum(col) to sum(col * Sign).

  The following reasons are available:

  • The time to perform compaction in the background cannot be predicted. When you initiate a query, state rows and cancel rows may not be collapsed or deleted.

  • ApsaraDB for ClickHouse cannot ensure that rows with the same primary key fall on the same node. Data cannot be collapsed or deleted across nodes.

The following example shows how to use the CollapsingMergeTree table engine:

1. Create a table named test_tbl_collapsing.

   CREATE TABLE test_tbl_collapsing
   (
       UserID UInt64,
       PageViews UInt8,
       Duration UInt8,
       Sign Int8
   )
   ENGINE = CollapsingMergeTree(Sign)
   ORDER BY UserID;

2. Insert a state row for which the value of the Sign column is 1.

   INSERT INTO test_tbl_collapsing VALUES (4324182021466249494, 5, 146, 1);

   Note

   If you insert a cancel row and then a state row, the cancel row and the state row may be out of order. In this case, even if the compaction is forcibly performed in the background, the data that has the same primary key cannot be collapsed or deleted.

3. Insert a cancel row for which the value of the Sign column is -1. Values in the cancel row are the same as the values in the inserted state row, except the value of the Sign column. At the same time, insert a new state row that has the same primary key as the cancel row.

   INSERT INTO test_tbl_collapsing VALUES (4324182021466249494, 5, 146, -1), (4324182021466249494, 6, 185, 1);

4. Query data.

   SELECT * FROM test_tbl_collapsing;

   The following query result is returned:

   ┌────────UserID───────┬─PageViews─┬─Duration─┬─Sign──┐ 
   │ 4324182021466249494 │    5      │    146   │   1   │ 
   │ 4324182021466249494 │    5      │    146   │  -1   │ 
   │ 4324182021466249494 │    6      │    185   │   1   │ 
   └─────────────────────┴───────────┴──────────┴───────┘

5. If you want to execute aggregate functions on specified columns, modify the SQL statement to obtain valid results. In this example, the aggregate function sum(col) is used and the SQL statement is modified to the following statement:

   SELECT
       UserID,
       sum(PageViews * Sign) AS PageViews,
       sum(Duration * Sign) AS Duration
   FROM test_tbl_collapsing
   GROUP BY UserID
   HAVING sum(Sign) > 0;

   After data is aggregated, the following query result is returned:

   ┌────────UserID───────┬─PageViews─┬─Duration──┐ 
   │ 4324182021466249494 │    6      │    185    │ 
   └─────────────────────┴───────────┴───────────┘

6. Execute the OPTIMIZE statement to forcibly perform compaction in the background. Compaction is forcibly performed because the table engines in the MergeTree family use a structure similar to an LSM tree and processing logic at the storage layer is not implemented until the compaction phase is started.

   optimize table test_tbl_collapsing final;

7. Query data again.

   SELECT * FROM test_tbl_collapsing;

   The following query result is returned:

   ┌────────UserID───────┬─PageViews─┬─Duration─┬─Sign──┐  
   │ 4324182021466249494 │    6      │    185   │   1   │ 
   └─────────────────────┴───────────┴──────────┴───────┘

The following example shows how to use the VersionedCollapsingMergeTree table engine:

1. Create a table named test_tbl_Versioned.

   CREATE TABLE test_tbl_Versioned
   (
       UserID UInt64,
       PageViews UInt8,
       Duration UInt8,
       Sign Int8,
       Version UInt8
   )
   ENGINE = VersionedCollapsingMergeTree(Sign, Version)
   ORDER BY UserID;

2. Insert a cancel row for which the value of the Sign column is -1.

   INSERT INTO test_tbl_Versioned VALUES (4324182021466249494, 5, 146, -1, 1);

3. Insert a state row for which the value of the Sign column is 1 and the value of the Version column is 1. The values in other columns are the same as the values in the cancel row that is inserted. At the same time, insert a new state row that has the same primary key as the cancel row.

   INSERT INTO test_tbl_Versioned VALUES (4324182021466249494, 5, 146, 1, 1),(4324182021466249494, 6, 185, 1, 2);

4. Query data.

   SELECT * FROM test_tbl_Versioned;

   The following query result is returned:

   ┌────────UserID───────┬─PageViews─┬─Duration─┬─Sign───┬Version─┐ 
   │ 4324182021466249494 │    5      │    146   │  -1    │    1   │
   │ 4324182021466249494 │    5      │    146   │   1    │    1   │
   │ 4324182021466249494 │    6      │    185   │   1    │    2   │
   └─────────────────────┴───────────┴──────────┴────────┴────────┘

5. If you want to execute aggregate functions on specified columns, modify the SQL statement to obtain valid results. In this example, the aggregate function sum(col) is used and the SQL statement is modified to the following statement:

   SELECT
       UserID,
       sum(PageViews * Sign) AS PageViews,
       sum(Duration * Sign) AS Duration
   FROM test_tbl_Versioned
   GROUP BY UserID
   HAVING sum(Sign) > 0;

   After data is aggregated, the following query result is returned:

   ┌────────UserID───────┬─PageViews─┬─Duration─┐ 
   │ 4324182021466249494 │    6      │    185   │
   └─────────────────────┴───────────┴──────────┘

6. Execute the OPTIMIZE statement to forcibly perform compaction in the background. Compaction is forcibly performed because the table engines in the MergeTree family use a structure similar to an LSM tree and processing logic at the storage layer is not implemented until the compaction phase is started.

   optimize table test_tbl_Versioned final;

7. Query data again.

   SELECT * FROM test_tbl_Versioned;

   The following query result is returned:

   ┌────────UserID───────┬─PageViews─┬─Duration─┬─Sign───┬Version─┐ 
   │ 4324182021466249494 │    6      │    185   │   1    │    2   │
   └─────────────────────┴───────────┴──────────┴────────┴────────┘

Before you use the SummingMergeTree table engine, take note of the following points:

• In ApsaraDB for ClickHouse, primary key columns are pre-aggregated only when compaction is performed in the background. The time to perform compaction cannot be predicted. Therefore, some data may be pre-aggregated, whereas some data may not be pre-aggregated. The GROUP BY clause is still required in the SQL statement to perform aggregation.

• During pre-aggregation, ApsaraDB for ClickHouse pre-aggregates all columns except the primary key columns. If these columns can be aggregated, such as the columns of the NUMERIC data type, the values are summed. If the columns cannot be aggregated, such as the columns of the STRING data type, a random value is used.

• We recommend that you use the SummingMergeTree table engine in conjunction with the MergeTree table engine. The MergeTree table engine stores complete data, and the SummingMergeTree table engine stores pre-aggregated results.

The following example shows how to use the SummingMergeTree table engine:

1. Create a table named test_tbl_summing.

   CREATE TABLE test_tbl_summing
   (
       key UInt32,
       value UInt32
   )
   ENGINE = SummingMergeTree()
   ORDER BY key;

2. Write data.

   INSERT INTO test_tbl_summing Values(1,1),(1,2),(2,1);

3. Query data.

   select * from test_tbl_summing;

   The following query result is returned:

   ┌─key─┬value─┐
   │  1  │  1   │
   │  1  │  2   │
   │  2  │  1   │
   └─────┴──────┘    

4. Execute the OPTIMIZE statement to forcibly perform compaction in the background. Compaction is forcibly performed because the table engines in the MergeTree family use a structure similar to an LSM tree and processing logic at the storage layer is not implemented until the compaction phase is started.

   optimize table test_tbl_summing final;

5. After compaction is forcibly performed in the background, execute the statement that contains the GROUP BY clause to aggregate data. Then, query data again.

   SELECT key, sum(value) FROM test_tbl_summing GROUP BY key;

   The following query result is returned. Data that has duplicate primary keys has been aggregated.

   ┌─key─┬value─┐
   │  1  │  3   │
   │  2  │  1   │
   └─────┴──────┘

The syntax of the AggregatingMergeTree table engine is complex. The AggregatingMergeTree table engine must be used in conjunction with a materialized view or the special data type AggregateFunction in ApsaraDB for ClickHouse.

The following example shows how to use the AggregatingMergeTree table engine:

• Use the AggregatingMergeTree table engine in conjunction with a materialized view

  1. Create a detail table named visits.

     CREATE TABLE visits
     (
         UserID UInt64,
         CounterID UInt8,
         StartDate Date,
         Sign Int8
     )
     ENGINE = CollapsingMergeTree(Sign)
     ORDER BY UserID;

  2. Create a materialized view named visits_agg_view for the visits table, and execute the sumState and uniqState functions to pre-aggregate the visits table.

     CREATE MATERIALIZED VIEW visits_agg_view
     ENGINE = AggregatingMergeTree() PARTITION BY toYYYYMM(StartDate) ORDER BY (CounterID, StartDate)
     AS SELECT
         CounterID,
         StartDate,
         sumState(Sign)    AS Visits,
         uniqState(UserID) AS Users
     FROM visits
     GROUP BY CounterID, StartDate;

  3. Write data to the visits table.

     INSERT INTO visits VALUES(0, 0, '2019-11-11', 1);
     INSERT INTO visits VALUES(1, 1, '2019-11-12', 1);

  4. Execute the sumMerge and uniqMerge aggregate functions to aggregate the materialized view. Then, query the aggregated data.

     SELECT
         StartDate,
         sumMerge(Visits) AS Visits,
         uniqMerge(Users) AS Users
     FROM visits_agg_view
     GROUP BY StartDate
     ORDER BY StartDate

     Note

     The sum and uniq functions can no longer be used. If you use the preceding functions, the following error message is returned when you execute the SQL statement: Illegal type AggregateFunction(sum, Int8) of argument for aggregate function sum...

     The following query result is returned:

     ┌──StartDate──┬─Visits─┬─Users──┐
     │  2019-11-11 │   1    │   1    │
     │  2019-11-12 │   1    │   1    │
     └─────────────┴────────┴────────┘

• Use the AggregatingMergeTree table engine in conjunction with the special data type AggregateFunction

  1. Create a detail table named detail_table.

     CREATE TABLE detail_table
     (   CounterID UInt8,
         StartDate Date,
         UserID UInt64
     ) ENGINE = MergeTree() 
     PARTITION BY toYYYYMM(StartDate) 
     ORDER BY (CounterID, StartDate);

  2. Write data to detail_table.

     INSERT INTO detail_table VALUES(0, '2019-11-11', 1);
     INSERT INTO detail_table VALUES(1, '2019-11-12', 1);

  3. Create an aggregate table named agg_table in which the data type of the UserID column is AggregateFunction.

     CREATE TABLE agg_table
     (   CounterID UInt8,
         StartDate Date,
         UserID AggregateFunction(uniq, UInt64)
     ) ENGINE = AggregatingMergeTree() 
     PARTITION BY toYYYYMM(StartDate) 
     ORDER BY (CounterID, StartDate);

  4. Execute the uniqState aggregate function to insert the data of the detail table into the aggregate table.

     INSERT INTO agg_table
     select CounterID, StartDate, uniqState(UserID)
     from detail_table
     group by CounterID, StartDate;

     Note

     You cannot execute the INSERT INTO agg_table VALUES(1, '2019-11-12', 1); statement to insert data into the aggregate table. Otherwise, the following error message is returned: Cannot convert UInt64 to AggregateFunction(uniq, UInt64)...

  5. Execute the uniqMerge aggregate function to aggregate data in the aggregate table. Then, query the aggregated data.

     SELECT uniqMerge(UserID) AS state 
     FROM agg_table 
     GROUP BY CounterID, StartDate;

     The following query result is returned:

     ┌─state─┐
     │   1   │
     │   1   │
     └───────┘