建表語法

Dynamic Table的建表語法如下：

CREATE DYNAMIC TABLE [IF NOT EXISTS] <schema.tablename>(
[col_name],
[col_name]
  ) [PARTITION BY LIST (col_name)]
WITH (
    [refresh_mode='[full|incremental]',]
    [auto_refresh_enable='[true|false',]

  --增量刷新專用參數(shù)：
    [incremental_auto_refresh_schd_start_time='[immediate|<timestamptz>]',]
    [incremental_auto_refresh_interval='[<num> {minute|minutes|hour|hours]',]  
    [incremental_guc_hg_computing_resource='[ local | serverless]',]
    [incremental_guc_hg_experimental_serverless_computing_required_cores='<num>',]

   --全量刷新專用參數(shù)：
    [full_auto_refresh_schd_start_time='[immediate|<timestamptz>]',]
    [full_auto_refresh_interval='[<num> {minute|minutes|hour|hours]',] 
    [full_guc_hg_computing_resource='[ local | serverless]',]--hg_full_refresh_computing_resource默認serverless，可以db級別設(shè)置，用戶可以不設(shè)置
    [full_guc_hg_experimental_serverless_computing_required_cores='<num>',]
    
   --共用參數(shù)，允許設(shè)置guc：
   [refresh_guc_<guc>='xxx]',] 
   
  -- Dynamic Table的通用屬性：
    [orientation = '[column | row | row,column]',]
    [table_group = '[tableGroupName]',]
    [distribution_key = 'columnName[,...]]',]
    [clustering_key = '[columnName{:asc]} [,...]]',]
    [event_time_column = '[columnName [,...]]',]
    [bitmap_columns = '[columnName [,...]]',]
    [dictionary_encoding_columns = '[columnName [,...]]',]
    [time_to_live_in_seconds = '<non_negative_literal>',]
    [storage_mode = '[hot | cold]']
    ) 
AS
<query> --query的定義

參數(shù)說明

增量刷新

對于已經(jīng)存在的基表，開啟Binlog并為其創(chuàng)建增量Dynamic Table后，基表中已經(jīng)存在的數(shù)據(jù)不會被同步到Dynamic Table中，只有新增的數(shù)據(jù)會被同步到Dynamic Table，因此可能會存在數(shù)據(jù)一致性問題。

全增量一體刷新

當前增量Dynamic Table也支持全增量一體消費，即首先消費基表的全量數(shù)據(jù)，再消費基表新增的數(shù)據(jù)，該功能需要通過GUC來控制：

-- 打開全增量數(shù)據(jù)一體消費的GUC
SET hg_experimental_enable_hybrid_incremental_mode = true;

示例如下：

--全增量一體化增量Dynamic table
-- 未開Binlog的基表
CREATE TABLE base_sales(
  day TEXT NOT NULL,
  hour INT,
  user_id BIGINT,
  ts TIMESTAMPTZ,
  amount FLOAT,
  pk text NOT NULL PRIMARY key
);

-- 導(dǎo)入數(shù)據(jù)
INSERT INTO base_sales VALUES ('2024-08-29',1,222222,'2024-08-29 16:41:19.141528+08',5,'ddd');


-- 為基表打開Binlog
ALTER TABLE base_sales SET (binlog_level = replica);

-- 再為基表導(dǎo)入增量數(shù)據(jù)
INSERT INTO base_sales VALUES ('2024-08-29',2,3333,'2024-08-29 17:44:19.141528+08',100,'aaaaa');


-- 創(chuàng)建增量Dynamic Table
CREATE DYNAMIC TABLE sales_incremental
  WITH (refresh_mode='incremental') 
AS 
  SELECT day, hour, SUM(amount), COUNT(1) 
    FROM base_sales 
  GROUP BY day, hour;

-- 打開全增量數(shù)據(jù)一體消費的GUC,然后刷新dynamic table
SET hg_experimental_enable_hybrid_incremental_mode = true;
REFRESH TABLE sales_incremental;

對比數(shù)據(jù)一致性：

• 查詢基表

  SELECT day, hour, SUM(amount), COUNT(1) 
      FROM base_sales 
    GROUP BY day, hour;

  返回結(jié)果：

  day	    hour	sum	count
  2024-08-29	1	5	1
  2024-08-29	2	100	1

• 查詢Dynamic Table

  SELECT * FROM sales_incremental;

  返回結(jié)果：

  day	    hour	sum	count
  2024-08-29	1	5	1
  2024-08-29	2	100	1

示例1：創(chuàng)建全量刷新Dynamic Table并自動開始執(zhí)行刷新

執(zhí)行下述操作前，請先參考一鍵導(dǎo)入公共數(shù)據(jù)集將tpch_10g公共數(shù)據(jù)集的數(shù)據(jù)導(dǎo)入至Hologres。

--創(chuàng)建單表全量刷新的dynamic table，并立即開始刷新，每1小時刷新一次。
CREATE DYNAMIC TABLE test.thch_q1_full
  WITH (
    refresh_mode='full',
    auto_refresh_enable='true',
    full_auto_refresh_interval='1 hours',
    full_guc_hg_computing_resource='serverless',
    full_guc_hg_experimental_serverless_computing_required_cores='32'
       ) 
AS
  SELECT
        l_returnflag,
        l_linestatus,
        SUM(l_quantity) AS sum_qty,
        SUM(l_extendedprice) AS sum_base_price,
        SUM(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
        SUM(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
        AVG(l_quantity) AS avg_qty,
        AVG(l_extendedprice) AS avg_price,
        AVG(l_discount) AS avg_disc,
        COUNT(*) AS count_order
FROM
        hologres_dataset_tpch_10.lineitem.lineitem
WHERE
        l_shipdate <= DATE '1998-12-01' - INTERVAL '120' DAY
GROUP BY
        l_returnflag,
        l_linestatus;

示例2：創(chuàng)建增量刷新的Dynamic Table并指定開始刷新時間

執(zhí)行下述操作前，請先參考一鍵導(dǎo)入公共數(shù)據(jù)集將tpch_10g公共數(shù)據(jù)集的數(shù)據(jù)導(dǎo)入至Hologres。

創(chuàng)建增量Dynamic Table示例如下：

--為基表開binlog：
BEGIN;
CALL set_table_property('hologres_dataset_tpch_100g.lineitem', 'binlog.level', 'replica');
COMMIT;

--創(chuàng)建單表贈量刷新的dynamic table，并指定開始刷新時間，每3min刷新一次
CREATE DYNAMIC TABLE public.tpch_q1_incremental 
WITH (
refresh_mode='incremental',
auto_refresh_enable='true',
incremental_auto_refresh_schd_start_time='2024-09-15 23:50:0',
incremental_auto_refresh_interval='3 minutes',
incremental_guc_hg_computing_resource='serverless',
incremental_guc_hg_experimental_serverless_computing_required_cores='30'
) AS SELECT
        l_returnflag,
        l_linestatus,
        COUNT(*) AS count_order
FROM
        hologres_dataset_tpch_100g.lineitem
WHERE
        l_shipdate <= DATE '1998-12-01' - INTERVAL '120' DAY
GROUP BY
        l_returnflag,
        l_linestatus
;

示例3：創(chuàng)建多表JOIN的全量刷新Dynamic Table

--創(chuàng)建query為多表join的dynamic table，全量刷新模式，每3小時刷新一次。
CREATE DYNAMIC TABLE test.dt_q_full
  WITH (
    refresh_mode='full',
    auto_refresh_enable='true',
    full_auto_refresh_schd_start_time='immediate'
    full_auto_refresh_interval='3 hours',
    full_guc_hg_computing_resource='serverless',
    full_guc_hg_experimental_serverless_computing_required_cores='64'
  ) 
AS 
SELECT
        o_orderpriority,
        COUNT(*) AS order_count
FROM
        public.orders
WHERE
        o_orderdate >= DATE '1996-07-01'
        AND o_orderdate < DATE '1996-07-01' + INTERVAL '3' MONTH
        AND EXISTS (
                SELECT
                        *
                FROM
                        public.lineitem
                WHERE
                        l_orderkey = o_orderkey
                        AND l_commitdate < l_receiptdate
        )
GROUP BY
        o_orderpriority;

示例4：創(chuàng)建維表JOIN的增量刷新Dynamic Table

創(chuàng)建維表JOIN的增量Dynamic Table示例如下：

維表JOIN的語義是：對每條數(shù)據(jù)，只會關(guān)聯(lián)當時維表的最新版本數(shù)據(jù)，即JOIN行為只發(fā)生在處理時間（Processing Time）。如果JOIN行為發(fā)生后，維表中的數(shù)據(jù)發(fā)生了變化（新增、更新或刪除），則已關(guān)聯(lián)的維表數(shù)據(jù)不會被同步變化。

--為基表開binlog，維表不需要開啟
BEGIN;
CALL set_table_property('hm.sale_detail ', 'binlog.level', 'replica');
COMMIT;


CREATE DYNAMIC TABLE dt_sales_incremental
  WITH (
    refresh_mode='incremental',
    auto_refresh_enable='true',
    incremental_auto_refresh_schd_start_time='2024-09-15 00:00:00'
    incremental_auto_refresh_interval='5 minutes',
    incremental_guc_hg_computing_resource='serverless',
    incremental_guc_hg_experimental_serverless_computing_required_cores='128') 
AS 
SELECT 
    sale_detail.app_id,
    sale_detail.uid,
    product,
    SUM(sale_detail.gmv) AS sum_gmv,
    sale_detail.order_time,
    user_info.province,
    user_info.city 
FROM hm.sale_detail 
INNER JOIN hm.user_info  FOR SYSTEM_TIME AS OF PROCTIME()
ON sale_detail.uid =user_info.uid
GROUP BY sale_detail.app_id,sale_detail.uid,sale_detail.product,sale_detail.order_time,user_info.province,user_info.city;

示例5：創(chuàng)建分區(qū)Dynamic Table

1. 準備基表和數(shù)據(jù)。

   -- 創(chuàng)建分區(qū)源表
   CREATE TABLE base_sales(
     uid INT,
     opreate_time TIMESTAMPTZ,
     amount FLOAT,
     pk text NOT NULL,
     ds text,
     PRIMARY key(ds)
   ) PARTITION BY LIST (ds) ;
   
   CREATE TABLE base_sales_20240616 PARTITION OF base_sales FOR VALUES IN ('20240616');
   
   INSERT INTO base_sales_20240616 VALUES (1,'2024-06-16 16:08:25.387466+08','2','1','20240616');

2. 創(chuàng)建分區(qū)Dynamic Table父表，父表只設(shè)置Query的定義，不設(shè)置刷新模式。

   CREATE DYNAMIC TABLE test.partition_dt_base_sales
   PARTITION BY LIST (ds)
   AS
   SELECT  
    public.RB_BUILD_AGG(uid),
    opreate_time,
    amount,
    pk,
    ds,
   COUNT(1)
   FROM    base_sales
   GROUP BY opreate_time ,amount,pk,ds;

3. 手動創(chuàng)建分區(qū)子表，并為分區(qū)子表設(shè)置刷新模式。

   -- 創(chuàng)建Dynamic Table分區(qū)子表,并指定刷新模式為全量刷新，建表成功后立即開始啟動刷新，刷新間隔為30mins，并使用本實例資源刷新。
   CREATE DYNAMIC TABLE test.partition_dt_base_sales_20240616 PARTITION OF partition_dt_base_sales FOR VALUES IN ('20240616')
     WITH (
       refresh_mode='full',
       auto_refresh_enable='true',
       full_auto_refresh_schd_start_time='immediate',
       full_auto_refresh_interval='30 minutes'
          );