Realtime Compute for Apache Flink:Lookup join

Shuffle strategies via LOOKUP hints

Configure 'shuffle' = 'true' in a LOOKUP hint to control how mainstream data is distributed before the join.

Only Streaming data lakehouse Paimon provides a custom shuffle strategy. If the join columns include all bucket fields, the dimension table is shuffled based on buckets.

Examples

All examples use FOR SYSTEM_TIME AS OF PROCTIME() to join dimension tables at processing time.

-- Apply shuffle to dim1 only
SELECT /*+ LOOKUP('table'='dim1', 'shuffle' = 'true') */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b

-- Apply shuffle to both dim1 and dim2
SELECT /*+ LOOKUP('table'='dim1', 'shuffle' = 'true'),LOOKUP('table'='dim2', 'shuffle' = 'true') */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b

-- Use alias D1 when the dimension table has an alias
SELECT /*+ LOOKUP('table'='D1', 'shuffle' = 'true') */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b

-- Use aliases for both dimension tables
SELECT /*+ LOOKUP('table'='D1', 'shuffle' = 'true'),LOOKUP('table'='D2', 'shuffle' = 'true') */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b

SHUFFLE_HASH

SHUFFLE_HASH shuffles mainstream data by join key before the join. With an LRU cache, this increases cache hit ratio and reduces I/O requests. With an ALL cache, this reduces per-operator memory usage to 1/parallelism.

Limitations:

• Mainstream data incurs additional network overhead from the shuffle. If the dimension table is small with an ALL cache and has no memory bottleneck, the network overhead may outweigh the memory savings.

• If mainstream data is heavily skewed on join keys, the join operator may become a bottleneck. Use REPLICATED_SHUFFLE_HASH instead.

You can specify multiple dimension tables in a single SHUFFLE_HASH hint.

-- Apply SHUFFLE_HASH to dim1 only
SELECT /*+ SHUFFLE_HASH(dim1) */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b

-- Apply SHUFFLE_HASH to both dim1 and dim2
SELECT /*+ SHUFFLE_HASH(dim1, dim2) */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b

-- Use alias D1 when the dimension table has an alias
SELECT /*+ SHUFFLE_HASH(D1) */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b

-- Use aliases for both dimension tables
SELECT /*+ SHUFFLE_HASH(D1, D2) */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b

REPLICATED_SHUFFLE_HASH

REPLICATED_SHUFFLE_HASH works like SHUFFLE_HASH but also randomly distributes records with the same join key across multiple concurrent threads. This resolves performance bottlenecks caused by data skew.

With an ALL cache, REPLICATED_SHUFFLE_HASH reduces per-operator memory usage to (number of buckets)/parallelism.

Limitations:

• Configure table.exec.skew-join.replicate-num to set the number of buckets for skewed data. The default is 16. The value cannot exceed the parallelism of the join operator. For configuration instructions, see How do I configure custom running parameters for a job?

• Update streams are not supported. If the mainstream is an update stream, an error is returned.

You can specify multiple dimension tables in a single REPLICATED_SHUFFLE_HASH hint.

-- Apply REPLICATED_SHUFFLE_HASH to dim1
SELECT /*+ REPLICATED_SHUFFLE_HASH(dim1) */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a

-- Use alias D1 when the dimension table has an alias
SELECT /*+ REPLICATED_SHUFFLE_HASH(D1) */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a

SKEW

SKEW is syntactic sugar for REPLICATED_SHUFFLE_HASH. When you specify SKEW on the primary (mainstream) table, Flink applies the REPLICATED_SHUFFLE_HASH strategy to the dimension table join.

Limitations:

• Each SKEW hint can reference only one table.

• Specify the primary table with data skew, not the dimension table.

• Update streams are not supported.

SELECT /*+ SKEW(src) */
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a