kafka_fdw

pig build pkg kafka_fdw;		# build rpm/deb

pig repo add pgsql -u   # add both repo and update cache

pig install kafka_fdw;		# install via package name, for the active PG version

pig install kafka_fdw -v 18;   # install for PG 18
pig install kafka_fdw -v 17;   # install for PG 17
pig install kafka_fdw -v 16;   # install for PG 16
pig install kafka_fdw -v 15;   # install for PG 15
pig install kafka_fdw -v 14;   # install for PG 14

CREATE EXTENSION kafka_fdw;

CREATE EXTENSION kafka_fdw;

CREATE SERVER kafka_server FOREIGN DATA WRAPPER kafka_fdw
  OPTIONS (brokers 'localhost:9092');

CREATE USER MAPPING FOR PUBLIC SERVER kafka_server;

CREATE FOREIGN TABLE kafka_csv (
  part int OPTIONS (partition 'true'),
  offs bigint OPTIONS (offset 'true'),
  some_int int,
  some_text text,
  some_date date,
  some_time timestamp
)
SERVER kafka_server
OPTIONS (format 'csv', topic 'my_topic', batch_size '30', buffer_delay '100');

CREATE FOREIGN TABLE kafka_json (
  part int OPTIONS (partition 'true'),
  offs bigint OPTIONS (offset 'true'),
  some_int int OPTIONS (json 'int_val'),
  some_text text OPTIONS (json 'text_val')
)
SERVER kafka_server
OPTIONS (format 'json', topic 'my_json_topic', batch_size '30', buffer_delay '100');

-- Read from a specific partition and offset
SELECT * FROM kafka_csv WHERE part = 0 AND offs > 1000 LIMIT 60;

-- Read from multiple partitions
SELECT * FROM kafka_csv
WHERE (part = 0 AND offs > 100) OR (part = 1 AND offs > 300);

-- Insert with explicit partition
INSERT INTO kafka_csv (part, some_int, some_text)
  VALUES (0, 42, 'hello from partition 0');

-- Insert with auto-partition selection
INSERT INTO kafka_csv (some_int, some_text)
  VALUES (42, 'auto-partitioned message');