#Flink SQL

CREATE TABLE IF NOT EXISTS source_table (
	`user_id` BIGINT,
	`name` STRING,
	`order_time` TIMESTAMP(3),
	`cost` AS price * quantity,  -- evaluate expression and supply the result to queries
	`record_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp',    -- reads and writes a Kafka record's timestamp
	`timestamp` TIMESTAMP_LTZ(3) METADATA,    -- 如果和 metadata 中字段名一致，可不写 from xxx
	`partition` BIGINT METADATA,     -- part of the query-to-sink schema，即可读取也可写入
	`offset` BIGINT METADATA VIRTUAL, -- not part of the query-to-sink schema，只能读取不可写入
	`proctime` AS PROCTIME(),
	WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND,
	PRIMARY KEY (`partition`, `offset`) NOT ENFORCED
)
PARTITIONED BY (`partition`) 
WITH (
	'connector' = 'kafka'
	...
)

-- INNER JOIN 案例

CREATE TABLE show_log_table (
    log_id BIGINT,
    show_params STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '2',
  'fields.show_params.length' = '1',
  'fields.log_id.min' = '1',
  'fields.log_id.max' = '10'
);

CREATE TABLE click_log_table (
  log_id BIGINT,
  click_params     STRING
)
WITH (
  'connector' = 'datagen',
  'rows-per-second' = '2',
  'fields.click_params.length' = '1',
  'fields.log_id.min' = '1',
  'fields.log_id.max' = '10'
);

CREATE TABLE sink_table (
    s_id BIGINT,
    s_params STRING,
    c_id BIGINT,
    c_params STRING
) WITH (
  'connector' = 'print'
);

INSERT INTO sink_table
SELECT
    show_log_table.log_id as s_id,
    show_log_table.show_params as s_params,
    click_log_table.log_id as c_id,
    click_log_table.click_params as c_params
FROM show_log_table
INNER JOIN click_log_table ON show_log_table.log_id = click_log_table.log_id;

-- Inner Join、Left Join、Right Join、Full Join 案例
SELECT *
FROM Orders
LEFT JOIN Product
ON Orders.product_id = Product.id

SELECT *
FROM Orders
RIGHT JOIN Product
ON Orders.product_id = Product.id

SELECT *
FROM Orders
FULL OUTER JOIN Product
ON Orders.product_id = Product.id

CREATE TABLE show_log_table (
	product_id STRING,
	unique_id BIGINT,
	event_time BIGINT,
	row_time AS TO_TIMESTAMP_LTZ(event_time, 3),
	WATERMARK FOR row_time AS row_time - INTERVAL '5' SECONDS
) WITH (
	'connector' = 'kafka',
	...
);

CREATE TABLE click_log_table (
	product_id STRING,
	unique_id BIGINT,
	event_time BIGINT,
	row_time AS TO_TIMESTAMP_LTZ(event_time, 3),
	WATERMARK FOR row_time AS row_time - INTERVAL '5' SECONDS
) WITH (
	'connector' = 'kafka',
	...
);

-- INNER JOIN
SELECT 
	show_table.unique_id AS unique_id,
	show_table.product_id AS product_id,
	show_table.event_time AS event_time,
	click_table.event_time AS click_event_time
FROM 
(
	SELECT * 
	FROM TABLE(TUMBLE(TABLE show_log_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES))
) show_table
INNER JOIN 
(
	SELECT * 
	FROM TABLE(TUMBLE(TABLE click_log_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES))
) click_table
ON show_table.unique_id = click_table.unique_id
AND show_table.window_start = click_table.window_start
AND show_table.window_end = click_table.window_end;

-- SEMI JOIN，也可用 EXISTS
SELECT 
	show_table.unique_id AS unique_id,
	show_table.product_id AS product_id,
	show_table.event_time AS event_time
FROM 
(
	SELECT * 
	FROM TABLE(TUMBLE(TABLE show_log_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES))
) show_table
WHERE
	show_table.product_id IN (
			SELECT * FROM (
					SELECT * 
					FROM TABLE(TUMBLE(TABLE click_log_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES))
				) click_table
			WHERE
				show_table.window_start = click_table.window_start
				AND show_table.window_end = click_table.window_end
		);
-- ANTI JOIN，也可用 NOT EXISTS

-- 使用 PROCTIME() 例子
CREATE TABLE show_log_table (
    log_id BIGINT,
    show_params STRING,
    proctime AS PROCTIME()
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.show_params.length' = '1',
  'fields.log_id.min' = '1',
  'fields.log_id.max' = '10'
);

CREATE TABLE click_log_table (
    log_id BIGINT,
    click_params STRING,
    proctime AS PROCTIME()
)
WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.click_params.length' = '1',
  'fields.log_id.min' = '1',
  'fields.log_id.max' = '10'
);

CREATE TABLE sink_table (
    s_id BIGINT,
    s_params STRING,
    c_id BIGINT,
    c_params STRING
) WITH (
  'connector' = 'print'
);

INSERT INTO sink_table
SELECT
    show_log_table.log_id as s_id,
    show_log_table.show_params as s_params,
    click_log_table.log_id as c_id,
    click_log_table.click_params as c_params
FROM show_log_table INNER JOIN click_log_table ON show_log_table.log_id = click_log_table.log_id
AND show_log_table.proctime BETWEEN click_log_table.proctime - INTERVAL '4' HOUR AND click_log_table.proctime;

-- 使用 event_time，watermark
CREATE TABLE show_log_table (
    log_id BIGINT,
    show_params STRING,
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    WATERMARK FOR row_time AS row_time
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.show_params.length' = '1',
  'fields.log_id.min' = '1',
  'fields.log_id.max' = '10'
);

CREATE TABLE click_log_table (
    log_id BIGINT,
    click_params STRING,
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    WATERMARK FOR row_time AS row_time
)
WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.click_params.length' = '1',
  'fields.log_id.min' = '1',
  'fields.log_id.max' = '10'
);

CREATE TABLE sink_table (
    s_id BIGINT,
    s_params STRING,
    c_id BIGINT,
    c_params STRING
) WITH (
  'connector' = 'print'
);

INSERT INTO sink_table
SELECT
    show_log_table.log_id as s_id,
    show_log_table.show_params as s_params,
    click_log_table.log_id as c_id,
    click_log_table.click_params as c_params
FROM show_log_table INNER JOIN click_log_table ON show_log_table.log_id = click_log_table.log_id
AND show_log_table.row_time BETWEEN click_log_table.row_time - INTERVAL '4' HOUR AND click_log_table.row_time;

-- 维表关联。维表只能用 PROCTIME() 查询。
CREATE TABLE show_log (
    log_id BIGINT,
    `timestamp` as cast(CURRENT_TIMESTAMP as timestamp(3)),
    user_id STRING,
    proctime AS PROCTIME()
)
WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10000000',
  'fields.user_id.length' = '1',
  'fields.log_id.min' = '1',
  'fields.log_id.max' = '10'
);

CREATE TABLE user_profile (
    user_id STRING,
    age STRING,
    sex STRING
    ) WITH (
  'connector' = 'redis',
  'hostname' = '127.0.0.1',
  'port' = '6379',
  'format' = 'json',
  'lookup.cache.max-rows' = '500',
  'lookup.cache.ttl' = '3600',
  'lookup.max-retries' = '1'
);

CREATE TABLE sink_table (
    log_id BIGINT,
    `timestamp` TIMESTAMP(3),
    user_id STRING,
    proctime TIMESTAMP(3),
    age STRING,
    sex STRING
) WITH (
  'connector' = 'print'
);

INSERT INTO sink_table
SELECT 
    s.log_id as log_id
    , s.`timestamp` as `timestamp`
    , s.user_id as user_id
    , s.proctime as proctime
    , u.sex as sex
    , u.age as age
FROM show_log AS s
LEFT JOIN user_profile FOR SYSTEM_TIME AS OF s.proctime AS u
ON s.user_id = u.user_id