【译】FLIP_132 Temporal Table DDL and Temporal Table Join

-- 版本表建表语句示例
CREATE TABLE versioned_rates (
    currency STRING,
    rate DECIMAL(38, 10),
    currency_time TIMESTAMP(3),
    WATERMARK FOR currency_time AS currency_time, -- event time  (currency_time)
    PRIMARY KEY(currency) NOT ENFORCED            -- primary key (currency)
) WITH (
...
)

CREATE TABLE latest_rates (  -- No event time and No primary key
    currency STRING,
    rate DECIMAL(38, 10),
    currency_time TIMESTAMP(3)      
) WITH (
...
)

-- 订单事实表
CREATE TABLE orders (
    order_id STRING,
    currency STRING,
    amount INT,
    proctime as PROCTIME(),                 
    order_time TIMESTAMP(3),                
    WATERMARK FOR order_time AS order_time - INTERVAL '30' SECOND
) WITH (
)

-- Event-time 时态表 join
SELECT 
  o.oder_id,
   o.order_time,
  o.amount * r.rate AS amount,
  r.currency
FROM orders AS o, versioned_rates FOR SYSTEM_TIME AS OF o.order_time r
on o.currency = r.currency;

-- Processing-time 时态表 join
SELECT 
  o.oder_id,
  o.proctime,
  o.amount * r.rate AS amount,
  r.currency
FROM orders AS o, latest_rates FOR SYSTEM_TIME AS OF o.proctime r
on o.currency = r.currency

-- 以下都是版本表/视图

-- changelog 流中的延迟删除事件将被忽略，因为DELETE事件中的事件时间戳通常小于水印。下面的changelog流来自数据库binglog，包含DELETE事件，这些事件中的事件时间戳通常小于水印
-- 如果changelog源包含删除事件（versioned_rates2），建议使用 changelog time 作为事件时间
CREATE TABLE versioned_rates1 (
    currency STRING,
    rate DECIMAL(38, 10),
    currency_time TIMESTAMP(3),
    WATERMARK FOR currency_time AS currency_time, -- event time  (currency_time)
    PRIMARY KEY(currency) NOT ENFORCED            -- primary key (currency)
) WITH (
  'format' = 'debezium-json'/'canal-json'         -- changelog source
)
-- 版本表“versioned_rates2”是从包含主键和事件时间（changelog time）的changelog流生成的，该表包含多个版本
-- 与“versioned_rates1”相比，versioned表“versioned_rates2”中的版本是按changelog时间（即binlog时间）划分的，后者更适合处理changelog的延迟删除事件
-- Changelog time 是系统中发生更改的时间（即数据库中DML执行的系统时间），DELETE event 中的事件时间将是理想情况下不小于水印的执行时间
-- 如果changelog源包含DELETE事件，建议使用 changelog_time 作为事件时间
CREATE TABLE versioned_rates2 (
    currency STRING,
    rate DECIMAL(38, 10),
    currency_time TIMESTAMP(3),
    changelog_time TIMESTAMP(3) as SYSTEM_METADATA("db_operation_time"),
    WATERMARK FOR changelog_time AS changelog_time, -- event time  (changelog_time)
    PRIMARY KEY(currency) NOT ENFORCED              -- primary key  (currency)
) WITH (
  'format' = 'debezium-json'/'canal-json'           -- changelog source
)
-- 版本表“versioned_rates3” 是从仅附加的流中生成的，该流包含主键和事件时间（currency_time）
-- 该表是一个版本表，但每个记录只包含一个版本
-- 因为append only流中的主键意味着记录不会在主键上重复，所以只有一个版本
CREATE TABLE versioned_rates3 (
    currency STRING,
    rate DECIMAL(38, 10),
    currency_time TIMESTAMP(3),
    WATERMARK FOR currency_time AS currency_time, -- event time
    PRIMARY KEY(currency) NOT ENFORCED            -- primary key
) WITH (
  'format' = 'json'                               -- append only source
)
-- 版本表“versioned_rates4”将仅追加的源转换为changelog流，它包含推断的主键并保留事件时间属性
-- 该表也是一个版本表，每个记录都包含多个版本
-- 如果仅附加源包含业务键上的重复记录，建议使用“versioned_rates4”视图定义版本表
CREATE TABLE rates (
    currency STRING,
    rate DECIMAL(38, 10),
    currency_time TIMESTAMP(3),
    WATERMARK FOR currency_time AS currency_time   -- event time
) WITH (
  'format' = 'json'                               -- append only source
)
-- 该视图返回带有推断主键（Currency）和事件时间（Currency_time）的有序changelog流
CREATE VIEW versioned_rates4 AS               
SELECT currency, rate, currency_time
  FROM (
      SELECT *,
      ROW_NUMBER() OVER (PARTITION BY currency ORDER BY currency_time DESC) AS rowNum
      FROM rates )
WHERE rowNum = 1;

CREATE TABLE latest_rates1 ( -- No event time and No primary key
    currency STRING,
    rate DECIMAL(38, 10) 
) WITH (
)

CREATE VIEW latest_rates2 AS -- No event time and No primary key
SELECT currency, rate 
FROM rates;                 

CREATE TABLE latest_rates3 ( -- No primary key
    currency STRING,
    rate DECIMAL(38, 10),
    currency_time TIMESTAMP(3),
    WATERMARK FOR currency_time AS currency_time, 
) WITH (
)

CREATE TABLE latest_rates4 ( -- No event time
    currency STRING,
    rate DECIMAL(38, 10),
    proctime AS PROCTIME(),
    PRIMARY KEY(currency) NOT ENFORCED 
) WITH (
)

CREATE VIEW latest_rates5 AS -- No event time and only has a inferred primary key
SELECT currency, rate, proctime
  FROM (
      SELECT *,
      ROW_NUMBER() OVER (PARTITION BY currency ORDER BY proctime DESC)
      AS rowNum FROM Rates)
WHERE rowNum = 1;

-- The proposed Temporal Table DDL way, create a versioned table
CREATE VIEW versioned_rates AS
SELECT currency, rate, currency_time
FROM (
      SELECT *,
      ROW_NUMBER() OVER (PARTITION BY currency ORDER BY currency_time  DESC) AS rowNum
      FROM rates)
WHERE rowNum = 1;


SELECT 
  o.oder_id,
  o.order_time,
  o.amount * r.rate AS amount,
  r.currency
FROM orders AS o, versioned_rates FOR SYSTEM_TIME AS OF o.order_time r
on o.currency = r.currency;