Hologres:CREATE DYNAMIC TABLE

Example 1: Create a regular Incremental Dynamic Table

Before proceeding, import the tpch_10g public dataset into Hologres by following the guide at Import public datasets.

-- Create an Incremental Dynamic Table that refreshes every 3 minutes.
CREATE DYNAMIC TABLE public.tpch_q1_incremental 
WITH (
auto_refresh_mode='incremental',
freshness='3 minutes'
) AS SELECT
        l_returnflag,
        l_linestatus,
        COUNT(*) AS count_order
FROM
        hologres_dataset_tpch_10g.lineitem
WHERE
        l_shipdate <= DATE '1998-12-01' - INTERVAL '120' DAY
GROUP BY
        l_returnflag,
        l_linestatus;

Example 2: Create an Incremental Dynamic Table from stream-stream joins

Before proceeding, import the tpch_10g public dataset into Hologres by following the guide at Import public datasets.

-- Create an Incremental Dynamic Table from stream-stream joins.
CREATE DYNAMIC TABLE dt_join
  WITH (
    auto_refresh_mode='incremental',
    freshness='30 minutes'
  ) 
AS 
SELECT
        l_shipmode,
        SUM(CASE
                WHEN o_orderpriority = '1-URGENT'
                        OR o_orderpriority = '2-HIGH'
                        THEN 1
                ELSE 0
        END) AS high_line_count,
        SUM(CASE
                WHEN o_orderpriority <> '1-URGENT'
                        AND o_orderpriority <> '2-HIGH'
                        THEN 1
                ELSE 0
        END) AS low_line_count
FROM
        hologres_dataset_tpch_10g.orders,
        hologres_dataset_tpch_10g.lineitem
WHERE
        o_orderkey = l_orderkey
        AND l_shipmode IN ('FOB', 'AIR')
        AND l_commitdate < l_receiptdate
        AND l_shipdate < l_commitdate
        AND l_receiptdate >= DATE '1997-01-01'
        AND l_receiptdate < DATE '1997-01-01' + INTERVAL '1' YEAR
GROUP BY
        l_shipmode;

Example 3: Create an Auto-refresh Dynamic Table

Set the refresh mode to auto, allowing the engine to automatically select the refresh mode. It will prioritize incremental refresh and fall back to full refresh if incremental is not supported.

Before proceeding, import the tpch_10g public dataset into Hologres by following the guide at Import public datasets.

-- Create an Auto-refresh Dynamic Table that intelligent decides the refresh mode. In this example, incremental refresh is the result.
CREATE DYNAMIC TABLE thch_q6_auto
  WITH (
    auto_refresh_mode='auto',
    freshness='1 hours'
       ) 
AS
SELECT
        SUM(l_extendedprice * l_discount) AS revenue
FROM
        hologres_dataset_tpch_100g.lineitem
WHERE
        l_shipdate >= DATE '1996-01-01'
        AND l_shipdate < DATE '1996-01-01' + INTERVAL '1' YEAR
        AND l_discount BETWEEN 0.02 - 0.01 AND 0.02 + 0.01
        AND l_quantity < 24;

Example 4: Create a logical partitioned Dynamic Table

For a real-time transaction dashboard, there's often a need for both near-real-time viewing of current data and correction of historical data, which requires an integrated real-time and offline analysis solution (see Business and data cognition). We typically use `Dynamic Table` logical partitions for this scenario. The approach is as follows:

• The base table is partitioned by day. The latest partition is written to by Flink in real-time/near-real-time, while historical partitions are written from MaxCompute.

• The Dynamic Table is created as a logical partitioned table. The latest two partitions are active and refreshed incrementally for near-real-time data analytics.

• Historical partitions are inactive and use full refresh. If the base table's historical partitions have been corrected or backfilled, they can be refreshed using a full refresh.

This example uses a public dataset from GitHub.

1. Prepare the base table.

   Use Flink to write the latest data to the base table. For detailed steps, see Integration of offline and real-time processing of data in GitHub public event datasets.

   DROP TABLE IF EXISTS gh_realtime_data;
   
   BEGIN;
   CREATE TABLE gh_realtime_data (
       id BIGINT,
       actor_id BIGINT,
       actor_login TEXT,
       repo_id BIGINT,
       repo_name TEXT,
       org_id BIGINT,
       org_login TEXT,
       type TEXT,
       created_at timestamp with time zone NOT NULL,
       action TEXT,
       iss_or_pr_id BIGINT,
       number BIGINT,
       comment_id BIGINT,
       commit_id TEXT,
       member_id BIGINT,
       rev_or_push_or_rel_id BIGINT,
       ref TEXT,
       ref_type TEXT,
       state TEXT,
       author_association TEXT,
       language TEXT,
       merged BOOLEAN,
       merged_at TIMESTAMP WITH TIME ZONE,
       additions BIGINT,
       deletions BIGINT,
       changed_files BIGINT,
       push_size BIGINT,
       push_distinct_size BIGINT,
       hr TEXT,
       month TEXT,
       year TEXT,
       ds TEXT,
       PRIMARY KEY (id,ds)
   )
   PARTITION BY LIST (ds);
   CALL set_table_property('public.gh_realtime_data', 'distribution_key', 'id');
   CALL set_table_property('public.gh_realtime_data', 'event_time_column', 'created_at');
   CALL set_table_property('public.gh_realtime_data', 'clustering_key', 'created_at');
   
   COMMENT ON COLUMN public.gh_realtime_data.id IS 'Event ID';
   COMMENT ON COLUMN public.gh_realtime_data.actor_id IS 'Event actor ID';
   COMMENT ON COLUMN public.gh_realtime_data.actor_login IS 'Event actor login name';
   COMMENT ON COLUMN public.gh_realtime_data.repo_id IS 'Repo ID';
   COMMENT ON COLUMN public.gh_realtime_data.repo_name IS 'Repo name';
   COMMENT ON COLUMN public.gh_realtime_data.org_id IS 'Repo organization ID';
   COMMENT ON COLUMN public.gh_realtime_data.org_login IS 'Repo organization name';
   COMMENT ON COLUMN public.gh_realtime_data.type IS 'Event type';
   COMMENT ON COLUMN public.gh_realtime_data.created_at IS 'Event time';
   COMMENT ON COLUMN public.gh_realtime_data.action IS 'Event action';
   COMMENT ON COLUMN public.gh_realtime_data.iss_or_pr_id IS 'Issue/pull_request ID';
   COMMENT ON COLUMN public.gh_realtime_data.number IS 'Issue/pull_request number';
   COMMENT ON COLUMN public.gh_realtime_data.comment_id IS 'Comment ID';
   COMMENT ON COLUMN public.gh_realtime_data.commit_id IS 'Commit ID';
   COMMENT ON COLUMN public.gh_realtime_data.member_id IS 'Member ID';
   COMMENT ON COLUMN public.gh_realtime_data.rev_or_push_or_rel_id IS 'Review/push/release ID';
   COMMENT ON COLUMN public.gh_realtime_data.ref IS 'Name of created/deleted resource';
   COMMENT ON COLUMN public.gh_realtime_data.ref_type IS 'Type of created/deleted resource';
   COMMENT ON COLUMN public.gh_realtime_data.state IS 'State of issue/pull_request/pull_request_review';
   COMMENT ON COLUMN public.gh_realtime_data.author_association IS 'Relationship between actor and repo';
   COMMENT ON COLUMN public.gh_realtime_data.language IS 'Programming language';
   COMMENT ON COLUMN public.gh_realtime_data.merged IS 'Whether merged';
   COMMENT ON COLUMN public.gh_realtime_data.merged_at IS 'Merge time';
   COMMENT ON COLUMN public.gh_realtime_data.additions IS 'Number of added lines';
   COMMENT ON COLUMN public.gh_realtime_data.deletions IS 'Number of deleted lines';
   COMMENT ON COLUMN public.gh_realtime_data.changed_files IS 'Number of changed files in pull request';
   COMMENT ON COLUMN public.gh_realtime_data.push_size IS 'Number of pushes';
   COMMENT ON COLUMN public.gh_realtime_data.push_distinct_size IS 'Number of distinct pushes';
   COMMENT ON COLUMN public.gh_realtime_data.hr IS 'Hour of event, e.g., 00 for 00:23';
   COMMENT ON COLUMN public.gh_realtime_data.month IS 'Month of event, e.g., 2015-10 for Oct 2015';
   COMMENT ON COLUMN public.gh_realtime_data.year IS 'Year of event, e.g., 2015';
   COMMENT ON COLUMN public.gh_realtime_data.ds IS 'Date of event, ds=yyyy-mm-dd';
   
   COMMIT;

2. Create a logical partitioned Dynamic Table.

   CREATE  DYNAMIC TABLE ads_dt_github_event
   LOGICAL PARTITION BY LIST(ds)
   WITH (
     -- Dynamic table properties
     freshness = '5 minutes', 
     auto_refresh_mode = 'auto', 
     auto_refresh_partition_active_time = '2 days' ,
     partition_key_time_format = 'YYYY-MM-DD'
   )
   AS
   SELECT
       repo_name,
       COUNT(*) AS events,
       ds
   FROM
       gh_realtime_data
   GROUP BY repo_name,ds

3. Query the Dynamic Table.

   SELECT * FROM ads_dt_github_event ;

4. Backfill a historical partition.

   If historical data in the base table changes (e.g., data for '2025-04-01'), and the Dynamic Table needs to be updated, set the historical partition to full refresh mode and trigger a refresh, preferably with Serverless Computing resources.

   REFRESH OVERWRITE DYNAMIC TABLE ads_dt_github_event
   PARTITION (ds = '2025-04-01') 
   WITH (
     refresh_mode = 'full'
   );

Example 5: Calculate UVs with Incremental Dynamic Table

Starting from Hologres V3.1, an Incremental Dynamic Table supports the RB_BUILD_AGG function for calculations like the number of UVs. Compared to pre-aggregation, this offers:

• Faster performance: Computes only incremental data.

• Lower cost: Reduced data volume and resource usage, enabling calculations over longer periods.

Example:

1. Prepare a user detail table.

   BEGIN;
   CREATE TABLE IF NOT EXISTS ods_app_detail (
        uid INT,
        country TEXT,
        prov TEXT,
        city TEXT,
        channel TEXT,
        operator TEXT,
        brand TEXT,
        ip TEXT,
        click_time TEXT,
        year TEXT,
        month TEXT,
        day TEXT,
        ymd TEXT NOT NULL
   );
   CALL set_table_property('ods_app_detail', 'orientation', 'column');
   CALL set_table_property('ods_app_detail', 'bitmap_columns', 'country,prov,city,channel,operator,brand,ip,click_time, year, month, day, ymd');
   -- Set distribution_key based on query needs for optimal sharding effect.
   CALL set_table_property('ods_app_detail', 'distribution_key', 'uid');
   -- For a field with full date-time used in WHERE filters, setting it as clustering_key and event_time_column is recommended.
   CALL set_table_property('ods_app_detail', 'clustering_key', 'ymd');
   CALL set_table_property('ods_app_detail', 'event_time_column', 'ymd');
   COMMIT;

2. Calculate UV using an Incremental Dynamic Table.

   CREATE DYNAMIC TABLE ads_uv_dt
     WITH (
       freshness = '5 minutes', 
       auto_refresh_mode = 'incremental') 
     AS 
   SELECT  
    RB_BUILD_AGG(uid),
    country,
    prov,
    city,
    ymd,
   COUNT(1)
   FROM    ods_app_detail
   WHERE ymd >= '20231201' AND ymd <='20240502'
   GROUP BY country,prov,city,ymd;

3. Query UV for a specific day.

   SELECT  
   RB_CARDINALITY(RB_OR_AGG(rb_uid)) AS uv,
     country,
     prov,
     city,
     SUM(pv) AS pv
   FROM    ads_uv_dt
   WHERE   ymd = '20240329'
   GROUP BY country,prov,city;

Example 1: Create an auto-starting full-refresh dynamic table

Before proceeding, import the tpch_10g public dataset into Hologres by following the guide at Import public datasets.

--Create "test" Schema
CREATE SCHEMA test;

--Create a single-table full refresh dynamic table, starting immediately and refreshing every hour.
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_10g.lineitem
WHERE
        l_shipdate <= DATE '1998-12-01' - INTERVAL '120' DAY
GROUP BY
        l_returnflag,
        l_linestatus;

Example 2: Create an incremental dynamic table with a start time

Before proceeding, import the tpch_10g public dataset into Hologres by following the guide at Import public datasets.

Example:

--Enable binlog for the base table:
BEGIN;
CALL set_table_property('hologres_dataset_tpch_10g.lineitem', 'binlog.level', 'replica');
COMMIT;

--Create a single-table incremental refresh dynamic table, specifying a start time and a 3-minute refresh interval.
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_10g.lineitem
WHERE
        l_shipdate <= DATE '1998-12-01' - INTERVAL '120' DAY
GROUP BY
        l_returnflag,
        l_linestatus
;

Example 3: Create a multi-join full-refresh dynamic table

--Create a dynamic table with a multi-table join query, using full refresh mode every 3 hours.
CREATE DYNAMIC TABLE 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
        hologres_dataset_tpch_10g.orders
WHERE
        o_orderdate >= DATE '1996-07-01'
        AND o_orderdate < DATE '1996-07-01' + INTERVAL '3' MONTH
        AND EXISTS (
                SELECT
                        *
                FROM
                        hologres_dataset_tpch_10g.lineitem
                WHERE
                        l_orderkey = o_orderkey
                        AND l_commitdate < l_receiptdate
        )
GROUP BY
        o_orderpriority;

Example 4: Create a dimension-join incremental dynamic table

Example:

The semantics of a dimension table JOIN are that each record is joined only with the latest version of the dimension table data at that time, i.e., the JOIN occurs at processing time. If the dimension table data changes (add, update, or delete) after the JOIN, the already joined data is not updated. SQL example:

--Detail table
BEGIN;
CREATE TABLE public.sale_detail(
        app_id TEXT,
        uid TEXT,
        product TEXT,
        gmv BIGINT,
        order_time TIMESTAMPTZ
);
--Enable binlog for the base table; dimension tables do not require it.
CALL set_table_property('public.sale_detail', 'binlog.level', 'replica');
COMMIT;

--Property table
CREATE TABLE public.user_info(
        uid TEXT,
        province TEXT,
        city TEXT
);

CREATE DYNAMIC TABLE public.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 public.sale_detail 
INNER JOIN public.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;

Example 5: Create a partitioned dynamic table

For a real-time transaction dashboard, there's often a need for both near-real-time viewing of current data and correction of historical data. This can be achieved using a combination of `Dynamic Table` incremental and full refresh. The approach is as follows:

1. Create a partitioned base table where the latest partition is written in real-time/near-real-time, and historical partitions are occasionally corrected.

2. Create a `Dynamic Table` as a partitioned parent table. Use incremental refresh for the latest partition to meet near-real-time analysis needs.

3. Switch historical partitions to full refresh mode. If the source table's historical partitions have been corrected, the `Dynamic Table`'s partitions can also be backfilled using full refresh, preferably with Serverless to speed it up.

Example:

1. Prepare the base table and data.

   The base table is a partitioned table, with the latest partition receiving real-time data.

   -- Create a partitioned source table
   CREATE TABLE base_sales(
     uid INT,
     opreate_time TIMESTAMPTZ,
     amount FLOAT,
     tt TEXT NOT NULL,
     ds TEXT,
     PRIMARY KEY(ds)
   ) PARTITION BY LIST (ds) ;
   
   --Historical partition
   CREATE TABLE base_sales_20240615 PARTITION OF base_sales FOR VALUES IN ('20240615');
   INSERT INTO base_sales_20240615 VALUES (2,'2024-06-15 16:18:25.387466+08','111','2','20240615');
   
   --Latest partition, typically for real-time writes
   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. Create a partitioned `Dynamic Table` parent table, defining only the query without a refresh mode.

   --Create extension
   CREATE EXTENSION roaringbitmap;
   
   CREATE DYNAMIC TABLE partition_dt_base_sales
   PARTITION BY LIST (ds)
   as
   SELECT  
    public.RB_BUILD_AGG(uid),
    opreate_time,
    amount,
    tt,
    ds,
   COUNT(1)
   FROM    base_sales
   GROUP BY opreate_time ,amount,tt,ds;

3. Create sub-tables and set their refresh modes.

   You can create `Dynamic Table` sub-partitions manually or dynamically using DataWorks. Set the latest partition to incremental refresh and historical partitions to full refresh.

   -- Enable Binlog for the base table
   ALTER TABLE base_sales SET (binlog_level = replica);
   
   -- Assume the historical Dynamic Table sub-partition is as follows:
   CREATE DYNAMIC TABLE partition_dt_base_sales_20240615 PARTITION OF partition_dt_base_sales FOR VALUES IN ('20240615')
     WITH (
       refresh_mode='incremental',
       auto_refresh_enable='true',
       incremental_auto_refresh_schd_start_time='immediate',
       incremental_auto_refresh_interval='30 minutes'
          );
   
   -- Create a new Dynamic Table sub-partition, set its refresh mode to incremental, start immediately, refresh every 30 minutes, and use instance resources.
   CREATE DYNAMIC TABLE partition_dt_base_sales_20240616 PARTITION OF partition_dt_base_sales FOR VALUES IN ('20240616')
     WITH (
       refresh_mode='incremental',
       auto_refresh_enable='true',
       incremental_auto_refresh_schd_start_time='immediate',
       incremental_auto_refresh_interval='30 minutes'
          );
   
   --Switch the historical partition to full refresh mode
   ALTER DYNAMIC TABLE partition_dt_base_sales_20240615 SET (refresh_mode = 'full');
   --If historical partition data needs correction, execute a refresh, preferably with serverless.
   SET hg_computing_resource = 'serverless';
   REFRESH DYNAMIC TABLE partition_dt_base_sales_20240615;