

-- Flink Table Create
CREATE OR REPLACE TABLE postgres_catalog.inbound.children (
    id            BIGINT,                     
    nationalId    VARCHAR(14),              
    data          STRING,                   
    created_at    TIMESTAMP_LTZ(3),          
    WATERMARK FOR created_at AS created_at - INTERVAL '5' SECOND
    PRIMARY KEY (nationalId)     NOT ENFORCED
) WITH (
      'connector'                         = 'postgres-cdc'
     ,'hostname'                          = 'postgrescdc'
     ,'port'                              = '5432'               -- NOTE: this is the port of the db on the container, not the external docker exported port via a port mapping.
     ,'username'                          = 'dbadmin'
     ,'password'                          = 'dbpassword'
     ,'database-name'                     = 'demog'
     ,'schema-name'                       = 'public'
     ,'table-name'                        = 'children'
     ,'slot.name'                         = 'children'
     ,'scan.incremental.snapshot.enabled' = 'true'               -- experimental feature: incremental snapshot (default off)
     ,'scan.startup.mode'                 = 'initial'            -- https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1/docs/connectors/flink-sources/postgres-cdc/#startup-reading-position
     ,'decoding.plugin.name'              = 'pgoutput'
);
