pipeline_kafka

pipeline_kafka copied to clipboard

Using PipelineDB 1.0 and pipeline_kafka built from latest master (as of today), when I run commands

CREATE FOREIGN TABLE foobar_transactions_stream (payload json) SERVER pipelinedb;

SELECT pipeline_kafka.consume_begin('foobar.public.transactions', 'foobar_transactions_stream',
  format := 'json', parallelism := 4, start_offset := -2);

it runs successfully, but when I try to select it, it always claims the format is CSV:

postgres=# SELECT * FROM pipeline_kafka.consumers;
 id |           topic            |          relation          | group_id | format | delimiter | quote | escape | batchsize | maxbytes | parallelism | timeout
----+----------------------------+----------------------------+----------+--------+-----------+-------+--------+-----------+----------+-------------+---------
  3 | foobar.public.transactions | foobar_transactions_stream |          | csv    | \x02      | \x01  |        |     10000 | 32000000 |           4 |     250

However, JSON operations like this seem to work:

CREATE VIEW view_txn_user_counts AS
  SELECT
    COUNT(*),
    payload -> 'after' ->> 'user_id' AS user_id
  FROM foobar_transactions_stream
  GROUP BY payload -> 'after' ->> 'user_id';

Am I doing something incorrectly, or is this a bug, or is this format something unrelated?