Ingest CDC data from MySQL

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

CREATE TABLE [ IF NOT EXISTS ] table_name (
   column_name data_type PRIMARY KEY , ...
   PRIMARY KEY ( column_name, ... )
)
[ INCLUDE timestamp AS column_name ]
WITH (
   snapshot='true'
)
FROM source TABLE table_name;

CREATE TABLE mytable (v1 int PRIMARY KEY, v2 varchar)
INCLUDE timestamp AS commit_ts
FROM pg_source TABLE 'public.mytable';

SELECT * FROM t2 ORDER BY v1;

----RESULT
 v1 | v2 |         commit_ts
----+----+---------------------------
  1 | aa | 1970-01-01 00:00:00+00:00
  2 | bb | 1970-01-01 00:00:00+00:00
  3 | cc | 2024-05-20 09:01:08+00:00
  4 | dd | 2024-05-20 09:01:08+00:00

CREATE SOURCE mysql_mydb WITH (
  connector = 'mysql-cdc',
  hostname = '127.0.0.1',
  port = '8306',
  username = 'root',
  password = '123456',
  database.name = 'mydb',
  server.id = 5888,
  debezium.schema.history.internal.skip.unparseable.ddl = 'true'
);

CREATE TABLE person (
    id int,
    name varchar,
    email_address varchar,
    credit_card varchar,
    city varchar,
    PRIMARY KEY (id)
) INCLUDE TIMESTAMP AS commit_ts
INCLUDE DATABASE_NAME as database_name
INCLUDE TABLE_NAME as table_name
FROM mysql_source TABLE 'public.person';

CREATE SOURCE mysql_mydb WITH (
  connector = 'mysql-cdc',
  hostname = '127.0.0.1',
  port = '8306',
  username = 'root',
  password = '123456',
  database.name = 'mydb',
  server.id = 5888
);

CREATE TABLE t1_rw (
    v1 int,
    v2 int,
    PRIMARY KEY(v1)
) FROM mysql_mydb TABLE 'mydb.t1';

CREATE TABLE t3_rw (
  v1 INTEGER,
  v2 timestamptz,
  PRIMARY KEY (v1)
) FROM mysql_mydb TABLE 'mydb.t3';

SELECT * FROM __internal_orders_rw_4002_streamcdcscan_5002;

-[ RECORD 1 ]-----+---------------------------------------------------------------
split_id          | 5001
o_orderkey        | 4024320
backfill_finished | f
row_count         | 1006080
cdc_offset        | {"MySql": {"filename": "binlog.000005", "position": 60946679}}

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

{{ config(materialized='table_with_connector') }}
CREATE TABLE {{ this }}  (
    v1 int,
    v2 int,
    PRIMARY KEY(v1)
) FROM {{ ref('mysql_mydb') }} TABLE 'mydb.t1';

CREATE TABLE supplier (*) FROM mysql_source TABLE 'public.supplier';

       Name        |       Type        | Is Hidden | Description
-------------------+-------------------+-----------+-------------
 s_suppkey         | bigint            | false     |
 s_name            | character varying | false     |
 s_address         | character varying | false     |
 s_nationkey       | bigint            | false     |
 s_phone           | character varying | false     |
 s_acctbal         | numeric           | false     |
 s_comment         | character varying | false     |
 primary key       | s_suppkey         |           |
 distribution key  | s_suppkey         |           |
 table description | supplier          |           |
(10 rows)

CREATE SOURCE mysql_source WITH (
 connector = 'mysql-cdc',
 hostname = 'localhost',
 port = '3306',
 username = 'root',
 password = 'your_password',
 database.name = 'mytest',
 server.id = '5701',
 auto.schema.change = 'true'
);

CREATE TABLE rw_customers (
    id BIGINT,
    modified TIMESTAMP,
    custinfo JSONB,
    PRIMARY KEY (id)
) FROM mysql_source TABLE 'mytest.customers';

-- In MySQL:
USE mytest;
ALTER TABLE customers ADD COLUMN v1 VARCHAR(255);
ALTER TABLE customers ADD COLUMN v2 DOUBLE(5,2);

-- In RisingWave (after a brief pause):
DESCRIBE rw_customers;

Name                        | Type                           | Is Hidden | Description
----------------------------+--------------------------------+-----------+-------------
id                          | bigint                         | false     | NULL
modified                    | timestamp without time zone    | false     | NULL
custinfo                    | jsonb                          | false     | NULL
v1                          | character varying              | false     | NULL
v2                          | double precision               | false     | NULL
primary key id              | NULL                           | NULL      | NULL
distribution key id         | NULL                           | NULL      | NULL
table description rw_customers | NULL                       | NULL      | NULL

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

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

CREATE TABLE t3 (id INTEGER, v1 TIMESTAMP WITH TIME ZONE, PRIMARY KEY(id)) FROM mysql_source TABLE 'mydb.t3';

SHOW INTERNAL TABLES;

Name
---------------------------------
__internal_t3_3_streamcdcscan_4
__internal_mysql_source_1_source_2
(2 rows)

SELECT * FROM __internal_t3_3_streamcdcscan_4;

split_id | id | backfill_finished | row_count | cdc_offset
----------+----+-------------------+-----------+--------------------------------------------------
3        |  5 | t                 |         4 | {"MySql": {"filename": "binlog.000067", "position": 39101}}
(1 row)

SHOW MASTER STATUS;