CREATE TABLE `user_dim` (
  `id` BIGINT,
  `dt` STRING,
  `name` STRING,
  `age` INT,
  `address` STRING,
  PRIMARY KEY (`id`, `dt`) NOT ENFORCED
) WITH (
  'connector' = 'setats-cdc',
  'warehouse' = 'cosn://<bucket>/warehouse',
  'catalog-name' = 'setats',
  'catalog-type' = 'hive',
  'uri' = 'thrift://<metastore-host-1>:7004,thrift://<metastore-host-2>:7004',
  'catalog-database' = 'testdb',
  'catalog-table' = 'user_dim',
  'manager-url' = '<Manager Url>',
  'lookup.force-local' = 'true',
  'lookup.bucket.discover.interval' = '10s',
  'lookup.refresh.async' = 'false',
  'lookup.table.store-type' = 'ROCKSDB',
  'lookup.table.store-cache-size' = '1024'
);


-- Create a streaming table
CREATE TABLE `order_stream` (
  `order_id` BIGINT,
  `user_id` BIGINT,
  `dt` STRING,
  `amount` DECIMAL(10, 2),
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'kafka'
  -- Omit other Kafka parameters
);
-- Create a Setats dimension table
CREATE TABLE `setats_user_info_lookup` (
  `user_id` BIGINT,
  `user_name` STRING,
  `user_level` INT,
  `city` STRING,
  PRIMARY KEY (`user_id`) NOT ENFORCED
) WITH (
  'connector' = 'setats-cdc',
  'warehouse' = 'cosn://<bucket>/warehouse',
  'catalog-name' = 'setats',
  'catalog-type' = 'hive',
  'uri' = 'thrift://<metastore-host-1>:7004,thrift://<metastore-host-2>:7004',
  'catalog-database' = 'testdb',
  'catalog-table' = 'user_info',
  'manager-url' = '<Manager Url>',
  'lookup.force-local' = 'true',
  'lookup.bucket.discover.interval' = '10s',
  'lookup.refresh.async' = 'false',
  'lookup.table.store-type' = 'ROCKSDB',
  'lookup.table.store-cache-size' = '1024'
);
-- Execute the Lookup Join
SELECT
  o.order_id,
  o.amount,
  u.user_name,
  u.user_level,
  u.city
FROM `order_stream` AS o
JOIN `setats_user_info_lookup` FOR SYSTEM_TIME AS OF o.proctime AS u
ON o.user_id = u.user_id;