Ingest data from MongoDB CDC

CREATE TABLE [ IF NOT EXISTS ] source_name (
    _id data_type PRIMARY KEY ,
    payload jsonb
)
[ INCLUDE timestamp AS column_name ]
WITH (
    connector='mongodb-cdc',
    connector_parameter='value', ...
);

CREATE TABLE test (_id JSONB PRIMARY KEY, payload JSONB)
INCLUDE timestamp AS commit_ts
WITH (
  connector = 'mongodb-cdc',
  mongodb.url = 'mongodb://localhost:27017/?replicaSet=rs0',
  collection.name = 'test.*'
);

SELECT * FROM test;

----RESULT
                 _id                  |                                      payload                                      |         commit_ts
--------------------------------------+-----------------------------------------------------------------------------------+---------------------------
 {"$oid": "664c48e87d2c84adfabfc03f"} | {"_id": {"$oid": "664c48e87d2c84adfabfc03f"}, "data": "mydata", "name": "Ada"} | 2024-05-21 08:18:25+00:00
 {"$oid": "660125a80f048c7c7eff4a6a"} | {"_id": {"$oid": "660125a80f048c7c7eff4a6a"}, "name": "Tom"}                       | 1970-01-01 00:00:00+00:00

CREATE TABLE users (_id JSONB PRIMARY KEY, payload JSONB)
INCLUDE TIMESTAMP as commit_ts
INCLUDE DATABASE_NAME as database_name
INCLUDE COLLECTION_NAME as collection_name
WITH (
  connector = 'mongodb-cdc',
  mongodb.url = 'mongodb://mongodb:27017/?replicaSet=rs0',
  collection.name = 'random_data.*'
);

CREATE TABLE source_name (
   _id varchar PRIMARY KEY,
   payload jsonb
) WITH (
   connector='mongodb-cdc',
   mongodb.url='mongodb://localhost:27017/?replicaSet=rs0',
   collection.name='dev.*'
);

CREATE TABLE source_name (
   _id varchar PRIMARY KEY,
   payload jsonb
) WITH (
   connector='mongodb-cdc',
   mongodb.url='mongodb://localhost:27017/?replicaSet=rs0',
   collection.name='db1.*, db2.*'
);