fix: parameterize schema and publication name in postgres-cdc-demo init SQL

[sfc-gh-pvillard]

Summary

• Add Publication Name parameter (default: demo_publication) to the flow's parameter context
• Replace hardcoded demo_postgres schema in the init SQL with #{Schema Name}
• Replace hardcoded demo_publication in the init SQL with #{Publication Name}
• Update documentation to include the new parameter

This ensures the flow is fully portable — any deployment can configure its own schema and publication name rather than being forced to use hardcoded values.

Test plan

• Import flow and verify Publication Name parameter appears in NiFi
• Verify init SQL creates schema and publication using the configured parameter values

.... Generated with Cortex Code

[github-actions]

🚀 Flow Deploy Available

This PR modifies flow definitions that have a CI test configuration. A maintainer (admin or maintain permission) can deploy and test the flow on an ephemeral Snowflake runtime by commenting:

deploy this flow or deploy the flow

This will:

1. Create a dedicated ephemeral runtime
2. Deploy the changed flow(s)
3. Run validation tests
4. Tear down the runtime and report results

To keep the runtime alive for debugging, add do not clean to the comment.

[github-actions]

Note

This GitHub Action is created and maintained by Snowflake.

Identified 1 changed flows in this Pull Request.

Executing Snowflake Flow Diff for flow: postgres-cdc-demo

No Checkstyle Violations found

Flow Changes

• In Processor of type ExecuteSQLStatement named ExecuteSQLStatement, the value of the property SQL changed from

CREATE SCHEMA IF NOT EXISTS demo_postgres;

CREATE TABLE IF NOT EXISTS demo_postgres.customers (
    customer_id INTEGER PRIMARY KEY,
    first_name VARCHAR(100) NOT NULL,
    last_name VARCHAR(100) NOT NULL,
    email VARCHAR(255),
    phone VARCHAR(50),
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS demo_postgres.orders (
    order_id INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    order_status VARCHAR(50) DEFAULT 'pending',
    total_amount NUMERIC(10,2),
    shipping_address TEXT,
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS demo_postgres.order_items (
    item_id SERIAL PRIMARY KEY,
    order_id INTEGER NOT NULL,
    product_name VARCHAR(255) NOT NULL,
    quantity INTEGER DEFAULT 1,
    unit_price NUMERIC(10,2)
);

DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = 'demo_publication') THEN
        CREATE PUBLICATION demo_publication FOR TABLE
            demo_postgres.customers,
            demo_postgres.orders,
            demo_postgres.order_items;
    END IF;
END
$$;

to

CREATE SCHEMA IF NOT EXISTS #{Schema Name};

CREATE TABLE IF NOT EXISTS #{Schema Name}.customers (
    customer_id INTEGER PRIMARY KEY,
    first_name VARCHAR(100) NOT NULL,
    last_name VARCHAR(100) NOT NULL,
    email VARCHAR(255),
    phone VARCHAR(50),
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS #{Schema Name}.orders (
    order_id INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    order_status VARCHAR(50) DEFAULT 'pending',
    total_amount NUMERIC(10,2),
    shipping_address TEXT,
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS #{Schema Name}.order_items (
    item_id SERIAL PRIMARY KEY,
    order_id INTEGER NOT NULL,
    product_name VARCHAR(255) NOT NULL,
    quantity INTEGER DEFAULT 1,
    unit_price NUMERIC(10,2)
);

DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = '#{Publication Name}') THEN
        CREATE PUBLICATION #{Publication Name} FOR TABLE
            #{Schema Name}.customers,
            #{Schema Name}.orders,
            #{Schema Name}.order_items;
    END IF;
END
$$;

• In the Parameter Context CDC Postgres Demo - Data Generator a parameter has been added: Publication Name = demo_publication with the description Name of the PostgreSQL publication to create for CDC. Must match the publication name used by the CDC connector.

[sfc-gh-dchaffelson]

LGTM. Good improvement.