Using the MultiJoin Operator in Flink SQL

Updated on 2024-12-13 GMT+08:00

View PDF

This topic is available for MRS 3.5.0 or later only.

Joining wide tables with Flink's Full outer Join operator increases backend pressure and slows performance. You can use MultiJoin operator to combine wide tables, doubling the computing performance.

Restrictions on Using the MultiJoin Operator in Flink SQL

The MultiJoin operator supports only FULL OUTER JOIN and INNER JOIN.
The MultiJoin operator supports only equi-joins. For non-equi joins, you can use a view for filtering.
When the MultiJoin operator is used to join multiple tables, the join keys of all tables must be the same.
The MultiJoin operator supports only table.exec.state.ttl and does not support table-level TTL or JTL.
States of Full outer join are incompatible with MultiJoin states. Jobs cannot be restored using snapshots if you switch between the two operators.

Using the MultiJoin Operator in Flink SQL

When you configure a Flink job, set table.optimizer.multi-join-enabled to true on the Flink job development page of the FlinkServer web UI to use the MultiJoin operator. For details, see Creating a FlinkServer Job.

The following is an example SQL statement:

CREATE TABLE datagen1 (
  id int,
  f1 int,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'datagen'
);

CREATE TABLE datagen2 (
  id int,
  f2 int,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'datagen'
);

CREATE TABLE datagen3 (
  id int,
  f3 int,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'datagen'
);

create view datagen3_view AS
select
  *
from
  datagen3
where
  id > 1;

CREATE TABLE print (
  id int,
  f1 int,
  f2 int,
  f3 int,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'print'
);

insert into print
select
  COALESCE(datagen1.id,datagen2.id,datagen3_view.id),
  datagen1.f1,
  datagen2.f2,
  datagen3_view.f3
from datagen1 
Full outer join datagen2 on datagen1.id = datagen2.id
Full outer join datagen3_view on datagen1.id = datagen3_view.id;

Parent topic: Enterprise-Class Enhancements of Flink

Previous topic: FlinkSQL Function Enhancements

Next topic: Flink O&M Management