CnchMergeTree is the Recommend table engine in ByteHouse Cloud Data Warehouse. It is designed to address high-performance analytical query challenges under massive data storage, helping the User gain rapid insights into big Data. It is suitable for Business Scenarios such as Log Analysis, TEA Event Analysis, Real-time data warehouse, and others that require fast aggregation and querying of large-scale datasets. This article will introduce the principles and syntax of the CnchMergeTree table engine.

Table engine introduction #

A table engine is the type of a table, which determines:

• the organization and storage method of Data
• the method and type of index
• which queries are supported and how they are supported
• some other specific functions and configurations

The Recommend table engine in ByteHouse Cloud Data Warehouse is CnchMergeTree. In addition, there are other special types of table engines, including Hive external table, Paimon external table, Iceberg external table, MySQL external table, Kafka table, and others.

CnchMergeTree table engine #

The core concept of CnchMergeTree is similar to LSM-Tree: Data is partitioned by partition key (PARTITION BY), and then stored in order by sorting key (ORDER BY). Main features include the following:

• Logical partition
  If a partition key is specified when creating the table, Data will be divided into different logical datasets (logical partition, Partition) according to the partition key. Each logical partition can contain zero or more data parts (DataPart). If the query condition can prune partitions, it usually accelerates the query.
  If no partition key is specified when creating the table, all Data will be in a single logical partition.
• Data part
  Data within a data part is sorted by sorting key. Each data part also has a min/max index to accelerate partition selection.
• Data granule (Granule)
  Each data part is logically divided into granules (Granule), with the default granule being 8192 rows (determined by the index_granularity setting when creating the table). A granule is the smallest indivisible dataset for querying Data in ByteHouse. The first row of each granule is marked by the primary key value of that row. ByteHouse creates an index file for each data part to store these marks. For each column, whether or not it is included in the primary key, ByteHouse stores similar marks. With these marks, the system can quickly locate the required data in the column files.
  A granule is both the indexing target of ByteHouse sparse indexes and the unit for data scanning in memory.
• Background merge
  Background tasks periodically merge DataParts within the same partition and maintain order according to the sorting key. Background merges reduce the number of DataParts for more efficient storage and improve query performance.

CnchMergeTree table creation syntax #

The CnchMergeTree table engine supports the following table creation semantics:

CREATE TABLE [IF NOT EXISTS] [db.]table_name
(
    name1 [type1] [DEFAULT|ALIAS expr1] [compression_codec] [TTL expr1],
    name2 [type2] [DEFAULT|ALIAS expr2] [compression_codec] [TTL expr2],
    ...
    INDEX index_name1 expr1 TYPE type1(...) GRANULARITY value1,
    INDEX index_name2 expr2 TYPE type2(...) GRANULARITY value2,
) ENGINE = CnchMergeTree()
ORDER BY expr
[PARTITION BY expr]
[CLUSTER BY (column, expression, ...) INTO value1 BUCKETS SPLIT_NUMBER value2 WITH_RANGE]
[PRIMARY KEY expr]
[UNIQUE KEY expr]
[SAMPLE BY expr]
[TTL expr]
[SETTINGS name=value, ...]

Table creation statement configuration parameter description #

Design partition key (PARTITION BY) #

The partition key is used to define partitions. A partition is a logical data set divided within a table according to specified rules. Partitions can be created by any standard, but in ByteHouse, it is recommended to use date-type fields as partition fields. To reduce the amount of data that needs to be operated on, each partition is stored separately. During queries, ByteHouse tries to use the smallest subset of these partitions. Specify during table creation using the PARTITION BY expr clause. The partition key can be any expression of columns in the table. For example, to partition by month, use the expression toYYYYMM(date); or partition by tuple expressions, such as (toMonday(date), EventType), and so on.
Caution: The value range calculated by the partition expression in the table should not be too large (recommended not to exceed ten thousand); too many partitions will occupy a large amount of memory and cause more IO and computation overhead.
A well-designed partition key can greatly reduce the amount of data that needs to be scanned during queries. Generally, consider designing the column with the most commonly used condition in queries and a value range not exceeding ten thousand as the partition key (such as date, and so on).

Design sorting key (ORDER BY) #

This can be a tuple of a group of columns or any expression. For example: ORDER BY (OrderID, Date).
If sorting is not required, use ORDER BY tuple(), and DataPart will be stored in the order of data insertion.

Primary key (PRIMARY KEY) #

By default, there is no need to explicitly specify it; ByteHouse will use the sorting key as the primary key. In special scenarios where the primary key and the sort key are not consistent, the primary key must be the leftmost prefix of the sort key. For example, if the sort key is (OrderID, Date), the primary key must be OrderID and cannot be Date.
ByteHouse establishes a sparse index on the primary key, with Granule as the unit. In contrast, a dense index means that index information is created for every row.
If the search condition matches the leftmost prefix of the primary key index, the primary key index can quickly filter out the data granules that may need to be read. Compared to scanning the entire DataPart, this is usually much more efficient.
Caution: PRIMARY KEY does not guarantee uniqueness, so it is possible to insert data rows with repeated primary keys.
Partition (PARTITION BY) and primary key (PRIMARY KEY) are two different methods to accelerate data search. When defining them, it is recommended to use different columns for each to cover more search scenarios. For example, since the primary key is by default the prefix of the sort key (ORDER BY), it is recommended not to select the first column of the sort key as the partition key.
Here are some considerations for selecting a primary key:

• Whether it is a commonly used column in the search condition;
• It is not the first column of the non-partition key;
• The selectivity of the column; for example, columns with too few possible values, such as gender or yes/no, are not recommended to be included in the primary key;
• If the current primary key (for example, (a, b)) filters data at too coarse a granularity, resulting in a large scope of data located during search (containing multiple Granule), you can consider adding another commonly used search column to the primary key for more accurate data filtering;
• Overly long primary keys have a negative impact on insert performance and memory consumption, but do not affect search performance.

Unique key (UNIQUE KEY) #

ByteHouse's primary key (PRIMARY KEY) does not guarantee uniqueness. If deduplication using a unique key is required, you need to set a unique key index when creating the table. After setting a unique key, ByteHouse provides upsert update write semantics. You can efficiently update data rows according to the unique key, or during upsert, by setting the virtual column _delete_flag_=1, you can use it to delete the specified key. Queries automatically return the latest value for each unique key.
A unique key can be a tuple of columns or any expression, such as UNIQUE KEY (product_id, sipHash64(city)).
When querying by unique key, the unique key index is used to filter data and accelerate the query, so the sorting primary key can often be set to columns different from the unique key to cover more query conditions.
For more details on unique key indexes and unique key tables, refer to Unique key table, Best practices for ByteHouse unique key tables.

Bucketing (Cluster By) #

Bucketing is commonly used in the following scenarios:

1. General scenarios: Data distribution is uneven
   • Definition and principle: When partitioning cannot achieve uniform data distribution, bucketing fields can be used. By setting bucketing fields, you can ensure that a column's data is evenly distributed across every node in the cluster. This maximizes cluster performance for queries. Proper settings of partition fields also help solve data skew problems, ensuring more uniform data distribution.
   • Field restriction: Nullable is not supported.
   • Configuration suggestion: Select fields that frequently appear in the grouping basis.
   • After the table is created successfully, the column type of this field cannot be changed.
2. Specific scenarios: Slow repeated data deletion speed
   • Definition and principle: When a Unique Key is set and there is too much data in a single partition (for example, more than 100 million rows), the data ingestion speed will be affected. This is because a lock must be acquired to perform repeated data deletion. In this case, you can divide the partition into buckets to improve data ingestion speed.
   • Field restriction: Nullable is not supported.
   • Configuration suggestion: The Bucket Key needs to be the same as the Unique Key. Each bucket should contain fewer than 10 million rows.

For related operations on bucketing tables, refer to:

• 创建动态分区表
• Bucket deduplication optimizations
• Bucket table dynamic expansion

Sampling #

The expression used for sampling; this configuration is optional.
To use a sampling expression, the primary key must contain this expression. For example:

SAMPLE BY intHash32(UserID) 
ORDER BY (CounterID, EventDate, intHash32(UserID))

TTL for columns and tables #

A list of rules specifying the duration of row storage and defining the logic for moving data fragments on disks and volumes; this configuration is optional. For more TTL usage, see 列和表的 TTL.
The expression must contain at least one column of Date or DateTime type, for example:

TTL date + INTERVAl 1 DAY

Compression #

The compression_codec field can be used to configure the codec. This configuration is optional, and the default value is LZ4.
ByteHouse Support project general purpose encoding and specific encoding. General codecs are similar to default codecs (LZ4, ZSTD) and their modified versions. Specific codecs are designed to leverage the particular characteristics of data to make compression more effective.

• General encoding
  • NONE: No compression.
  • LZ4: Default value, lossless high-speed compression algorithm.
  • LZ4HC[(level)]: LZ4HC high compression ratio algorithm with configurable levels. The default value for level is 9, supported values are [1 ~ 12], recommended to use [4 ~ 9].
  • ZSTD[(level)]: ZSTD compression algorithm with configurable levels. The default value for level is 1, supported values are [1 ~ 22].
• Specific encoding algorithms
  • Delta(delta_bytes): Delta encoding, which retains the first value and stores the difference between every two subsequent values. The default value is sizeof(type). Optional values are 1, 2, 4, or 8. If other values are provided, it is treated as 1.
• Multiple codecs
  Use multiple codecs as described above. Compression will be carried out according to the order declared by the codecs, while decompression will be performed in the reverse order.

Example:

CREATE TABLE codec_example
(
    date Date CODEC(Delta, ZSTD), 
    ts DateTime CODEC(LZ4HC),
    float_value Float32 CODEC(NONE),
    double_value Float64 CODEC(LZ4HC(9))
)
ENGINE = CnchMergeTree()
PARTITION BY tuple()
ORDER BY date

More configuration #

For more table creation related configuration, such as Unique tables, bucketing tables, and others, refer to the corresponding document in Best practice for creating tables in ByteHouse.

Description of architectural advantages and disadvantages #

The core value of CnchMergeTree merge lies in store in small batches, retrieve in bulk: Data is imported into the table in different batches, but merging reduces the number of files and stores data sequentially, allowing ByteHouse to fully utilize the disk's powerful sequential read capability and significantly improve search performance.
But the problem with merging is also obvious: If the background writes are too fragmented (for example, only a few hundred or dozens of rows are inserted each time), it will bring a large number of parts. The merge Mission will lead to increased CPU overhead and memory usage, resulting in query task performance falling or even errors. In addition, if too many small files slow down the merge, it will also cause performance degradation when searching for the latest data, as the parts have not yet been merged.

FAQs #

Why is it not recommended to use SELECT *? #

Because ByteHouse is a columnar storage database, data is stored in different column storage files (.bin). This design allows only a limited number of files to be read when searching for specified columns, accelerating the search.
If SELECT* is used, the backend needs to read all .bin column storage files, which essentially cancels the advantages brought by columnar storage.

Why is it not recommended to use INSERT INTO to insert data? #

Each INSERT INTO operation creates a new part folder, and repeatedly calling INSERT INTO results in many parts, with each part containing only a small amount of data. The backend requires a long merge process to reduce the number of parts. Problems caused:

• During this period, search is extremely slow, because a range query may cross several column storage files in different parts.
• Merge may not finish for a long time, occupying system resources.

Why is it suggested to add conditions restricting partition fields when searching? #

After restricting partition, the search will only scan a limited number of part directories, reducing the amount of data scanned and greatly accelerating the search.

Why is it recommended to set the partition field as a date field? #

Currently, ByteHouse only supports partition key configuration using fields that can be converted to date (int, string, date, datetime). From a business perspective, the data volume per day or per hour is similar. Partitioning by date field ensures each partition is more balanced in size, preventing significant latency fluctuations for individual searches.

Why is it recommended to set the sorting index as the most commonly used field in queries? #

As previously mentioned, each block is sorted according to the sorting index, and a sparse index is built based on this field. As long as the query condition includes the sorting index, the MergeTree engine uses the mark in the index to map rows to data, pruning unnecessary granules and reducing the number of scanned rows, thereby lifting search performance.
If the search does not include the sorting index, only a full file scan can be performed, which is highly inefficient.