COMPACTION

Features

Delta Tables support near real-time incremental writes and time travel queries. Frequent data writes create many small files. An efficient merge policy combines these small files and removes duplicate data. This process improves I/O efficiency and reduces the load on the storage system. However, frequent compactions can cause severe write amplification and conflict failures.

Two data merge methods are currently supported:

• Clustering: Merges committed DeltaFiles into a single large file without changing the data. The system runs this process periodically based on the size and number of new files. You do not need to run it manually. Clustering primarily addresses issues with I/O efficiency and stability caused by small files.

• Compaction: Merges all data files based on a specific policy to generate new BaseFiles. For data rows with the same primary key (PK), only the latest state is stored. Historical states and system column information are not included. The main purpose of compaction is to improve query efficiency.

Syntax

alter table <table_name> [partition 
                          (<partition_key> = '<partition_value>' [, ...])
                         ] 
                         compact major;

Notes

• To run compaction manually, you can set the following flag.

   set odps.merge.task.mode=service;

• Compaction creates new files. While this improves query efficiency, it also increases storage costs. Choose a trigger frequency that balances these trade-offs for your specific scenario.

• Compaction uses compute resources to read and rewrite data. If you use the pay-as-you-go billing method, you are charged for compaction. The cost is calculated as follows: Amount of data scanned × 1 × Unit price. For more information, see Compute costs (Pay-as-you-go). If you use the subscription billing method, compaction consumes your subscription's compute resource quota.

Examples

--Create a table
create table mf_dt (pk bigint not null primary key, val bigint not null) 
             partitioned by (dd string, hh string) 
             tblproperties ("transactional"="true");
--Insert data            
insert into table mf_dt partition(dd='01', hh='01') values (1, 1), (2, 2);
insert into table mf_dt partition(dd='01', hh='01') values (2, 20), (3, 3);
insert into table mf_dt partition(dd='01', hh='01') values (3, 30), (4, 4);

--Run compaction. After the operation completes, you can still query historical data.
set odps.merge.task.mode=service;
alter table mf_dt partition(dd='01', hh='01') compact major;
select * from mf_dt timestamp as of get_latest_timestamp('mf_dt') where dd='01' and hh='01';
select * from mf_dt timestamp as of get_latest_timestamp('mf_dt', 2) where dd='01' and hh='01';