CREATE CATALOG my_catalog WITH (
  'type' = 'paimon',
  'warehouse' = 'hdfs://HDFS14979/usr/hive/warehouse'
);

CREATE CATALOG my_hive WITH (
  'type' = 'paimon',
  'metastore' = 'hive',
  -- 'uri' = 'thrift://<hive-metastore-host-name>:<port>', -- 默认使用 HiveConf 中的 ‘hive.metastore.uris’
  -- 'warehouse' = 'hdfs://HDFS14979/usr/hive/warehouse', -- 默认使用 HiveConf 中的 ‘hive.metastore.warehouse.dir’
);

-- 创建主键表
CREATE TABLE my_table (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) WITH (
    'bucket' = '10',
    -- 其他 with 参数
);

-- 创建分区表
CREATE TABLE my_table (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) PARTITIONED BY (dt, hh) WITH (
    'bucket' = '10',
    -- 其他 with 参数
);


-- 创建 logger sink
CREATE TABLE logger_sink (
    id BIGINT,
    name STRING
) WITH (
  'connector' = 'logger',
  'print-identifier' = 'DebugData'
);

-- 创建 paimon catalog
CREATE CATALOG paimon_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'hdfs://HDFS14979/usr/hive/warehouse'
);

-- 从 paimon 读取数据写入 logger sink
INSERT INTO logger_sink 
SELECT * FROM paimon_catalog.test_db.user_info;


-- 创建 datagen source
CREATE TABLE datagen_source (
    id BIGINT,
    name STRING
) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '10'
);

-- 创建 paimon catalog
CREATE CATALOG paimon_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'hdfs://HDFS14979/usr/hive/warehouse'
);

-- 创建 paimon 库和表（存在则不创建）
CREATE DATABASE IF NOT EXISTS paimon_catalog.test_db;
CREATE TABLE IF NOT EXISTS paimon_catalog.test_db.user_info (
    id BIGINT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'bucket' = '2'
);

-- datagen 生成数据写入 paimon 表
INSERT INTO paimon_catalog.test_db.user_info 
SELECT * FROM datagen_source;


containerized.taskmanager.env.HADOOP_USER_NAME: hadoop
containerized.master.env.HADOOP_USER_NAME: hadoop