1.Temporal Joins（时态JOIN）

时态表是一个随时间演变的表，在Flink中也称为动态表。

时态表中的行与一个或多个时态周期相关联，并且所有Flink表都是时态的（动态的）。时态表包含一个或多个版本化的表快照，它可以是跟踪更改的更改历史表（例如数据库更改日志，包含所有快照），也可以是具体化更改的维表（例如包含最新快照的数据库表）。

时态表可以分为版本表和普通表。

版本表：如果时态表中的记录可以追踪和并访问它的历史版本，这种表我们称之为版本表，来自数据库的 changelog （如mysql binlog）可以定义成版本表，版本表内的数据始终不会自动清理，只能通过upsert触发。
普通表：如果时态表中的记录仅仅可以追踪并和它的最新版本，这种表我们称之为普通表，来自数据库或 HBase 、redis的表可以定义成普通表。

特征：

只支持INNER JOIN、LEFT JOIN
只有左流触发更新
输出流保留时间属性

时态join类型

JOIN Lookup
JOIN 版本表

2.语法

使用FOR SYSTEM_TIME AS OF table1.proctime表示当左边表的记录与右边的维表join时，只匹配当前处理时间维表所对应的的快照数据（即关联维表当前最新的状态）

SELECT [column_list]
FROM table1 [AS <alias1>]
[LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.{ proctime | rowtime } [AS <alias2>]
ON table1.column-name1 = table2.column-name1

注意：Temporary table（临时表）和Temporal table（时态表）是两个不同概念。Temporary table是临时的表对象，属于当前Session，随着Session的结束而消失，该表不属于Catalog和DB。

3.JOIN Lookup

概念：Lookup join是针对于由作业流表触发，关联右侧维表来补全数据的场景。默认情况下，在流表有数据变更，都会触发维表查询（可以通过设置维表是否缓存，来减轻查询压力），由于不保存状态，因此对内存占用较小。

特性：

左侧为流表、右侧为维表
流表需要指定处理时间
具备lookup能力的外部系统
自己实现LookupTableSource接口connector

示例：

-- 维表  
CREATE TEMPORARY TABLE users (
  `user_id` STRING,
  `name` STRING,
  `age` INT,
  `gmt_time` TIMESTAMP(3)
) WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://localhost:3306/flink',
   'table-name' = 'user',
   'username' = 'root',
   'password' = '123456'
);

-- 流表
CREATE TABLE orders (
    order_id    STRING,
    price       DECIMAL(32,2),
    user_id    STRING,
    order_time  TIMESTAMP(3),
    proctime AS PROCTIME()) WITH (
  'connector' = 'kafka',
  'topic' = 'order',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json'
);

 -- 使用FOR SYSTEM_TIME AS OF table1.proc_time表示当左边表的记录与右边的维表join时，只匹配当前处理时间维表所对应的的快照数据（即关联维表当前最新的状态）
select orders.order_id,orders.price,orders.order_time,c.name  
FROM orders 
 LEFT  JOIN users FOR SYSTEM_TIME AS OF orders.proctime  AS c 
ON orders.user_id = c.user_id;

4.JOIN 版本表

版本表是可以追溯数据历史版本的表，如：数据库changelog，数据源有：mysql-binlog、kafka-upsert、oracle-cdc等。需要具备事件时间和主键两个属性。

特性：

与双流join不同，尽管构建端发生了更改，但之前的临时表结果不会受到影响。与间隔join相比，时态表join没有定义join记录的时间窗口。左侧表的记录总是在时间属性指定的时间与右侧表的版本连接。因此，构建端的行可能任意陈旧。

左侧为流表、右侧为版本表
两侧表都需要指定事件时间
版本表的数据会持续增加

满足场景：

左输入表为流表，右输入表为版本表（ Changelog 动态表，即 Upsert、Retract 数据流，而非 Append 数据流）
两侧表都需要设置watermark，版本表需要设置主键，主键必须包含在 JOIN 等值条件中
版本表发生变更，不会触发查询结果输出，会根据主键更新临时表

示例：

用户在下订单时，需要根据订单时间的汇率，计算订单金额，其中下单是以不同的货币，需要将他输出到特定货币（CNY）

# 订单表（普通表）
CREATE TABLE orders (
    order_id    STRING,
    price       DECIMAL(32,2),
    currency    STRING,
    order_time   timestamp(3),
    WATERMARK FOR order_time AS order_time
) WITH (
  'connector' = 'kafka',
  'topic' = 'order',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'orders2ConsumerGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json'
);


-- 汇率表 （版本表）

CREATE TABLE currency_rates (
    currency STRING,
    conversion_rate DECIMAL(32, 2),
    update_time TIMESTAMP(3),
    WATERMARK FOR update_time AS update_time,
    PRIMARY KEY(currency) NOT ENFORCED
) WITH  (
  'connector' = 'kafka',
  'topic' = 'currency_rates',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'currencyRatesGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'debezium-json',
  'debezium-json.schema-include' = 'true'
);

select o.order_id,o.price,o.order_time,c.currency  
FROM orders AS o 
LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF o.order_time  AS c 
ON o.currency = c.currency;

-- 汇率表(版本视图)

CREATE TABLE ratesHistory (
    currency STRING,
    conversion_rate DECIMAL(32, 2),
    update_time TIMESTAMP(3),
    WATERMARK FOR update_time AS update_time
) WITH  (
  'connector' = 'kafka',
  'topic' = 'currency_rates',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'currencyRatesGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json'
);

CREATE VIEW versionedRates AS
SELECT currency,conversion_rate,update_time 
FROM (
	SELECT * ,ROW_NUMBER() OVER(PARTITION BY currency
                              ORDER BY update_time DESC) AS rowNum 
  FROM ratesHistory)
where rowNum=1;

select o.order_id,o.price,o.order_time,c.currency  
FROM orders AS o 
LEFT JOIN versionedRates FOR SYSTEM_TIME AS OF o.order_time  AS c 
ON o.currency = c.currency;