Support pgsql binding

[jfallows]

Support psql binding with server and client kinds, so that pgsql client can be proxied to a pgsql server.

Support route by command, such as CREATE TABLE or SELECT so that different commands can be proxied to different pgsql servers.

See pgsql protocol specification.

Support pgsql command transformations to dynamically inject deployment details.

[akrambek]

CREATE TABLE IF NOT EXISTS balances(
    *,
    PRIMARY KEY (user_id)
)
INCLUDE KEY AS user_id
WITH (
    connector='kafka',
    topic='streampay-balances',
    properties.bootstrap.server='kafka:29092',
    scan.startup.mode='latest',
    scan.startup.timestamp.millis='140000000'
) FORMAT UPSERT ENCODE AVRO (
    schema.registry = 'http://schema-registry:8081'
);

CREATE SOURCE IF NOT EXISTS commands
INCLUDE KEY AS key
INCLUDE header'zilla:correlation-id' AS correlation_id
INCLUDE header 'zilla:identity' AS owner_id
INCLUDE timestamp as timestamp
WITH (
    connector='kafka',
    topic='streampay-commands',
    properties.bootstrap.server='kafka:29092',
    scan.startup.mode='latest',
    scan.startup.timestamp.millis='140000000'
) FORMAT PLAIN ENCODE AVRO (
    schema.registry = 'http://schema-registry:8081'
);

CREATE FUNCTION column(name varchar) RETURNS VARCHAR LANGUAGE javascript AS $$
    return name;
$$;

INSERT INTO balances (user_id, balance, timestamp) VALUES ('1', 1964, 1723593113);

SELECT * FROM balances;

SHOW TABLES;
DESCRIBE balances;
DROP TABLE balances;

psql-queries.pcapng.zip

[jfallows]

Looks like we want a single stream of pgsql messages, mapping to a bidirectional zilla stream, with each pgsql query message mapping to a zilla DATA frame or a zilla FLUSH frame (maintaining received order).

pgsql messages flowing towards the client include type (row description), row (data row), completion (command completion) and ready (ready for query).

query => DATA + PgsqlQueryDataEx
type => FLUSH + PgsqlTypeFlushEx
row => DATA + PgsqlRowDataEx
completion => FLUSH + PgsqlCompletedFlushEx
ready => FLUSH + PgsqlReadyFlushEx

[jfallows]

For pgsql message mappings in pgsql proxy binding, we need the following.

CREATE TABLE IF NOT EXISTS balances (name VARCHAR, type VARCHAR, description VARCHAR, PRMARY KEY (name));

becomes

CREATE TOPIC IF NOT EXISTS balances (name VARCHAR, type VARCHAR, description VARCHAR, PRMARY KEY (name)));

and

CREATE TABLE IF NOT EXISTS balances(
    *,
    PRIMARY KEY (name)
)
INCLUDE KEY AS name
WITH (
    connector='kafka',
    topic='balances',
    properties.bootstrap.server='kafka:29092',
    scan.startup.mode='latest',
    scan.startup.timestamp.millis='140000000'
) FORMAT UPSERT ENCODE AVRO (
    schema.registry = 'http://schema-registry:8081'
);

Note: the injected kafka bootstrap server and schema registry endpoint would be configured in pgsql proxy options.

The pgsql response to CREATE TABLE is a complete message followed by a ready message, so if we are splitting this into multiple commands, we can propagate complete and ready messages only after the last split command completes.

[jfallows]

We also need to sync back to kafka.

CREATE MATERIALIZED VIEW request_payments as
SELECT
    generate_guid() as id,
    encode(cmd.owner_id, 'escape') as from_user_id,
    u2.username as from_username,
    cmd.user_id as to_user_id,
    u1.username as to_username,
    amount,
    notes
FROM
    commands as cmd
JOIN
    users u1 ON u1.id = cmd.user_id
JOIN
    users u2 ON u2.id = encode(cmd.owner_id, 'escape')
WHERE
    key IS NOT NULL
    AND type = 'RequestPayment';

becomes

CREATE MATERIALIZED VIEW request_payments as
SELECT
    generate_guid() as id,
    encode(cmd.owner_id, 'escape') as from_user_id,
    u2.username as from_username,
    cmd.user_id as to_user_id,
    u1.username as to_username,
    amount,
    notes
FROM
    commands as cmd
JOIN
    users u1 ON u1.id = cmd.user_id
JOIN
    users u2 ON u2.id = encode(cmd.owner_id, 'escape')
WHERE
    key IS NOT NULL
    AND type = 'RequestPayment';

DESCRIBE MATERIALIZED VIEW request_payments;

CREATE TOPIC request_payments (
    id VARCHAR,
    from_user_id VARCHAR,
    from_username VARCHAR,
    to_user_id VARCHAR,
    to_username VARCHAR,
    amount DOUBLE,
    notes VARCHAR);

CREATE SINK request_payments_sink
FROM request_payments
WITH (
    connector='kafka',
    topic='request-payments',
    properties.bootstrap.server='kafka:29092',
    primary_key='id'
) FORMAT UPSERT ENCODE AVRO (
    schema.registry = 'http://schema-registry:8081'
);

Note: the injected kafka bootstrap server and schema registry endpoint would be configured in pgsql proxy options.