How Dynamic Filters Supercharge MaxCompute Joins and Cut CPU by 70%

Business Background

Merchant billing is a core product of Alipay’s acquiring business, providing accurate reconciliation for merchants. To improve the timeliness of offline billing pipelines, the system often faces cross‑period data association challenges, where related data resides in different date partitions for years.

What Is Cross‑Period Association?

Offline processing frequently merges data from multiple systems stored in separate ODS tables. Because different systems may write records at different times, the data for a single business event can be spread across multiple date partitions. For example, a trade record may appear in trade_base for a given date, while the corresponding trade_ext row remains unchanged, requiring a cross‑partition join to retrieve the full information.

Business Pain Point

Querying a small set of primary‑key rows from a massive table (e.g., large_his_table) via a join with a small index table forces the optimizer to scan the entire large table before filtering, leading to high CPU and I/O consumption and long execution times.

SELECT t2.*
FROM (
  SELECT id
  FROM small_index_table -- small table
  WHERE dt = '20240818'
) t1
JOIN (
  SELECT *
  FROM large_his_table -- large table
  WHERE dt = '20240818'
) t2
ON t1.id = t2.id;

MaxCompute Dynamic Filter Capability

JOIN operations are resource‑intensive due to the shuffle phase. MaxCompute can exploit the equality‑join property to generate a runtime filter on the producer side and apply it to the consumer side before the shuffle, reducing data transferred.

Two main filter types are used:

Bloom Filter : a probabilistic structure that efficiently tests set membership.

Range Filter / IN Predicate : deterministic filters based on value ranges.

When the join key is a partition column, MaxCompute further applies Dynamic Partition Pruning (DPP) to eliminate whole partitions before reading them.

Dynamic Partition Pruning Example

-- A is a non‑partitioned table, column a = 20200701
-- B is a partitioned table, column ds has partitions 20200701, 20200702, 20200703
SELECT * FROM (table1) A JOIN (table2) B ON A.a = B.ds;

With DPP enabled, the optimizer builds a Bloom filter from table A, prunes irrelevant partitions of table B, and skips scheduling tasks that would read only pruned partitions.

How to Enable Dynamic Filters

Two methods are provided:

Use a SQL hint: /*+dynamicfilter(Producer, Consumer1[, Consumer2,...])*/. Example:

SELECT /*+dynamicfilter(A, B)*/ *
FROM (table1) A JOIN (table2) B ON A.a = B.b;

Set a session‑level switch so the optimizer automatically inserts filters when it estimates a benefit: set odps.optimizer.enable.dynamic.filter=true; Because this method relies on accurate table statistics, the hint approach is generally recommended.

Best Practice in Ant Group

For the described merchant‑billing scenario, a map‑join can reduce one shuffle but does not lower the amount of data scanned. By converting the large table to a hash‑clustered table and applying a dynamic filter, most irrelevant rows are eliminated before the join, dramatically cutting I/O and CPU.

-- Prevent mapjoin rule from being chosen
set odps.optimizer.cbo.rule.filter.black=hj;
set odps.optimizer.enable.nested.conditional.mapjoin=false;
set odps.optimizer.enable.conditional.mapjoin=false;

SELECT /*+dynamicfilter(t1, t2)*/ t2.*
FROM (
  SELECT id FROM small_index_table WHERE dt = '20240818'
) t1
JOIN (
  SELECT * FROM large_his_table -- hash‑clustered on id
  WHERE dt = '20240818'
) t2
ON t1.id = t2.id;

The job log shows a DynamicFilterConsumer operator, confirming that the filter was applied.

Online Business Optimization Results

Performance comparison:

Map‑join only: 38 s, 833 GB read, 135 core·min CPU.

Hash‑cluster + dynamic filter: 10 s, 38.8 GB read, 31 core·min CPU, 70 % CPU reduction and 95 % I/O reduction.

Summary

MaxCompute, Alibaba’s industry‑leading distributed big‑data platform, continuously improves SQL expressiveness and performance. In the merchant‑billing offline‑production project, combining dynamic filters with hash‑clustered tables reduced scanned files by over 95 %, cut CPU consumption by 70 %, and dramatically accelerated query execution.