Overview

Healthcare Appointments Demo Dataset

IMPORTANT - Demo Data Disclaimer: All patient data, doctor information, and healthcare records in this demo are entirely fictitious and created for demonstration purposes only. No real PHI (Protected Health Information) is used.

What is Openflow

What is Change Data Capture (CDC)

What You Will Learn

What You Will Build

Prerequisites

Openflow Deployment: This quickstart uses Openflow Snowflake Deployment (Snowflake-managed). If you're using Openflow BYOC deployment, the connector setup and concepts are the same, but runtime deployment steps will differ.

Clone the QuickStart Repository

git clone https://github.com/Snowflake-Labs/sfguide-getting-started-openflow-postgresql-cdc.git
cd sfguide-getting-started-openflow-postgresql-cdc

Setup PostgreSQL Database

PostgreSQL Requirements

Required PostgreSQL Configuration:

• PostgreSQL instance available: You have a running PostgreSQL instance (Snowflake Postgres, AWS RDS, GCP Cloud SQL, Azure Database, or self-hosted) with network accessibility
• Logical replication enabled: wal_level = logical (required for CDC) - see configuration instructions below
• User with REPLICATION privileges: The PostgreSQL user must have replication permissions
• Network access: Configure network access from Snowflake SPCS to your PostgreSQL instance
• Supported versions: PostgreSQL 11-17

This quickstart was tested with Snowflake Postgres v17.

Enable Logical Replication (If Not Already Enabled)

ALTER USER snowflake_admin WITH REPLICATION;

Important: For detailed configuration instructions specific to your PostgreSQL service, see Snowflake's PostgreSQL CDC Setup Guide - Configure wal_level.

Setup Notes

Setup PostgreSQL Client Environment

Don't have psql installed? The psql client is part of the PostgreSQL installation:

• macOS: brew install postgresql (via Homebrew)
• Ubuntu/Debian: sudo apt-get install postgresql-client
• RHEL/CentOS: sudo yum install postgresql
• Windows: Download from PostgreSQL Downloads

Using a different client? If you prefer pgAdmin, DBeaver, or another GUI client, you can adapt the psql commands throughout this guide by running the SQL portion in your client's query window.

Set Environment Variables

export PGHOST='YOUR-POSTGRES-HOST'        # e.g., your Snowflake Postgres endpoint
export PGPORT='5432'
export PGDATABASE='postgres'
export PGUSER='snowflake_admin'

Configure Password File

# Create .pgpass file (Unix/Linux/macOS)
echo "$PGHOST:$PGPORT:$PGDATABASE:$PGUSER:YOUR-PASSWORD" >> ~/.pgpass

# Set proper permissions (required for security)
chmod 0600 ~/.pgpass

Password File Format: The .pgpass file format is hostname:port:database:username:password. You can use * as a wildcard for any field. Learn more about PostgreSQL password files.

Test PostgreSQL Connection

Check PostgreSQL Accessibility

Test Basic Connection

psql -c "SELECT version();"

psql -h YOUR-POSTGRES-HOST -p 5432 -U snowflake_admin -d postgres -c "SELECT version();"

                                                 version                                                  
----------------------------------------------------------------------------------------------------------
 PostgreSQL 17.2 on x86_64-pc-linux-gnu, compiled by gcc (Debian 12.2.0-14) 12.2.0, 64-bit
(1 row)

Connection Failed? If you cannot connect to PostgreSQL, resolve this before proceeding. Common fixes:

• Check firewall: Ensure port 5432 is open
• Check PostgreSQL config: Verify listen_addresses in postgresql.conf
• Check pg_hba.conf: Ensure host-based authentication allows your connection
• Check cloud provider: Verify security groups, firewall rules, authorized networks
• Verify credentials: Ensure username and password are correct

All subsequent steps require a working PostgreSQL connection.

Verify psql Configuration (Optional)

psql -c "SELECT current_database(), current_user;"

 current_database | current_user 
------------------+--------------
 postgres         | snowflake_admin
(1 row)

Verify Logical Replication

Snowflake Postgres Advantage: If you're using Snowflake Postgres, logical replication (wal_level = logical) is enabled by default. You can skip this verification step, but feel free to confirm it works as expected.

psql -c "SHOW wal_level;"

 wal_level 
-----------
 logical
(1 row)

Initialize Healthcare Database

Execute the Initialization Script

Run the Script

psql -f sql/0.init_healthcare.sql

Using a different PostgreSQL client? If you're using pgAdmin, DBeaver, or another GUI client instead of psql, simply open sql/0.init_healthcare.sql, copy the entire contents, and execute it in your client's query window. The script contains standard PostgreSQL SQL and will work with any client.

Script Duration: Approximately 10-15 seconds to complete.

Verify PostgreSQL Setup

Using a different PostgreSQL client? All verification queries below use psql for consistency. If you're using a GUI client (pgAdmin, DBeaver, etc.), simply run the SQL portion of each command. For example, for psql -c "SELECT ...", just run SELECT ... in your client.

Check Tables Exist

psql -c "\dt healthcare.*"

              List of relations
   Schema   |     Name     | Type  |      Owner      
------------+--------------+-------+-----------------
 healthcare | appointments | table | snowflake_admin
 healthcare | doctors      | table | snowflake_admin
 healthcare | patients     | table | snowflake_admin
 healthcare | visits       | table | snowflake_admin
(4 rows)

Check Row Counts

psql -c "SELECT 
  (SELECT COUNT(*) FROM healthcare.patients) as patients,
  (SELECT COUNT(*) FROM healthcare.doctors) as doctors,
  (SELECT COUNT(*) FROM healthcare.appointments) as appointments,
  (SELECT COUNT(*) FROM healthcare.visits) as visits;"

 patients | doctors | appointments | visits 
----------+---------+--------------+--------
      100 |      10 |          170 |    100
(1 row)

What's a Publication? A PostgreSQL publication defines which tables are tracked for CDC. The Openflow connector subscribes to this publication to receive change events (INSERT, UPDATE, DELETE) in real-time.

Verify CDC Publication

psql -c "SELECT * FROM pg_publication WHERE pubname = 'healthcare_cdc_publication';"

          pubname           | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate 
----------------------------+----------+--------------+-----------+-----------+-----------+-------------
 healthcare_cdc_publication |    16390 | t            | t         | t         | t         | t
(1 row)

Verify Published Tables

psql -c "SELECT schemaname, tablename FROM pg_publication_tables WHERE pubname = 'healthcare_cdc_publication';"

 schemaname |  tablename   
------------+--------------
 healthcare | appointments
 healthcare | doctors
 healthcare | patients
 healthcare | visits
(4 rows)

Check CDC Configuration

psql -c "SELECT
    current_setting('wal_level') AS wal_level,
    current_setting('max_replication_slots') AS max_replication_slots,
    current_setting('max_wal_senders') AS max_wal_senders,
    (SELECT count(*) FROM pg_publication) AS publication_count,
    (SELECT count(*) FROM pg_replication_slots) AS active_slots;"

 wal_level | max_replication_slots | max_wal_senders | publication_count | active_slots 
-----------+-----------------------+-----------------+-------------------+--------------
 logical   | 10                    | 10              | 1                 | 0
(1 row)

What to check:

• wal_level should be logical
• max_replication_slots should be at least 1 (typically 10 or more)
• max_wal_senders should be at least 1 (typically 10 or more)
• publication_count should be 1 (our healthcare_cdc_publication)
• active_slots will be 0 now, and will increase to 1 once Openflow connects

Verify Replication Privileges

psql -c "SELECT rolname, rolsuper, rolreplication FROM pg_roles WHERE rolname = '$PGUSER';"

 rolname         | rolsuper | rolreplication 
-----------------+----------+----------------
 snowflake_admin | f        | t
(1 row)

Snowflake Postgres Advantage: If you're using Snowflake Postgres, the snowflake_admin user is pre-configured with replication privileges (rolreplication = t) and is not a superuser (rolsuper = f). This is the recommended secure configuration for CDC workloads.

Note: For other managed services (AWS RDS, Azure), rolsuper is typically f (false). For self-hosted installations where you're using the superuser account, it may be t (true). The critical column to verify is rolreplication which must be t (true).

psql -c "ALTER USER $PGUSER WITH REPLICATION;"

Setup Complete! Your PostgreSQL database is now fully configured for CDC with:

• ✅ Logical replication enabled
• ✅ Healthcare schema with 100 patients, 10 doctors, 170 appointments, 100 visits
• ✅ CDC publication tracking all healthcare tables
• ✅ Replication privileges granted

Setup Snowflake Environment

Create Snowflake Objects

What We're Creating

Execute the Setup Script

IMPORTANT: Before running the script, update line 64 with your PostgreSQL endpoint:

VALUE_LIST = ('YOUR-POSTGRES-HOST:5432');

Replace YOUR-POSTGRES-HOST with:

• Snowflake Postgres: Your Snowflake Postgres connection endpoint
• GCP Cloud SQL: Your instance's public IP (e.g., 34.123.45.67)
• AWS RDS: Your RDS endpoint hostname
• Azure Database: Your Azure PostgreSQL server name
• Self-hosted: Your server hostname or IP address

Verify Snowflake Setup

Verify Schemas

USE ROLE QUICKSTART_ROLE;
USE WAREHOUSE QUICKSTART_PGCDC_WH;
SHOW SCHEMAS IN DATABASE QUICKSTART_PGCDC_DB;

| created_on          | name                  | is_default | is_current | database_name      | owner            | comment | options | retention_time | owner_role_type |
|---------------------|-----------------------|------------|------------|--------------------|--------------------|---------|---------|----------------|-----------------|
| 2025-10-07 10:00:00 | INFORMATION_SCHEMA    | N          | N          | QUICKSTART_PGCDC_DB | ACCOUNTADMIN      |         |         | 1              | ROLE            |
| 2025-10-07 10:00:00 | NETWORKS              | N          | N          | QUICKSTART_PGCDC_DB | QUICKSTART_ROLE   |         |         | 1              | ROLE            |
| 2025-10-07 10:00:00 | PUBLIC                | Y          | N          | QUICKSTART_PGCDC_DB | QUICKSTART_ROLE   |         |         | 1              | ROLE            |

Verify Integration

SHOW INTEGRATIONS LIKE 'quickstart_pgcdc_access';

| name                      | type              | category          | enabled | comment                                    | created_on          |
|---------------------------|-------------------|-------------------|---------|--------------------------------------------|--------------------|
| quickstart_pgcdc_access   | EXTERNAL_ACCESS   | SECURITY          | true    | OpenFlow SPCS runtime access for PostgreSQL CDC | 2025-10-07 10:00:00 |

Verify Network Rule

⚠️ IMPORTANT - Snowflake Network Connectivity

For Openflow to connect to your PostgreSQL instance, two things must be true:

1. PostgreSQL must allow inbound connections from Snowflake: Configure your cloud provider's firewall/security groups to allow connections from Snowflake SPCS IP ranges (or use broad access for demo purposes)
2. Snowflake network rule must match your PostgreSQL host/port: The network rule created in 1.snowflake_setup.sql must specify the exact host and port of your PostgreSQL instance

This section verifies requirement #2. If you encounter connection errors when starting the Openflow connector later, you'll need to verify both requirements.

DESC NETWORK RULE QUICKSTART_PGCDC_DB.NETWORKS.postgres_network_rule;

| name                     | type       | mode   | value_list                     | comment |
|--------------------------|------------|--------|--------------------------------|---------|
| postgres_network_rule    | HOST_PORT  | EGRESS | ['your-actual-host.com:5432']  |         |

⚠️ Host Mismatch? If the value_list doesn't match your PostgreSQL host, you must recreate the network rule and external access integration with the correct host:

-- Drop existing objects
USE ROLE ACCOUNTADMIN;
DROP INTEGRATION IF EXISTS quickstart_pgcdc_access;
USE ROLE QUICKSTART_ROLE;
DROP NETWORK RULE IF EXISTS QUICKSTART_PGCDC_DB.NETWORKS.postgres_network_rule;

-- Create network rule with CORRECT PostgreSQL host
USE DATABASE QUICKSTART_PGCDC_DB;
USE SCHEMA NETWORKS;
CREATE OR REPLACE NETWORK RULE postgres_network_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('YOUR-CORRECT-POSTGRES-HOST:5432');  -- ⚠️ Update this!

-- Recreate external access integration
USE ROLE ACCOUNTADMIN;
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION quickstart_pgcdc_access
  ALLOWED_NETWORK_RULES = (QUICKSTART_PGCDC_DB.NETWORKS.postgres_network_rule)
  ENABLED = true
  COMMENT = 'OpenFlow SPCS runtime access for PostgreSQL CDC';

-- Grant to role
GRANT USAGE ON INTEGRATION quickstart_pgcdc_access TO ROLE QUICKSTART_ROLE;

-- Verify
USE ROLE QUICKSTART_ROLE;
SHOW INTEGRATIONS LIKE 'quickstart_pgcdc_access';

Note: Full end-to-end connectivity (Snowflake → PostgreSQL) will be tested when the Openflow connector starts. If you encounter connection errors during connector startup, verify:

1. PostgreSQL firewall/security groups allow inbound connections from Snowflake SPCS
2. Network rule configuration is correct (verified above)
3. PostgreSQL credentials are correct
4. PostgreSQL is running and accessible

Openflow Configuration

Complete Openflow Snowflake Deployments Setup

Create Runtime for PostgreSQL CDC

Verify External Access Integration

USE ROLE QUICKSTART_ROLE;

-- Verify integration exists and is enabled
SHOW INTEGRATIONS LIKE 'quickstart_pgcdc_access';
DESC INTEGRATION quickstart_pgcdc_access;

| name                     | type              | category | enabled | comment                                              | created_on          |
|--------------------------|-------------------|----------|---------|------------------------------------------------------|---------------------|
| quickstart_pgcdc_access  | EXTERNAL_ACCESS   | SECURITY | true    | OpenFlow SPCS runtime access for PostgreSQL CDC      | 2025-10-07 10:00:00 |

What's an External Access Integration? This allows your Openflow runtime to communicate with external services (your PostgreSQL database) from within Snowpark Container Services. The postgres_network_rule you created earlier defines which PostgreSQL endpoints are accessible.

Create Runtime via Snowsight

Runtime Creation Time: Runtime creation typically takes 3-5 minutes. The status will progress from CREATING → ACTIVE.

Verify Runtime Status

Troubleshooting: If the runtime status shows ERROR or stays in CREATING for more than 10 minutes:

• Verify the external access integration exists: SHOW INTEGRATIONS LIKE 'quickstart_pgcdc_access';
• Check that QUICKSTART_ROLE has all required grants: SHOW GRANTS TO ROLE QUICKSTART_ROLE;
• Review the Openflow Snowflake Deployments troubleshooting guide

Access the Runtime Canvas

Authentication: When opening the runtime canvas for the first time, you may be prompted to authenticate/authorize access. Accept these prompts to allow the canvas to access the runtime and required resources.

REFERENCE DOCUMENTATION:

For detailed information, see Create runtime in the official documentation.

Add PostgreSQL Connector

Openflow Database CDC Connectors

Supported Databases

Why Use Openflow for CDC

Add PostgreSQL Connector to Runtime

New Tab & Authentication: Adding the connector to your runtime will open a new runtime tab in your browser with the canvas showing the PostgreSQL processor group. If prompted, authenticate/authorize access to allow the canvas to access the runtime. You'll configure this connector in the next section.

Configure PostgreSQL Connector

Production Tip: For production deployments, use a SecretManagerParameterProvider (like AwsSecretsManagerParameterProvider) to store credentials in a secrets manager rather than directly in the connector configuration.

Configure PostgreSQL Source Parameters

Download PostgreSQL JDBC Driver

Access PostgreSQL Source Parameters

Connection URL Format: The PostgreSQL JDBC URL format is jdbc:postgresql://HOST:PORT/DATABASE.

• For Snowflake Postgres: Use your Snowflake Postgres endpoint (e.g., jdbc:postgresql://your-snowflake-postgres-endpoint:5432/postgres)
• For GCP Cloud SQL: Use the public IP address (e.g., jdbc:postgresql://34.123.45.67:5432/postgres)
• For AWS RDS: Use the endpoint hostname (e.g., jdbc:postgresql://mydb.abc123.us-east-1.rds.amazonaws.com:5432/postgres)
• For other services: Use your specific connection details

Upload JDBC Driver as Reference Asset

Note: The screenshot shows postgres as the PostgreSQL Username. Based on your setup, use the appropriate username (e.g., Snowflake Postgres uses snowflake_admin by default).

Configure PostgreSQL Destination Parameters

IMPORTANT: Ensure you have completed the Create Snowflake Objects section before proceeding. The PostgreSQL connector requires the database, warehouse, and role to be created in advance.

Access PostgreSQL Destination Parameters

Session Token Authentication: When using SNOWFLAKE_SESSION_TOKEN, the connector authenticates using the Openflow runtime's Snowflake session. This is the recommended approach for SPCS deployments as it eliminates the need to manage separate credentials.

Configure PostgreSQL Ingestion Parameters

Access PostgreSQL Ingestion Parameters

Note: The screenshot shows postgres as the PostgreSQL Username. Based on your setup, use the appropriate username (e.g., Snowflake Postgres uses snowflake_admin by default).

Note: The screenshot shows postgres as the PostgreSQL Username. Based on your setup, use the appropriate username (e.g., Snowflake Postgres uses snowflake_admin by default).

Configuration Notes:

• Table Regex Pattern: The regex healthcare\..* matches all tables in the healthcare schema. The backslash escapes the dot to match a literal period (e.g., healthcare.patients, healthcare.doctors, healthcare.appointments, healthcare.visits).
• Merge Task Schedule: The CRON expression * * * * * ? means "every second", providing near real-time CDC updates. For production workloads, adjust based on your latency requirements and cost considerations. Learn more about Quartz CRON expressions.

Verify Snapshot Load

Enable Services and Start Connector

Note: You may see an error message about "Snowflake Private Key Service" failing to enable. This error can be safely ignored since this quickstart uses SNOWFLAKE_SESSION_TOKEN authentication instead of private key authentication. The connector will function normally with session token authentication.

What's Happening? The connector performs two main operations:

1. Snapshot Load: Captures all existing data from PostgreSQL tables and loads it into Snowflake
2. CDC Initialization: Establishes the replication slot and begins listening for changes

The snapshot typically takes 1-2 minutes for the demo dataset. For larger datasets, the duration will be proportional to the data volume.

Verify Data in Snowflake

IMPORTANT: Identifier Quoting Rule

PostgreSQL creates lowercase unquoted identifiers (e.g., healthcare, appointments, patient_id). When Openflow replicates these to Snowflake, they are preserved as case-sensitive lowercase names.

In Snowflake, you MUST quote these identifiers in lowercase:

• ✅ Correct: USE SCHEMA "healthcare"; SELECT "patient_id", "status" FROM "appointments";
• ❌ Wrong: USE SCHEMA healthcare; SELECT patient_id, status FROM appointments;
• ❌ Wrong: SELECT PATIENT_ID, STATUS FROM APPOINTMENTS;

Why? Snowflake uppercases unquoted identifiers, so appointments becomes APPOINTMENTS, which won't match the actual table name appointments (lowercase). All queries in this quickstart follow this quoting convention.

Check Record Counts

USE ROLE QUICKSTART_ROLE;
USE DATABASE QUICKSTART_PGCDC_DB;
USE SCHEMA "healthcare";
USE WAREHOUSE QUICKSTART_PGCDC_WH;

SELECT 'patients' as table_name, COUNT(*) as record_count FROM "patients"
UNION ALL
SELECT 'doctors', COUNT(*) FROM "doctors"
UNION ALL
SELECT 'appointments', COUNT(*) FROM "appointments"
UNION ALL
SELECT 'visits', COUNT(*) FROM "visits"
ORDER BY table_name;

+----------------+--------------+
| TABLE_NAME     | RECORD_COUNT |
+----------------+--------------+
| APPOINTMENTS   |          170 |
| DOCTORS        |           10 |
| PATIENTS       |          100 |
| VISITS         |          100 |
+----------------+--------------+

Note: The initial snapshot loads 100 rows in "visits". Additional visit records are generated during the Live CDC demo, so this count will increase after running the steps in "Generate Live CDC Events".

Verify CDC Metadata

SELECT 
    COUNT(*) as total_rows,
    MIN(_SNOWFLAKE_INSERTED_AT) as earliest_inserted,
    MAX(_SNOWFLAKE_INSERTED_AT) as latest_inserted,
    COUNT(DISTINCT _SNOWFLAKE_INSERTED_AT) as unique_insert_timestamps
FROM "appointments";

+------------+---------------------------+---------------------------+---------------------------+
| TOTAL_ROWS | EARLIEST_INSERTED         | LATEST_INSERTED           | UNIQUE_INSERT_TIMESTAMPS  |
+------------+---------------------------+---------------------------+---------------------------+
|        170 | 2025-10-08 10:15:33.000   | 2025-10-08 10:15:45.000   |                        10 |
+------------+---------------------------+---------------------------+---------------------------+

CDC Metadata Columns: Openflow automatically adds the following metadata columns to track changes:

• _SNOWFLAKE_INSERTED_AT (TIMESTAMP): When the row was first inserted into Snowflake during the snapshot or initial CDC event
• _SNOWFLAKE_UPDATED_AT (TIMESTAMP): When the row was last updated by CDC processing (remains NULL for snapshot-only rows)
• _SNOWFLAKE_DELETED (BOOLEAN): Logical delete flag (FALSE for active rows, TRUE when a DELETE event is captured)

During the Live CDC demo, you'll see _SNOWFLAKE_UPDATED_AT populated when records are modified, and _SNOWFLAKE_DELETED set to TRUE for deleted records.

Verify Appointment Status Distribution

SELECT 
    "status",
    COUNT(*) as count,
    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 2) as percentage
FROM "appointments"
GROUP BY "status"
ORDER BY count DESC;

+-----------+-------+------------+
| STATUS    | COUNT | PERCENTAGE |
+-----------+-------+------------+
| completed |   100 |      58.82 |
| cancelled |    40 |      23.53 |
| scheduled |    15 |       8.82 |
| no_show   |    10 |       5.88 |
| confirmed |     5 |       2.94 |
+-----------+-------+------------+

Note on Data Variability: The exact counts and percentages will vary each time you run the initialization script because the appointment status is assigned using RANDOM() in PostgreSQL. Your actual output might show completed: 102, cancelled: 44, no_show: 4 or other values, which is completely normal. The important thing is that you see a mix of different statuses with completed being the most common (~60%), followed by cancelled (~25%).

Save Baseline Results: These snapshot results serve as your baseline for comparison. After running the CDC demo, you'll re-run this same query to see how the status distribution changed. Consider saving these results or taking a screenshot for easy side-by-side comparison later.

Verify Doctor Workload

SELECT 
    d."first_name" || ' ' || d."last_name" as doctor_name,
    d."specialization",
    d."department",
    COUNT(a."appointment_id") as total_appointments,
    SUM(CASE WHEN a."status" = 'completed' THEN 1 ELSE 0 END) as completed_appointments,
    SUM(CASE WHEN a."status" IN ('scheduled', 'confirmed') THEN 1 ELSE 0 END) as upcoming_appointments
FROM "doctors" d
LEFT JOIN "appointments" a ON d."doctor_id" = a."doctor_id"
GROUP BY d."doctor_id", d."first_name", d."last_name", d."specialization", d."department"
ORDER BY total_appointments DESC
LIMIT 5;

Run Additional Verification Queries

Snapshot Load Complete! If all queries return expected results, your PostgreSQL tables are successfully replicated to Snowflake. The connector is now listening for changes and will automatically capture any INSERT, UPDATE, or DELETE operations in real-time.

Snowflake Intelligence

Why Set Up Intelligence First? By asking baseline questions about your snapshot data now, you'll be able to compare results after the CDC demo and see exactly how the data changed. This creates a powerful before/after experience for understanding CDC!

What You'll Build

Prerequisites

Upload Semantic Model

NOTE: The semantic_models stage and snowflake_intelligence database were already created when you ran 1.snowflake_setup.sql. You just need to upload the semantic model file!

snow stage put semantic-models/healthcare_cdc_semantic_model.yaml @semantic_models \
  --database QUICKSTART_PGCDC_DB \
  --schema PUBLIC \
  --role QUICKSTART_ROLE \
  --overwrite

Create the Agent

IMPORTANT: Before creating the agent, ensure you are using the QUICKSTART_ROLE role in Snowsight. This ensures the agent is owned by the correct role and has proper access to resources. You can switch roles using the role selector in the top-right corner of Snowsight.

Access Agent Creation Interface

Configure Agent Basics

IMPORTANT: As you configure each section below (About, Tools, Orchestration, Access), remember to click "SAVE" after completing all configurations to ensure your changes are preserved.

"Show me the appointment status distribution"

"Which doctors have the most appointments?"

"What appointments were updated via CDC today?"

"Show revenue by doctor specialization"

"How many records were soft-deleted in the last 24 hours?"

Configure Agent Tools

Add Semantic Model

TIP: The semantic model includes definitions for all four tables (patients, doctors, appointments, visits) plus their relationships and CDC metadata columns (_Snowflake_INSERTED_AT,_Snowflake_UPDATED_AT, _Snowflake_DELETED).

Configure Orchestration

Whenever you can answer visually with a chart, always choose to generate a chart even if the user didn't specify to. 
Respond in the same language as the question wherever possible.
When querying appointments or visits, consider filtering out soft-deleted records (_SNOWFLAKE_DELETED = FALSE) unless the user specifically asks about deleted data.

Always provide specific metrics and counts when answering quantitative questions.
When showing CDC-related data, explain what the metadata columns mean (e.g., _SNOWFLAKE_UPDATED_AT indicates when the record was modified via CDC).
Focus on actionable insights for healthcare operations.

TIP: Use auto (default) to let Snowflake automatically select the best available model for your region and query type. Supported models include Claude 4.0, Claude 3.7, Claude 3.5, and GPT 4.1.

Set Access Controls

NOTE: All queries use the user's credentials. Ensure users have appropriate access to the QUICKSTART_PGCDC_DB."healthcare" schema and underlying tables.

Ask Baseline Questions (Snapshot Data)

Getting Started with Queries

"How many total appointments do we have?"

"Show me the appointment status distribution"

"Which doctors have the most appointments?"

"What's the total revenue from all visits?"

"Show me appointments by specialization"

"Are there any records with CDC updates yet?"

Baseline Established! You now have a clear picture of your snapshot data. Keep these numbers in mind - after running the CDC demo in the next section, you'll ask follow-up questions to see exactly what changed!

Generate Live CDC Events

What CDC Events Will Be Generated

Real CDC Patterns: This script demonstrates all major CDC event types you'll encounter in production: customer records being created (INSERT), order statuses changing (UPDATE), and old data being archived (DELETE). The timing is designed to align with Openflow's 1-minute sync interval!

Run CDC Demo Script

Option 1: Using psql Command Line

psql -f sql/3.live_appointments.sql

psql -h YOUR_POSTGRES_HOST \
     -p 5432 \
     -U postgres \
     -d postgres \
     -f sql/3.live_appointments.sql

Option 2: Copy and Paste in pgAdmin or PostgreSQL Client

Perfect Timing for CDC: The script includes 60-second pauses after key operations to align with Openflow's 1-minute CDC sync interval. This lets you:

• Switch to Snowflake between operations to observe changes appearing
• See multiple CDC sync cycles in action
• Understand the near real-time nature of CDC (not instant, but very fast!)

The script displays "⏳ Waiting 60 seconds for CDC sync..." messages to indicate when it's a good time to check Snowflake.

Expected Output

🕐 8:00 AM - New appointment requests coming in...
✅ 3 new appointments scheduled

🕐 8:15 AM - Front desk confirming today's appointments...
✅ Today's appointments confirmed

🕐 8:30 AM - Patients checking in for their appointments...
✅ 4 patients checked in

...

═══════════════════════════════════════════════════════════
📊 CDC Demo Summary - Changes Generated
═══════════════════════════════════════════════════════════

 activity                            | count 
-------------------------------------+-------
 New appointments created            |    10
 Appointments updated (status changes)|    20
 New visit records created           |     4
 Doctor records updated              |     1

✅ All CDC events have been generated!

Verify Changes in Snowflake

Tip - Refresh Openflow Canvas: You can also monitor CDC activity in the Openflow UI. Right-click on the canvas and select Refresh to see updated statistics (records processed, events captured, etc.) on the connector flows in real-time.

Check Updated Row Counts

USE ROLE QUICKSTART_ROLE;
USE DATABASE QUICKSTART_PGCDC_DB;
USE SCHEMA "healthcare";
USE WAREHOUSE QUICKSTART_PGCDC_WH;

SELECT 'patients' as table_name, COUNT(*) as record_count FROM "patients"
UNION ALL
SELECT 'doctors', COUNT(*) FROM "doctors"
UNION ALL
SELECT 'appointments', COUNT(*) FROM "appointments"
UNION ALL
SELECT 'visits', COUNT(*) FROM "visits"
ORDER BY table_name;

+----------------+--------------+
| TABLE_NAME     | RECORD_COUNT |
+----------------+--------------+
| appointments   |          180 | -- +10 new appointments (net: 12 inserts - 2 deletes)
| doctors        |           10 | -- unchanged
| patients       |          100 | -- unchanged
| visits         |          104 | -- +4 new visit records
+----------------+--------------+

Verify CDC Metadata Shows Updates

SELECT 
    COUNT(*) as total_records,
    SUM(CASE WHEN _SNOWFLAKE_DELETED THEN 1 ELSE 0 END) as deleted_records,
    COUNT(DISTINCT _SNOWFLAKE_INSERTED_AT) as distinct_insert_times,
    COUNT(DISTINCT _SNOWFLAKE_UPDATED_AT) as distinct_update_times
FROM "appointments";

+----------------+-----------------+-----------------------+-----------------------+
| TOTAL_RECORDS  | DELETED_RECORDS | DISTINCT_INSERT_TIMES | DISTINCT_UPDATE_TIMES |
+----------------+-----------------+-----------------------+-----------------------+
|           180  |               2 |                    2+ |                    2+ |
+----------------+-----------------+-----------------------+-----------------------+

SELECT 
    DATE_TRUNC('minute', _SNOWFLAKE_INSERTED_AT) as insert_minute,
    COUNT(*) as records_inserted
FROM "appointments"
GROUP BY insert_minute
ORDER BY insert_minute;

Query the Journal Tables

Journal Table Naming: The timestamp portion (e.g., 1759908563) will be different in your environment based on when tables are added to replication. The schema generation number starts at 1 and increments if the source table schema changes. For more details, see Track data changes in tables in the Snowflake documentation.

Find Your Journal Tables: You can see which journal tables have been created by running:

SHOW TABLES LIKE '%_JOURNAL_%' IN SCHEMA "healthcare";

This will list all journal tables that Openflow has created as CDC events occurred. The tables appear in order as different table types experience their first CDC event.

-- View recent CDC events for appointments
-- Replace the timestamp with your actual journal table name
SELECT 
    SEEN_AT as event_time,
    EVENT_TYPE,
    "PRIMARY_KEY__appointment_id" as appointment_id,
    "PAYLOAD__patient_id" as patient_id,
    "PAYLOAD__doctor_id" as doctor_id,
    "PAYLOAD__status" as status,
    "PAYLOAD__reason_for_visit" as reason_for_visit,
    "PAYLOAD__appointment_date" as appointment_date
FROM QUICKSTART_PGCDC_DB."healthcare"."appointments_JOURNAL_1759908563_1"
WHERE SEEN_AT >= DATEADD(minute, -10, CURRENT_TIMESTAMP)
ORDER BY SEEN_AT DESC
LIMIT 20;

Journal Table Structure: Journal tables track all CDC events with:

• EVENT_TYPE: The operation type - IncrementalInsertRows, IncrementalUpdateRows, or IncrementalDeleteRows
• SEEN_AT: Timestamp when the CDC event was captured
• PRIMARY_KEY__<column>: Primary key column(s) from the source table (quoted, lowercase)
• PAYLOAD__<column>: All data columns from the source table (quoted, lowercase)
• SF_METADATA: Additional Snowflake CDC metadata

SHOW TABLES LIKE '%_JOURNAL_%' IN SCHEMA QUICKSTART_PGCDC_DB."healthcare";

View Deleted Records

SELECT 
    "appointment_id",
    "patient_id",
    "status",
    "appointment_date",
    "reason_for_visit",
    _SNOWFLAKE_INSERTED_AT,
    _SNOWFLAKE_UPDATED_AT,
    _SNOWFLAKE_DELETED
FROM "appointments"
WHERE _SNOWFLAKE_DELETED = TRUE;

+-----------------+------------+------------+------------------+-----------------------------+---------------------------+---------------------------+----------------------+
| APPOINTMENT_ID  | PATIENT_ID | STATUS     | APPOINTMENT_DATE | REASON_FOR_VISIT            | _SNOWFLAKE_INSERTED_AT    | _SNOWFLAKE_UPDATED_AT     | _SNOWFLAKE_DELETED   |
+-----------------+------------+------------+------------------+-----------------------------+---------------------------+---------------------------+----------------------+
|             123 |         45 | cancelled  | 2024-08-15       | Annual physical examination | 2025-10-08 10:15:33.000   | 2025-10-08 15:30:22.000   | TRUE                 |
|             156 |         78 | cancelled  | 2024-09-01       | Routine checkup             | 2025-10-08 10:15:40.000   | 2025-10-08 15:30:22.000   | TRUE                 |
+-----------------+------------+------------+------------------+-----------------------------+---------------------------+---------------------------+----------------------+

Note on Deleted Record IDs: The specific appointment IDs deleted will be consistent across multiple runs of the demo because the DELETE query uses ORDER BY appointment_id to select the 2 oldest cancelled appointments deterministically. The actual IDs depend on which appointments were randomly assigned cancelled status during your PostgreSQL initialization.

Soft Deletes: Notice that deleted records are still queryable in Snowflake! The _SNOWFLAKE_DELETED flag marks them as deleted, but the data is preserved. This is crucial for:

• Audit trails and compliance
• Historical analysis
• Data recovery if needed

You can filter deleted records out of queries with WHERE _SNOWFLAKE_DELETED = FALSE or WHERE NOT _SNOWFLAKE_DELETED.

Compare Before and After Status Distribution

SELECT 
    "status",
    COUNT(*) as count,
    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 2) as percentage
FROM "appointments"
WHERE NOT _SNOWFLAKE_DELETED  -- Exclude deleted records
GROUP BY "status"
ORDER BY count DESC;

Real-time CDC in Action! The changes you made in PostgreSQL appeared in Snowflake within seconds. This demonstrates the power of CDC for real-time data integration and analytics.

---

Return to Snowflake Intelligence (Ask CDC Questions)

CDC Monitoring Questions

"How many appointments were updated via CDC in the last hour?"

"Show me all appointments that were modified today via CDC"

"Which appointments were soft-deleted?"

Note on Counts: The exact numbers you see may vary based on when you run the CDC script, how long it runs, and the timing of CDC sync intervals. For example, you might see "12 appointments were updated" or "30 appointments were updated" - both are correct depending on which CDC events have been synced at query time. The key is to verify that CDC events are being captured, not to match exact counts.

Compare Before and After

"What's the new appointment status distribution?"

"How many new visit records were created today?"

"Show me appointments that changed from scheduled to completed"

Operational Analytics

"Which doctors completed the most visits today?"

"What's today's total revenue from visits?"

"Show me all urgent appointments added today"

CDC Audit Trail

"Show me appointments with status changes in the last 2 hours?"

"How many total records have been modified via CDC?"

"What percentage of appointments were updated via incremental vs snapshot only?"

Advanced CDC Questions

"Show me the timeline of appointment status changes today"

"Which patient had the most appointment updates?"

"Compare snapshot-only vs CDC-updated appointment counts"

Business Insights

"What's the cancellation rate for today's appointments?"

"Show revenue by doctor for visits completed in the last hour"

"How many appointments need follow-up visits?"

Pro Tip: Snowflake Intelligence understands the CDC metadata columns and can differentiate between snapshot data and real-time changes. Try asking comparative questions like "Show me only the appointments that changed today" to see CDC-specific insights!

Alternative: SQL Queries

SELECT 
    'appointments' as table_name,
    COUNT(*) as total_records,
    SUM(CASE WHEN _SNOWFLAKE_UPDATED_AT IS NULL THEN 1 ELSE 0 END) as snapshot_only,
    SUM(CASE WHEN _SNOWFLAKE_UPDATED_AT IS NOT NULL THEN 1 ELSE 0 END) as updated_via_cdc,
    SUM(CASE WHEN _SNOWFLAKE_DELETED THEN 1 ELSE 0 END) as soft_deleted
FROM "appointments";

SELECT 
    "appointment_id",
    "patient_id",
    "status",
    "appointment_date",
    _SNOWFLAKE_UPDATED_AT,
    _SNOWFLAKE_DELETED
FROM "appointments"
WHERE _SNOWFLAKE_UPDATED_AT >= DATEADD(hour, -24, CURRENT_TIMESTAMP)
ORDER BY _SNOWFLAKE_UPDATED_AT DESC;

SELECT 
    d."first_name" || ' ' || d."last_name" as doctor_name,
    d."specialization",
    COUNT(a."appointment_id") as total_appointments,
    SUM(CASE WHEN a."status" = 'completed' THEN 1 ELSE 0 END) as completed
FROM "doctors" d
LEFT JOIN "appointments" a ON d."doctor_id" = a."doctor_id"
WHERE NOT a._SNOWFLAKE_DELETED OR a._SNOWFLAKE_DELETED IS NULL
GROUP BY d."doctor_id", d."first_name", d."last_name", d."specialization"
ORDER BY completed DESC;

Best of Both Worlds: Use Snowflake Intelligence for ad-hoc exploration and business user queries, and SQL for repeatable reporting and dashboards!

Cleanup

⚠️ DESTRUCTIVE OPERATIONS: Some cleanup steps below will permanently delete databases, schemas, tables, and all data created during this quickstart. Make sure you've captured any queries or insights you want to keep before proceeding!

Step 1: Stop the PostgreSQL Connector (Optional)

TIP: Stopping the connector preserves your configuration while preventing further CDC processing and associated compute costs. You can restart it later by right-clicking and selecting Start. Use this option if you want to keep your runtime and connector configuration intact.

Step 2: Delete Openflow Runtime (Permanent)

For Dedicated Runtime (Created for This Quickstart)

What Gets Deleted: Deleting the runtime removes all connectors, services (NiFi, Registry), and the SPCS infrastructure. This automatically closes all database connections, including PostgreSQL replication slots.

For Shared Runtime (From Companion Quickstart)

⚠️ IMPORTANT: If you're using the Openflow runtime from the companion quickstart "Getting Started with Openflow on SPCS", DO NOT delete the runtime! It may be used by other connectors or projects.

Keep the Runtime: The shared runtime can be reused for future Openflow projects, including other CDC connectors, API integrations, or file ingestion pipelines. Stopping services preserves all configuration while eliminating compute costs.

Step 3: Clean Up PostgreSQL

Terminate Active Replication Slots

-- View replication slots
SELECT slot_name, slot_type, active, restart_lsn 
FROM pg_replication_slots 
WHERE database = 'postgres';

-- Drop inactive replication slot(s)
SELECT pg_drop_replication_slot(slot_name) 
FROM pg_replication_slots 
WHERE database = 'postgres' 
  AND NOT active;

Replication Slot Naming: The replication slot name is automatically generated by Openflow following the pattern snowflake_connector_<unique_id>. The query above will show you the exact name.

If Slot is Active: If the slot still shows active = true (connector not fully stopped), first terminate the backend connection:

-- Terminate the active backend connection
SELECT pg_terminate_backend(active_pid)
FROM pg_replication_slots
WHERE slot_name = 'snowflake_connector_XXXXX' AND active = true;

-- Then drop the slot
SELECT pg_drop_replication_slot('snowflake_connector_XXXXX');

Drop the Publication

-- Drop the publication for healthcare tables
DROP PUBLICATION IF EXISTS healthcare_cdc_publication;

Drop the Healthcare Schema (Optional)

-- Drop the healthcare schema and all its objects
DROP SCHEMA IF EXISTS healthcare CASCADE;

-- Verify cleanup
SELECT schema_name 
FROM information_schema.schemata 
WHERE schema_name = 'healthcare';

Keep PostgreSQL? If you're using a shared PostgreSQL instance, you may want to keep the schema and just drop the publication. If this was a dedicated test instance, you can safely drop everything.

Step 4: Drop Snowflake Intelligence Agent

-- Use ACCOUNTADMIN role (required for agent operations)
USE ROLE ACCOUNTADMIN;

-- Drop the Snowflake Intelligence agent
DROP AGENT IF EXISTS snowflake_intelligence.agents.HEALTHCARE_DATA_INTELLIGENCE;

Step 5: Drop Snowflake Objects

-- Use ACCOUNTADMIN role for cleanup
USE ROLE ACCOUNTADMIN;

-- Drop the main database (includes healthcare schema and all tables)
DROP DATABASE IF EXISTS QUICKSTART_PGCDC_DB CASCADE;

-- Drop the warehouse
DROP WAREHOUSE IF EXISTS QUICKSTART_PGCDC_WH;

-- Drop the external access integration
DROP INTEGRATION IF EXISTS quickstart_pgcdc_access;

-- Drop the role
DROP ROLE IF EXISTS QUICKSTART_ROLE;

Verify Cleanup

-- Check for remaining objects
SHOW DATABASES LIKE 'QUICKSTART_PGCDC_DB';
SHOW DATABASES LIKE 'snowflake_intelligence';
SHOW WAREHOUSES LIKE 'QUICKSTART_PGCDC_WH';
SHOW INTEGRATIONS LIKE 'quickstart_pgcdc_access';
SHOW ROLES LIKE 'QUICKSTART_ROLE';

Step 6: Remove Local Files (Optional)

# Navigate to parent directory
cd ..

# Remove the quickstart directory
rm -rf openflow-postgresql-cdc-demo

# Verify removal
ls -la | grep openflow-postgresql-cdc-demo

Cleanup Complete

Cost Savings: By cleaning up unused resources, you avoid unnecessary compute costs. Warehouses, services, and replication slots all consume credits when active!

Conclusion And Resources

What You Learned

Related Resources

---