JOIN statements for dimension tables

Tuning parameter categories and how to pass them

Tuning a dimension table JOIN involves three categories of parameters. Each category has its own way of being passed, and using the wrong mechanism causes the hint to be silently ignored.

Override connector options with the OPTIONS hint

An OPTIONS() hint lets you override the WITH options of a table directly in a query, without modifying the CREATE TABLE statement. It is commonly used to tune cache behavior.

SELECT t.id, t.name, w.phoneNumber
FROM kafka_input AS t
LEFT JOIN phoneNumber /*+ OPTIONS(
  'lookup.cache' = 'PARTIAL',
  'lookup.partial-cache.max-rows' = '1000'
) */ FOR SYSTEM_TIME AS OF PROCTIME() AS w
ON t.name = w.name;

Every key inside OPTIONS must be a WITH option that the connector actually supports.

Configure lookup behavior with the LOOKUP hint

The LOOKUP hint behaves consistently with the Apache Flink community version. It configures sync/async mode, retry, and shuffle behavior for a single join operation. For details, see Apache Flink Lookup Hint.

• The LOOKUP hint is available only in VVR 8.0 and later.

• The 'shuffle' = 'true' option is available only in VVR 8.0.8 and later.

• Aliases are supported in VVR 8.0 and later. When the dimension table is referenced by an alias, the hint must use that alias.

-- Apply the shuffle strategy only to dimension table dim1.
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 the shuffle strategy 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;

-- When dim1 is referenced by the alias D1, the hint must use the 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;

-- Apply the shuffle strategy to both dim1 and dim2 using their aliases.
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;

Set job-level TableConfig with SET

table.exec.* and similar TableConfig keys are job-level global settings. They must be passed via SET; neither the LOOKUP hint nor the OPTIONS hint accepts them.

Common TableConfig keys for dimension table JOIN:

-- Set job-level defaults at the beginning of the job.
SET 'table.exec.async-lookup.buffer-capacity' = '3072';
SET 'table.exec.async-lookup.timeout' = '180s';

INSERT INTO sink_table
SELECT ...
FROM src_table
LEFT JOIN dim_table FOR SYSTEM_TIME AS OF PROCTIME() AS d
ON src_table.key = d.key;

Control shuffle with join-strategy hints

Join-strategy hints control shuffle behavior for dimension table joins. The available hints are SHUFFLE_HASH, REPLICATED_SHUFFLE_HASH, and SKEW. The table below shows which join strategy fits each cache strategy.

-- Enable SHUFFLE_HASH for dim1 only.
SELECT /*+ SHUFFLE_HASH(dim1) */ ...

-- Enable SHUFFLE_HASH for both dim1 and dim2.
SELECT /*+ SHUFFLE_HASH(dim1, dim2) */ ...

-- When dim1 is referenced by the alias D1, the hint must use the alias.
SELECT /*+ SHUFFLE_HASH(D1) */ ...

SELECT /*+ REPLICATED_SHUFFLE_HASH(dim1) */ ...

SELECT /*+ SKEW(src) */  ...

Example 1: A basic dimension table JOIN

The most basic form: enrich a Kafka stream with a MySQL dimension table. No tuning hints are used.

CREATE TEMPORARY TABLE kafka_input (
  id   BIGINT,
  name VARCHAR,
  age  BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE phoneNumber (
  name        VARCHAR,
  phoneNumber BIGINT,
  PRIMARY KEY (name) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

CREATE TEMPORARY TABLE result_infor (
  id          BIGINT,
  phoneNumber BIGINT,
  name        VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO result_infor
SELECT
  t.id,
  w.phoneNumber,
  t.name
FROM kafka_input AS t
JOIN phoneNumber FOR SYSTEM_TIME AS OF PROCTIME() AS w
ON t.name = w.name;

Example 2: Enable cache with the OPTIONS hint and configure async lookup with the LOOKUP hint

Tuning scenario: the dimension table's QPS is insufficient. You want to turn on partial cache on the dimension table and enable async lookup with a larger buffer queue.

INSERT INTO user_behavior_wide
SELECT /*+ LOOKUP('table' = 't2', 'async' = 'true', 'capacity' = '3072') */
  t1.member_id AS member_id,
  t2.tag AS tag
FROM user_behavior_datagen AS t1
LEFT JOIN fluss.fluss.user_active_info /*+ OPTIONS(
    'lookup.cache' = 'PARTIAL',
    'lookup.partial-cache.max-rows' = '1000'
  ) */
FOR SYSTEM_TIME AS OF PROCTIME() AS t2
ON t1.member_id = t2.member_id;

• Connector options (lookup.cache, lookup.partial-cache.max-rows) are passed through an OPTIONS() hint placed right after the dimension table reference; this overrides the corresponding WITH options at query time.

• Lookup behavior (async, capacity) is configured through a LOOKUP() hint placed after SELECT. OPTIONS and LOOKUP are two different hint types and must be written separately; combining them silently disables the hint.

Example 3: Tune the job-level async lookup buffer with SET

Tuning scenario: multiple dimension table JOINs in the same job need a larger asynchronous lookup buffer. You want the change to apply globally, and to enable async via the LOOKUP hint.

SET 'table.exec.async-lookup.buffer-capacity' = '3072';
SET 'table.exec.async-lookup.timeout' = '180s';

INSERT INTO user_behavior_wide
SELECT /*+ LOOKUP('table' = 't2', 'async' = 'true') */
  t1.member_id,
  t2.tag
FROM user_behavior_datagen AS t1
LEFT JOIN fluss.fluss.user_active_info
FOR SYSTEM_TIME AS OF PROCTIME() AS t2
ON t1.member_id = t2.member_id;

• table.exec.async-lookup.buffer-capacity and similar parameters are job-level TableConfig keys and can be set only with SET. Putting them inside OPTIONS or LOOKUP has no effect.

• When the LOOKUP hint does not specify capacity, the global value from SET is used. If both are specified, the LOOKUP hint takes precedence on that individual join.