Connect to PostgreSQL CDC

CREATE SOURCE [ IF NOT EXISTS ] <shared_source_name> WITH (
    connector='postgres-cdc',
    <field>=<value>, ...
);

CREATE TABLE [ IF NOT EXISTS ] <rw_table_name> (
    <column_name> <data_type> PRIMARY KEY , ...
    PRIMARY KEY ( <column_name>, ... )
)
[ INCLUDE timestamp AS <column_name> ]
WITH (
    snapshot='true'
)
FROM <shared_source_name> TABLE <upstream_table_name>;

-- Create a shared CDC source
CREATE SOURCE shared_source WITH (
    connector='postgres-cdc',
    hostname='localhost',
    port='5432',
    username='your_user',
    password='your_password',
    database.name='your_database',
    schema.name='public' -- Optional, defaults to 'public'
);

-- Create a table from the source, representing a specific PostgreSQL table
CREATE TABLE my_table (
    id INT PRIMARY KEY,
    name VARCHAR
)
FROM shared_source TABLE 'public.my_upstream_table';

CREATE TABLE large_table (
  id integer primary key,
  data varchar
)
WITH (
  backfill.parallelism = '4',
  backfill.num_rows_per_split = '50000',
  backfill.as_even_splits = 'true'
)
FROM pg_mydb TABLE 'public.large_table';

debezium.schema.history.internal.skip.unparseable.ddl = 'true'

CREATE TABLE test_toast (
    id int PRIMARY KEY,
    v_text text,
    v_json jsonb,
    v_bytea bytea
);

INSERT INTO test_toast (id, v_text, v_json, v_bytea)
VALUES (
    1,
    repeat('long_text_', 10000),  -- large text triggers TOAST
    jsonb_build_object('data', repeat('x', 50000)),  -- large JSON triggers TOAST
    repeat('a', 2000000)::bytea  -- large bytea triggers TOAST
);

CREATE SOURCE src_test_toast WITH (
    connector = 'postgres-cdc',
    hostname = 'localhost',
    port = 5432,
    username = 'pguser',
    password = 'pgpassword',
    database.name = 'mydb',
    schema.name = 'public',
    slot.name = 'rw_slot_toast_example'
);

CREATE TABLE rw_test_toast(*) FROM src_test_toast TABLE 'public.test_toast';

SELECT
    id,
    CASE WHEN octet_length(v_text) > 15000 THEN 'toast-triggered' ELSE 'small' END AS v_text_check,
    CASE WHEN octet_length(v_json::text) > 50000 THEN 'toast-triggered' ELSE 'small' END AS v_json_check,
    CASE WHEN octet_length(v_bytea) > 1500000 THEN 'toast-triggered' ELSE 'small' END AS v_bytea_check
FROM rw_test_toast
WHERE id = 1;

ALTER TABLE test_toast ADD COLUMN v_counter int DEFAULT 0;
UPDATE test_toast SET v_counter = 1 WHERE id = 1;

SELECT
    id,
    octet_length(v_text) AS text_len,
    octet_length(v_json::text) AS json_len,
    octet_length(v_bytea) AS bytea_len,
    v_counter
FROM rw_test_toast
WHERE id = 1;

{{ config(materialized='source') }}
CREATE SOURCE {{ this }} WITH (
    connector = 'postgres-cdc',
    hostname = '127.0.0.1',
    port = '8306',
    username = 'root',
    password = '123456',
    database.name = 'mydb',
    slot.name = 'mydb_slot'
);

{{ config(materialized='table_with_connector') }}
CREATE TABLE {{ this }} (
    v1 integer primary key,
    v2 timestamp with time zone
) FROM {{ ref('pg_mydb') }} TABLE 'public.tt3';

CREATE TABLE <table_name> (*) FROM <source_name> TABLE '<schema_name>.<table_name>';

CREATE SOURCE pg_source WITH (
 connector = 'postgres-cdc',
 hostname = 'localhost',
 port = '5432',
 username = 'your_user',
 password = 'your_password',
 database.name = 'your_database',
 schema.name = 'public',
 auto.schema.change = 'true'
);

CREATE TABLE my_table (
    id INT PRIMARY KEY,
    name VARCHAR
)
FROM pg_source TABLE 'public.my_upstream_table';

-- In your PostgreSQL database:
ALTER TABLE my_upstream_table ADD COLUMN v1 VARCHAR(255);
ALTER TABLE my_upstream_table ADD COLUMN v2 NUMERIC(5,2);

-- In RisingWave:
DESCRIBE my_table;

CREATE PUBLICATION publication_name FOR table_name WITH (publish_via_partition_root = true);

CREATE TABLE person (
  id integer PRIMARY KEY,
  name varchar,
  next_id int AS id + 1,
  PRIMARY KEY (id)
) FROM pg_mydb TABLE 'public.person';

CREATE TABLE person (
  id integer PRIMARY KEY,
  next_id int AS id + 1,
  name varchar,
  PRIMARY KEY (id)
) FROM pg_mydb TABLE 'public.person';