Clickzetta Table Stream Pipeline

Table Stream Change Data Capture Workflow

Step 1: Enable Change Tracking on Source Table (required prerequisite)

ALTER TABLE <source_table> SET PROPERTIES ('change_tracking' = 'true');

This is a mandatory prerequisite — without it, the Stream cannot capture changes correctly.

Verify it took effect:

-- Method 1
DESC EXTENDED <source_table>;

-- Method 2
SELECT table_name, properties FROM information_schema.tables WHERE table_name = '<source_table>';

Step 2: Create Table Stream

CREATE [ OR REPLACE ] TABLE STREAM <stream_name>
  ON TABLE <source_table>
  [ TIMESTAMP AS OF '<timestamp>' ]
  [ COMMENT '<description>' ]
  WITH PROPERTIES (
    'TABLE_STREAM_MODE' = 'STANDARD | APPEND_ONLY',
    'SHOW_INITIAL_ROWS' = 'TRUE | FALSE'
  );

Mode selection:

SHOW_INITIAL_ROWS:

• TRUE: first consumption returns rows that existed in the table when the Stream was created
• FALSE (default): first consumption returns only changes after Stream creation

TIMESTAMP AS OF specifies the Stream's starting read position. This feature may be unstable in some scenarios — prefer the default behavior (start from creation time).

Step 3: Prepare Target Table

Create a target table compatible with the source table structure. Add metadata columns for tracking:

CREATE TABLE <target_table> (
  -- source business columns
  order_id INT, status STRING, amount DOUBLE,
  -- optional tracking columns
  sync_version BIGINT, sync_timestamp TIMESTAMP
);

Step 4: Preview Stream Data (does not advance offset)

SELECT *, __change_type, __commit_version, __commit_timestamp
FROM <stream_name>;

Metadata fields:

• __change_type: INSERT / UPDATE_BEFORE / UPDATE_AFTER / DELETE
• __commit_version: internal version number
• __commit_timestamp: when the change was committed

UPDATE handling: UPDATE produces two records — UPDATE_BEFORE (old value, usually ignored) and UPDATE_AFTER (new value, write to target). Always filter __change_type to avoid writing UPDATE_BEFORE stale values to the target table.

Step 5: Consume Stream Data (advances offset)

Option A: Full consumption (INSERT INTO)

INSERT INTO <target_table>
SELECT <columns> FROM <stream_name>;

Option B: Idempotent consumption (MERGE — recommended)

MERGE INTO <target_table> t
USING (SELECT * FROM <stream_name> WHERE __change_type != 'UPDATE_BEFORE') s
ON t.<pk_column> = s.<pk_column>
WHEN MATCHED AND s.__change_type IN ('INSERT', 'UPDATE_AFTER') THEN UPDATE SET t.col1 = s.col1, t.col2 = s.col2
WHEN MATCHED AND s.__change_type = 'DELETE' THEN DELETE
WHEN NOT MATCHED AND s.__change_type = 'INSERT' THEN INSERT (<columns>) VALUES (s.<columns>);

MERGE is preferred because it's idempotent — re-running won't create duplicates. Filter out UPDATE_BEFORE in the USING subquery to prevent stale values from interfering with MERGE logic.

⚠️ MERGE clause order: when multiple WHEN MATCHED clauses exist, UPDATE must come before DELETE, otherwise: update statement must be before delete statement

Step 6: Verify Consumption

SELECT COUNT(*) FROM <stream_name>;
-- After successful consumption, COUNT should be 0 or contain only new changes

Record the last consumed __commit_version for fault recovery.

Offset Advancement Rules

⚠️ Offset advancement is all-or-nothing. Once a DML consumes the Stream, all change records advance — partial consumption is not possible. If DML fails (e.g. target table doesn't exist), offset does not advance.

Mode Selection Quick Reference

Performance Tips

• Select only necessary columns, avoid SELECT *
• Consume Stream regularly to prevent data accumulation
• High change rate tables: consume more frequently; low change rate: reduce frequency
• Large Streams: split by primary key range for parallel processing
• Set appropriate data retention period on source table

Examples

Example 1: Orders table real-time sync

-- 1. Enable change tracking
ALTER TABLE orders SET PROPERTIES ('change_tracking' = 'true');

-- 2. Create Stream
CREATE TABLE STREAM orders_stream ON TABLE orders
WITH PROPERTIES ('TABLE_STREAM_MODE' = 'STANDARD', 'SHOW_INITIAL_ROWS' = 'FALSE');

-- 3. Create target table
CREATE TABLE orders_sync (order_id INT, status STRING, amount DOUBLE);

-- 4. Preview (no offset advance)
SELECT *, __commit_version, __commit_timestamp FROM orders_stream;

-- 5. Consume (advances offset)
MERGE INTO orders_sync t
USING (SELECT * FROM orders_stream WHERE __change_type != 'UPDATE_BEFORE') s
ON t.order_id = s.order_id
WHEN MATCHED AND s.__change_type IN ('INSERT', 'UPDATE_AFTER') THEN UPDATE SET t.status = s.status, t.amount = s.amount
WHEN MATCHED AND s.__change_type = 'DELETE' THEN DELETE
WHEN NOT MATCHED AND s.__change_type = 'INSERT' THEN INSERT (order_id, status, amount) VALUES (s.order_id, s.status, s.amount);

-- 6. Verify
SELECT COUNT(*) FROM orders_stream;

Example 2: User behavior audit (retain all insert history)

ALTER TABLE user_actions SET PROPERTIES ('change_tracking' = 'true');

CREATE TABLE STREAM user_actions_audit_stream ON TABLE user_actions
WITH PROPERTIES ('TABLE_STREAM_MODE' = 'APPEND_ONLY', 'SHOW_INITIAL_ROWS' = 'TRUE');

INSERT INTO user_actions_audit
SELECT *, __commit_version AS audit_version, __commit_timestamp AS audit_time
FROM user_actions_audit_stream;

Troubleshooting