This documentation is for an unreleased version of Apache Flink Table Store. We recommend you use the latest stable version.

Lookup Join #

A Lookup Join is used to enrich a table with data that is queried from Flink Table Store. The join requires one table to have a processing time attribute and the other table to be backed by a lookup source connector.

First, create a table, and update it in real-time.

-- Create a table store catalog
CREATE CATALOG my_catalog WITH (
  'type'='table-store',
  'warehouse'='hdfs://nn:8020/warehouse/path' -- or 'file://tmp/foo/bar'
);

USE CATALOG my_catalog;

-- Create a table in table-store catalog
CREATE TABLE customers (
  id INT PRIMARY KEY NOT ENFORCED,
  name STRING,
  country STRING,
  zip STRING
);

-- Launch a streaming job to update customers table
INSERT INTO customers ...

-- Create a temporary left table, like from kafka
CREATE TEMPORARY TABLE Orders (
  order_id INT,
  total INT,
  customer_id INT,
  proc_time AS PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic' = '...',
  'properties.bootstrap.servers' = '...',
  'format' = 'csv'
  ...
);

Then, you can use this table in lookup join.

-- enrich each order with customer information
SELECT o.order_id, o.total, c.country, c.zip
FROM Orders AS o
         JOIN customers FOR SYSTEM_TIME AS OF o.proc_time AS c
              ON o.customer_id = c.id;

The lookup join node will maintain the rocksdb cache locally and pull the latest updates of the table in real time, and only pull the necessary data. Therefore, your filter conditions are also very important.

In order to avoid excessive use of local disks, the lookup join feature is only suitable for table sizes below tens of millions.

Note: Partitioned or non-pk tables are not supported now.

Project pushdown can effectively reduce the overhead, FLINK-29138 fixed the bug that the project cannot be pushed down to the source. So it is preferable to use a version greater than or equal to flink 1.14.6, flink 1.15.3 and flink 1.16.0. Or you can cherry-pick the commit to your own Flink branch.

RocksDBOptions #

Options for rocksdb cache, you can configure options in WITH or dynamic table hints.

SELECT o.order_id, o.total, c.country, c.zip
  FROM Orders AS o JOIN customers /*+ OPTIONS('lookup.cache-rows'='20000') */
  FOR SYSTEM_TIME AS OF o.proc_time AS c ON o.customer_id = c.id;