https://flink.apache.org/ logo
Join Slack
Powered by
```CREATE OR REPLACE TABLE postgres_catalog.inboun...
# random
g
Copy code
CREATE OR REPLACE TABLE postgres_catalog.inbound.children (
     id             BIGINT
    ,nationalId     VARCHAR(14) -- NOT NULL
    ,data           STRING
    ,created_at     TIMESTAMP_LTZ(3)
    ,WATERMARK FOR created_at AS created_at - INTERVAL '15' SECOND
) 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'                            = 'test1'
    ,'scan.incremental.snapshot.enabled'    = 'false'           -- experimental feature: incremental snapshot (default of
    ,'scan.startup.mode'                    = 'latest-offset'         -- <https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1/docs/connectors/flink-sources/postgres-cdc/#startup-reading-position>
    ,'decoding.plugin.name'                 = 'pgoutput'
    ,'scan.incremental.snapshot.enabled'    = 'true'
);
g
Copy code
t{id=37457,nationalid=0282428R,data={"_id": "43065c0b-8388-4fbb-8f3f-e20f4aa88093", "dob": "20/10/13", "name": "Colleen", "gender": "Female", "address": {"town": "Galway City", "county": "Galway"
maybe try changing nationalId to nationalid?
g
it use to be uniqueId, had the same then... I was thinking that maybe some how that was a reserved word somewhere causing a clash so changed to this, same result.
g
Your issue sits somewhere between naming and data type mapping.
g
it's hehehe as simple as i can imagine. I have a second "adults" table with the same structure with the same problem, also on nationalId column. in pgAdmin the column is populated. source pg table create.
Copy code
CREATE TABLE adults (
    id            SERIAL      NOT NULL,
    nationalId    varchar(14) NOT NULL, 
    data          JSONB,
    created_at    timestamptz DEFAULT NOW() NOT NULL,
    PRIMARY KEY   (nationalId)
) TABLESPACE pg_default;
flink create
Copy code
CREATE OR REPLACE TABLE postgres_catalog.inbound.children (
     id             BIGINT
    ,nationalId     VARCHAR(14) -- NOT NULL
    ,data           STRING
    ,created_at     TIMESTAMP_LTZ(3)
    ,WATERMARK FOR created_at AS created_at - INTERVAL '15' SECOND
) 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'                            = 'test2'
    ,'scan.incremental.snapshot.enabled'    = 'false'           -- experimental feature: incremental snapshot (default of
    ,'scan.startup.mode'                    = 'latest-offset'   -- <https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1/docs/connectors/flink-sources/postgres-cdc/#startup-reading-position>
    ,'decoding.plugin.name'                 = 'pgoutput'
    ,'scan.incremental.snapshot.enabled'    = 'true'
);
Copy code
(3435973888 bytes), networkMemory=256.000mb (268435456 bytes)}
.
2025-08-28 09:31:27,413 INFO  org.apache.flink.runtime.resourcemanager.slotmanager.DefaultSlotStatusSyncer [] - Starting allocation of slot c7e594f5ad948c416da1d8067a81c83c from 172.20.0.13:34049-8ce550 for job 2dd2ef6d45c81eef51a9506497295ee5 with resource profile ResourceProfile{cpuCores=1, taskHeapMemory=142.507mb (149429068 bytes), taskOffHeapMemory=0 bytes, managedMemory=109.227mb (114532462 bytes), networkMemory=8.533mb (8947848 bytes)}.
2025-08-28 09:31:27,414 DEBUG io.debezium.relational.TableSchemaBuilder                    [] - Mapping table 'public.children' to schemas under 'postgres_cdc_source.public.children'
2025-08-28 09:31:27,416 DEBUG io.debezium.relational.TableSchemaBuilder                    [] - - field 'nationalid' (STRING) from column nationalid varchar(14, 0) NOT NULL
2025-08-28 09:31:27,418 DEBUG io.debezium.relational.TableSchemaBuilder                    [] - - field 'id' (INT32) from column id serial(10, 0) NOT NULL AUTO_INCREMENTED DEFAULT VALUE nextval('children_id_seq'::regclass)
2025-08-28 09:31:27,419 DEBUG io.debezium.relational.TableSchemaBuilder                    [] - - field 'nationalid' (STRING) from column nationalid varchar(14, 0) NOT NULL
2025-08-28 09:31:27,419 DEBUG io.debezium.relational.TableSchemaBuilder                    [] - - field 'data' (STRING) from column data jsonb(2147483647, 0) DEFAULT VALUE NULL
2025-08-28 09:31:27,420 DEBUG io.debezium.relational.TableSchemaBuilder                    [] - - field 'created_at' (STRING) from column created_at timestamptz(35, 6) NOT NULL DEFAULT VALUE now()
2025-08-28 09:31:27,422 DEBUG io.debezium.relational.TableSchemaBuilder                    [] - Mapped primary key for table 'public.children' to schema: {"name" : "postgres_cdc_source.public.children.Key", "type" : "STRUCT", "optional" : "false", "default" : null, "fields" : [{"name" : "nationalid", "index" : "0", "schema" : {"type" : "STRING", "optional" : "false", "default" : null}}]}
2025-08-28 09:31:27,422 DEBUG io.debezium.relational.TableSchemaBuilder                    [] - Mapped columns for table 'public.children' to schema: {"name" : "postgres_cdc_source.public.children.Value", "type" : "STRUCT", "optional" : "true", "default" : null, "fields" : [{"name" : "id", "index" : "0", "schema" : {"type" : "INT32", "optional" : "false", "default" : "0"}}, {"name" : "nationalid", "index" : "1", "schema" : {"type" : "STRING", "optional" : "false", "default" : null}}, {"name" : "data", "index" : "2", "schema" : {"name" : "io.debezium.data.Json", "type" : "STRING", "optional" : "true", "default" : null, "version" : "1"}}, {"name" : "created_at", "index" : "3", "schema" : {"name" : "io.debezium.time.ZonedTimestamp", "type" : "STRING", "optional" : "false", "default" : "1970-01-01T00:00:00Z", "version" : "1"}}]}
2025-08-28 09:31:27,429 INFO  io.debezium.connector.postgresql.PostgresObjectUtils         [] - Creating a new replication connection for io.debezium.connector.postgresql.PostgresTaskContext@1b8fec5e
2025-08-28 09:31:27,433 DEBUG io.debezium.connector.postgresql.connection.PostgresReplicationConnection [] - Creating new replication slot 'test2' for plugin 'PGOUTPUT'
2025-08-28 09:31:27,446 DEBUG io.debezium.jdbc.JdbcConnection                              [] - Connected to jdbc:<postgresql://postgrescdc:5432/demog> with {replication=database, ApplicationName=Debezium Streaming, server.name=postgres_cdc_source, tcpKeepAlive=true, history=org.apache.flink.cdc.connectors.base.source.EmbeddedFlinkDatabaseHistory, password=***, assumeMinServerVersion=9.4, user=dbadmin, preferQueryMode=simple}
2025-08-28 09:31:27,447 INFO  io.debezium.connector.postgresql.connection.PostgresReplicationConnection [] - Initializing PgOutput logical decoder publication
2025-08-28 09:31:27,453 INFO  io.debezium.connector.postgresql.connection.PostgresReplicationConnection [] - Creating replication slot with command CREATE_REPLICATION_SLOT "test2"  LOGICAL pgoutput
2025-08-28 09:31:27,454 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Source: children[24] -> WatermarkAssigner[25] (1/2) (e940d159e292bd4507a0933a2fa6182a_cbc357ccb763df2852fee8c4fc7d55f2_0_0) switched from SCHEDULED to DEPLOYING.
2025-08-28 09:31:27,455 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Deploying Source: children[24] -> WatermarkAssigner[25] (1/2) (attempt #0) with attempt id e940d159e292bd4507a0933a2fa6182a_cbc357ccb763df2852fee8c4fc7d55f2_0_0 and vertex id cbc357ccb763df2852fee8c4fc7d55f2_0 to 172.20.0.13:34049-8ce550 @ graph-taskmanager-2.graph (dataPort=46825) with allocation id c7e594f5ad948c416da1d8067a81c83c
2025-08-28 09:31:27,460 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Source: children[24] -> WatermarkAssigner[25] (2/2) (e940d159e292bd4507a0933a2fa6182a_cbc357ccb763df2852fee8c4fc7d55f2_1_0) switched from SCHEDULED to DEPLOYING.
2025-08-28 09:31:27,460 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Deploying Source: children[24] -> WatermarkAssigner[25] (2/2) (attempt #0) with attempt id e940d159e292bd4507a0933a2fa6182a_cbc357ccb763df2852fee8c4fc7d55f2_1_0 and vertex id cbc357ccb763df2852fee8c4fc7d55f2_1 to 172.20.0.13:34049-8ce550 @ graph-taskmanager-2.graph (dataPort=46825) with allocation id b2dc1e457261aeabce117e9fd2bc5b56
2025-08-28 09:31:27,461 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Sink: Collect table sink (1/1) (e940d159e292bd4507a0933a2fa6182a_9dd63673dd41ea021b896d5203f3ba7c_0_0) switched from SCHEDULED to DEPLOYING.
2025-08-28 09:31:27,461 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Deploying Sink: Collect table sink (1/1) (attempt #0) with attempt id e940d159e292bd4507a0933a2fa6182a_9dd63673dd41ea021b896d5203f3ba7c_0_0 and vertex id 9dd63673dd41ea021b896d5203f3ba7c_0 to 172.20.0.13:34049-8ce550 @ graph-taskmanager-2.graph (dataPort=46825) with allocation id c7e594f5ad948c416da1d8067a81c83c
2025-08-28 09:31:27,514 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Source: children[24] -> WatermarkAssigner[25] (2/2) (e940d159e292bd4507a0933a2fa6182a_cbc357ccb763df2852fee8c4fc7d55f2_1_0) switched from DEPLOYING to INITIALIZING.
2025-08-28 09:31:27,517 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Sink: Collect table sink (1/1) (e940d159e292bd4507a0933a2fa6182a_9dd63673dd41ea021b896d5203f3ba7c_0_0) switched from DEPLOYING to INITIALIZING.
2025-08-28 09:31:27,520 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Source: children[24] -> WatermarkAssigner[25] (1/2) (e940d159e292bd4507a0933a2fa6182a_cbc357ccb763df2852fee8c4fc7d55f2_0_0) switched from DEPLOYING to INITIALIZING.
2025-08-28 09:31:27,547 INFO  org.apache.flink.streaming.api.operators.collect.CollectSinkOperatorCoordinator [] - Received sink socket server address: /172.20.0.13:35587
2025-08-28 09:31:27,550 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Sink: Collect table sink (1/1) (e940d159e292bd4507a0933a2fa6182a_9dd63673dd41ea021b896d5203f3ba7c_0_0) switched from INITIALIZING to RUNNING.
2025-08-28 09:31:27,570 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Source: children[24] -> WatermarkAssigner[25] (2/2) (e940d159e292bd4507a0933a2fa6182a_cbc357ccb763df2852fee8c4fc7d55f2_1_0) switched from INITIALIZING to RUNNING.
2025-08-28 09:31:27,570 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Source: children[24] -> WatermarkAssigner[25] (1/2) (e940d159e292bd4507a0933a2fa6182a_cbc357ccb763df2852fee8c4fc7d55f2_0_0) switched from INITIALIZING to RUNNING.
2025-08-28 09:31:27,589 INFO  org.apache.flink.streaming.api.operators.collect.CollectSinkOperatorCoordinator [] - Sink connection established
2025-08-28 09:31:27,604 DEBUG io.debezium.connector.postgresql.connection.PostgresReplicationConnection [] - Closing message decoder
2025-08-28 09:31:27,605 INFO  io.debezium.jdbc.JdbcConnection                              [] - Connection gracefully closed
2025-08-28 09:31:27,605 DEBUG io.debezium.connector.postgresql.connection.PostgresReplicationConnection [] - Closing replication connection
2025-08-28 09:31:27,606 INFO  io.debezium.jdbc.JdbcConnection                              [] - Connection gracefully closed
2025-08-28 09:31:27,606 INFO  org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Source Source: children[24] registering reader for parallel task 1 (#0) @ 172.20.0.13
2025-08-28 09:31:27,607 INFO  org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Source Source: children[24] received split request from parallel task 1 (#0)
2025-08-28 09:31:27,616 INFO  io.debezium.jdbc.JdbcConnection                              [] - Connection gracefully closed
2025-08-28 09:31:27,619 INFO  org.apache.flink.cdc.connectors.base.source.enumerator.IncrementalSourceEnumerator [] - Assign split StreamSplit{splitId='stream-split', offset=Offset{lsn=LSN{1/7A3D7CB8}, txId=21527, lastCommitTs=-9223372036854775808], endOffset=Offset{lsn=LSN{FFFFFFFF/FFFFFFFF}, txId=null, lastCommitTs=-9223372036853775810], isSuspended=false} to subtask 1
2025-08-28 09:31:27,620 INFO  org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Source Source: children[24] registering reader for parallel task 0 (#0) @ 172.20.0.13
2025-08-28 09:31:27,621 INFO  org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Source Source: children[24] received split request from parallel task 0 (#0)
2025-08-28 09:31:27,663 INFO  org.apache.flink.cdc.connectors.base.source.enumerator.IncrementalSourceEnumerator [] - The enumerator receives notice from subtask 1 for the stream split assignment.
changed nationalId to a string in flink sql, same result.
g
again i see nationalId in you DDL and the payload has nationalid
g
this is the postgres-init.sql that is executed when the db is spun up.
Copy code
CREATE TABLE adults (
    id            SERIAL      NOT NULL,
    nationalId    varchar(14) NOT NULL, 
    data          JSONB,
    created_at    timestamptz DEFAULT NOW() NOT NULL,
    PRIMARY KEY   (nationalId)
) TABLESPACE pg_default;

CREATE TABLE children (
    id            SERIAL      NOT NULL,
    nationalId    varchar(14) NOT NULL, 
    data          JSONB,
    created_at    timestamptz DEFAULT NOW() NOT NULL,
    PRIMARY KEY   (nationalId)
) TABLESPACE pg_default;

CREATE TABLE families (
    id            SERIAL      NOT NULL,
    data          JSONB,
    created_at    timestamptz DEFAULT NOW() NOT NULL,
    PRIMARY KEY   (id)
) TABLESPACE pg_default;

CREATE INDEX adults_nationalId_uidx ON public.adults (nationalId);
CREATE INDEX children_nationalId_uidx ON public.children (nationalId);
CREATE INDEX families_id_uidx ON public.families (id);

ALTER TABLE public.adults REPLICA IDENTITY FULL;
ALTER TABLE public.adults OWNER TO dbadmin;

ALTER TABLE public.children REPLICA IDENTITY FULL;
ALTER TABLE public.children OWNER TO dbadmin;

ALTER TABLE public.families REPLICA IDENTITY FULL;
ALTER TABLE public.families OWNER TO dbadmin;
strange, if i ask pgAdmin to five me ddl for the table I get this, and yes as you noticed the id is lowercase. it's as if everything is lower cased. let me go major search/replace across and lowercase the table name everywhere and see if that's the problem....
Copy code
CREATE TABLE IF NOT EXISTS public.children
(
    id integer NOT NULL DEFAULT nextval('children_id_seq'::regclass),
    nationalid character varying(14) COLLATE pg_catalog."default" NOT NULL,
    data jsonb,
    created_at timestamp with time zone NOT NULL DEFAULT now(),
    CONSTRAINT children_pkey PRIMARY KEY (nationalid)
)
thanks for noticing.
RESOLVED πŸŽ‰
πŸ™ 1
Open in Slack
PreviousNext
Powered by