CREATE TYPE abc AS (a INT, b TEXT, c TIMESTAMP);
CREATE FOREIGN TABLE kafka_json (
part int OPTIONS (partition 'true'),
offs bigint OPTIONS (offset 'true'),
x abc)
SERVER kafka_server
OPTIONS (format 'json', topic 'json', batch_size '30', buffer_delay '100');
INSERT INTO kafka_json (x) VALUES ((1, 'test', current_timestamp));
SELECT * FROM kafka_json;
ERROR: malformed record literal: "{"a":1,"b":"test","c":"2018-07-04T11:16:55.671986"}"
DETAIL: Missing left parenthesis.Postgres expects an input string formatted like:
(1,"test","2018-07-04T11:16:55.671986")
for composite type. The easy solution would be to just remove keys from input string and replace {} with (). But this won't work if JSON document has different key order or extra or missing keys. So it makes sense to write a simple JSON parser to collect individual key-value pairs, reorder them if needed and fill the gaps with NULLs. It's also possible to use built-in JSONB facilities from postgres to parse and manage JSON documents but I anticipate that it would be less efficient than custom designed solution.
Postgres expects an input string formatted like:
for composite type. The easy solution would be to just remove keys from input string and replace
{}with(). But this won't work if JSON document has different key order or extra or missing keys. So it makes sense to write a simple JSON parser to collect individual key-value pairs, reorder them if needed and fill the gaps withNULLs. It's also possible to use built-inJSONBfacilities from postgres to parse and manage JSON documents but I anticipate that it would be less efficient than custom designed solution.