Kafka-connect-jdbc: Please support Postgres jsonb

[understructure]

Getting this warning for a table with a JSONB column:

WARN JDBC type 1111 not currently supported

I haven't tried regular JSON.

+1 for supporting Postgres boolean type:

WARN Ignoring record due to SQL error: (io.confluent.connect.jdbc.JdbcSourceTask:73)
org.postgresql.util.PSQLException: Bad value for type byte : f

Thanks in advance!

[vikasdp]

+1

[ryananguiano]

Same issue

[robconrad]

+1

[Oktaz]

+1

[dandanmylady]

+1

[khgitting]

I'd also like to see this. Is this another issue that will be easier with https://github.com/confluentinc/kafka-connect-jdbc/issues/303 ?

[venergiac]

+1

[Deninc]

+1

[minh5]

Running into this as well, +1

[rhauch]

Thanks for the suggestion, and all the +1s. The naive approach to solving this is to treat json and jsonb as a string, which would essentially encode the JSON into a single string field. This would work across all converters (not just JSON). Is this sufficient? I've already incorporated this into the [RDBMS-specific dialects|https://github.com/confluentinc/kafka-connect-jdbc/pull/333/files#diff-9a03fcf4950ba15217add89d2a4664a9R113] in #333 (for issue #303), at least on the source connector side.

OTOH, extracting the JSON content into a Connect Struct or Map is also possible, but likely has other implications and complications with respect to schemas. Here, any variation in JSON structure would cause schema changes that could make things messy, unless we have a "smart schema inference" mechanism that would compute a schema that covers all JSON structures seen so far. But this also has lots of risk and complications.

[austinnichols101]

It would be sufficient for my current use-cases.

[akashdeep01]

+1

[matthewrj]

Currently when using JdbcSinkConnector, it attempts to write jsonb data as VARCHAR.
Connect config:

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "table.name.format": "messages",
    "tasks.max": "1",
    "topics": "topic1",
    "transforms": "HoistPayload",
    "transforms.HoistPayload.field": "message_json",
    "auto.evolve": "false",
    "name": "postgres-sink",
    "auto.create": "false",
    "connection.url": "jdbc:postgresql://postgres:5432/postgres?user=postgres&password=example",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "insert.mode": "upsert",
    "transforms.HoistPayload.type": "org.apache.kafka.connect.transforms.HoistField$Value",
    "pk.mode": "kafka",
    "pk.fields": "topic,partition,offset"
}

Table:

CREATE TABLE messages (
    "topic" VARCHAR(64) NOT NULL,
    "partition" INTEGER NOT NULL,
    "offset" BIGINT NOT NULL,
    "message_json" jsonb,
    PRIMARY KEY("topic", "partition", "offset")
);

Error:

INSERT INTO "messages" ("topic","partition","offset","message_json") VALUES ('topic1',0,0,'{"foo": "bar"}') ON CONFLICT ("topic","partition","offset") DO UPDATE SET "message_json"=EXCLUDED."message_json" was aborted: ERROR: column "message_json" is of type jsonb but expression is of type character varying
Hint: You will need to rewrite or cast the expression.

Is it possible for the PostgreSqlDatabaseDialect to use ?::JSON in buildUpsertQueryStatement for column types jsonb?
https://github.com/confluentinc/kafka-connect-jdbc/blob/master/src/main/java/io/confluent/connect/jdbc/dialect/PostgreSqlDatabaseDialect.java#L240

[gopalakrishnan-subramani]

This has been long pending issue, really a show stopped at many projects. Surprised Confluence team doesn't care about it.

[rhyek]

Working on a new project where I am considering having Kafka as my single-source-of-truth, where a SQL database and Firestore would both behave as parallel Kafka Connect sinks. With Firestore, Kafka, and Avro schemas all handling nested Json documents well, what alternatives are there for PostgreSQL and this JDBC sink plugin not working well with JSONB columns?

[OneCricketeer]

what alternatives are there for PostgreSQL

Try a noSQL database?

Neo4J is great for such types of nested collections

[Dishwasha]

This works instead:

CREATE OR REPLACE FUNCTION varchar_to_jsonb (varchar) RETURNS jsonb AS $$ SELECT to_jsonb($1) $$ LANGUAGE SQL;
CREATE CAST (varchar as jsonb) WITH FUNCTION varchar_to_jsonb(varchar) AS IMPLICIT;

[kabureu]

varchar_to_jsonb

How can use this PG function with Kafka connector during sink time?

[Dishwasha]

@kabureu It's automatic within postgres on insert/upsert. Just add this on the PG side on your database/schema.

[kabureu]

@kabureu It's automatic within postgres on insert/upsert. Just add this on the PG side on your database/schema.

Cool, also do we specify field(s) that we want this to apply to? as I only need to cast subset of columns

[Dishwasha]

@kabureu This would cast all fields with a source type of varchar (internally this is what the kafka connector uses even for text) that is going to a destination column of type jsonb. If you somehow limited those fields, you would get an error due to no ability to cast. Not sure why you would want that.

[sergyv]

@Dishwasha I have noticed that it escapes double quotes, so I have changed your function a bit to remove escaping as follows:
CREATE OR REPLACE FUNCTION varchar_to_jsonb (varchar) RETURNS jsonb AS $$ SELECT regexp_replace(trim(both '"' from $1), '\\"', '"', 'g')::jsonb $$ LANGUAGE SQL;

[KyleMccr]

For awareness the CREATE CAST solution does not work in Amazon RDS. It requires full admin owner which belongs to Amazon while your admin account is slightly restricted.

[sourgoyal]

Target type json/jsonb is not supported in io.confluent.connect.jdbc.JdbcSinkConnector.
Solution is using varchar_to_jsonb but it will be applied to all varchar columns. How can we specify the column name in this case?
Also it doesn't work in AWS RDS? then what are the alternatives?

[alstel]

Would still love to see this data type supported.

[robdesbois]

This appears to be fixed in the current connector. The Postgresql dialect for JdbcSinkConnector will now automatically convert to UUID, JSON and JSONB column types [source].

I then configured the connector as below. I've omitted irrelevant parts, so it's incomplete as-is. It uses the ToJSON transform from jcustenborder's common transforms plugin, to convert the message value to JSON, then hoists it into a data field in a struct so it maps onto a column in the table also called data.

Note that auto table-creation won't create JSONB columns for you (see PostgreSqlDatabaseDialect.getSqlType()), so you'll have to manually create your table.

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "transforms": "valueToJSON,wrapValue",
    "transforms.valueToJSON.type": "com.github.jcustenborder.kafka.connect.transform.common.ToJSON$Value",
    "transforms.valueToJSON.schemas.enable": "false",

    "transforms.wrapValue.type":"org.apache.kafka.connect.transforms.HoistField$Value",
    "transforms.wrapValue.field":"data"
}