Clickzetta SQL Pipeline Manager

ClickZetta SQL Pipeline Manager

See references/scenarios.md for worked pipeline examples (Kafka, CDC, MV, operations, parameterized DT). See references/troubleshooting.md for common errors and delivery checklist. See references/dynamic-table.md, references/pipe.md, references/table-stream.md, references/materialized-view.md for object-specific syntax.

⚠️ Key Syntax Differences: ClickZetta vs Standard SQL / Snowflake

Wizard: Clarify Intent

On receiving a request, determine the user's intent:

A. Design and create a new pipeline (complete SQL from source to layered DTs) → Pipeline Wizard below B. Manage existing objects (modify refresh interval, suspend/resume, view history) → Execute directly C. Troubleshoot (DT refresh failure, Pipe stopped, Stream backlog) → Read references/troubleshooting.md

If the user has already stated clearly what they want, execute directly without asking.

Pipeline Wizard

Triggered by: "design/build ETL", "complete data pipeline", "ingest from Kafka/OSS", "ODS→DWD→DWS", "Medallion Architecture", "Bronze/Silver/Gold".

Layer naming

Preserve the user's chosen naming — do not remap Bronze→ODS, Silver→DWD, etc.:

Always add a project/domain prefix to schema names to avoid conflicts. If the user hasn't provided one, ask.

Requirements collection

If the user has already provided enough information, generate the complete SQL directly. Otherwise ask two questions: (1) data source type (Kafka / OSS/S3/COS / existing table INSERT-only / existing table with UPDATE/DELETE); (2) refresh frequency (seconds / minutes / hours/days).

Generate complete SQL

Produce end-to-end SQL covering all parts:

1. Schema creation (using user's layer names)
2. Ingestion layer table (if external source)
3. Data entry point (Pipe or Table Stream, based on source)
4. Intermediate layer Dynamic Table (cleanse/filter)
5. Service layer Dynamic Table (aggregate/dimension)
6. REFRESH DYNAMIC TABLE immediately after each DT creation
7. Verification commands

Source → entry object:

• Kafka → CREATE PIPE ... AS COPY INTO ... FROM (SELECT ... FROM read_kafka(...))
• OSS/S3/COS → CREATE PIPE ... INGEST_MODE = 'LIST_PURGE' AS COPY INTO ... FROM VOLUME ... PURGE=true
• Existing table + UPDATE/DELETE → CREATE TABLE STREAM ... WITH PROPERTIES ('TABLE_STREAM_MODE' = 'STANDARD')
• Existing table + INSERT only → Dynamic Table reads source directly

After generating SQL, save each segment as a Studio task (code asset management):

# DDL → DRAFT task (no Cron, no dependencies — DDL must never be scheduled)
cz-cli task save-content <ddl_task_name> --content "<ddl_sql>"

# ETL/transform → scheduled task
cz-cli task save-content <etl_task_name> --content "<etl_sql>"
cz-cli task save-cron <etl_task_name> --cron '0 30 2 * * ? *'
cz-cli task deploy <etl_task_name>

Scheduling rules (hard constraints):

Object Type Reference

Decision Tree

User requirement
├── Continuously ingest from external source (Kafka / OSS / S3)
│   └── → Pipe
├── Real-time / incremental transform on existing table
│   ├── Need to detect UPDATE/DELETE → Table Stream + Dynamic Table
│   └── INSERT append only → Dynamic Table (reads source directly)
├── Fixed aggregation, real-time not required
│   └── → Materialized View
└── Multi-layer ETL (ODS→DWD→DWS or Bronze→Silver→Gold)
    └── → Cascaded Dynamic Tables (each layer sets its own REFRESH INTERVAL)

Routing