FlinkSQL 几种join

SELECT columns
FROM t1  [AS <alias1>]
[LEFT/INNER/FULL OUTER] JOIN t2
ON t1.column1 = t2.key-name1

--写法1
SELECT columns
FROM t1  [AS <alias1>]
[LEFT/INNER/FULL OUTER] JOIN t2
ON t1.column1 = t2.key-name1 AND t1.timestamp BETWEEN t2.timestamp  AND  BETWEEN t2.timestamp + + INTERVAL '10' MINUTE;
--写法2
SELECT columns
FROM t1  [AS <alias1>]
[LEFT/INNER/FULL OUTER] JOIN t2
ON t1.column1 = t2.key-name1 AND t2.timestamp <= t1.timestamp and t1.timestamp <=  t2.timestamp + + INTERVAL ’10' MINUTE ;

CREATE TABLE `Order`
(
  id         INT,
  product_id INT,
  quantity   INT,
  order_time TIMESTAMP(3),
  WATERMARK FOR order_time AS order_time,
   PRIMARY KEY (id) NOT ENFORCED
) WITH (
     'connector' = 'datagen',
     'fields.id.kind' = 'sequence',
     'fields.id.start' = '1',
     'fields.id.end' = '100000',
     'fields.product_id.min' = '1',
     'fields.product_id.max' = '100',
     'rows-per-second' = '1'
); 


CREATE TABLE `Product`
(
  id          INT,
  name        VARCHAR,
  price       DOUBLE,
  record_time TIMESTAMP(3),
  WATERMARK FOR record_time AS record_time,
   PRIMARY KEY (id) NOT ENFORCED
) WITH (
     'connector' = 'datagen',
     'fields.id.min' = '1',
     'fields.id.max' = '100',
     'rows-per-second' = '1'
);


CREATE TABLE `OrderDetails`
(
  id           INT,
  product_name VARCHAR,
  total_price  DOUBLE,
  order_time   TIMESTAMP
) WITH (
     'connector' = 'print'
);




INSERT INTO `OrderDetails`
SELECT o.id, p.name, o.quantity * p.price, o.order_time
FROM `Order` o,
    `Product` p
WHERE o.product_id = p.id
 AND o.order_time BETWEEN p.record_time - INTERVAL '5' MINUTE AND p.record_time;

-- 定义一张版本表
CREATE TABLE product_changelog (
  product_id STRING,
  product_name STRING,
  product_price DECIMAL(10, 4),
  update_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,
  PRIMARY KEY(product_id) NOT ENFORCED,      -- (1) 定义主键约束
  WATERMARK FOR update_time AS update_time   -- (2) 通过 watermark 定义事件时间              
) WITH (
  'connector' = 'kafka',
  'topic' = 'products',
  'scan.startup.mode' = 'earliest-offset',
  'properties.bootstrap.servers' = 'localhost:9092',
  'value.format' = 'debezium-json'
);

CREATE VIEW versioned_rates AS              
SELECT currency, rate, currency_time            -- (1) `currency_time` 保留了事件时间
  FROM (
      SELECT *,
      ROW_NUMBER() OVER (PARTITION BY currency  -- (2) `currency` 是去重 query 的 unique key，可以作为主键
         ORDER BY currency_time DESC) AS rowNum 
      FROM RatesHistory )
WHERE rowNum = 1; 

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


# 例如：
SELECT * FROM LatestRates FOR SYSTEM_TIME AS OF TIME '10:15';


currency   rate
======== ======
US Dollar   102
Euro        114
Yen           1

CREATE TABLE orders (
    order_id    STRING,
    price       DECIMAL(32,2),
    total       DECIMAL(32,2),
    currency    STRING,
    order_time  TIMESTAMP(3),
    proc_time as PROCTIME()
) WITH (/* ... */);
-- Customers is backed by the JDBC connector and can be used for lookup joins
CREATE TEMPORARY TABLE Customers (
  id INT,
  name STRING,
  country STRING,
  zip STRING
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://mysqlhost:3306/customerdb',
  'table-name' = 'customers'
);


-- 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;