Vlad

RPPD - Rust Python Postgres Discovery

overview

Build

[!IMPORTANT] The target platform is Linux! The key component is pxrg, a rust postgres trigger engine that does not support windows.

[!NOTE] https://grpc.io/docs/protoc-installation/

Build

To build the project

./make.sh

FAQ

[!NOTE] reinstall

IF

Mismatched rust versions: cargo-pgrx Mismatched rust versions: cargo-pgrx

Then (this takes time)

cargo install cargo-pgrx --force
cargo pgrx init 

Overall architecture

topics

Usage

  1. Create postgres extension: will copy rppd.so and create few tables in desired schema
  2. Run as many service as required:
    • a text file in any format OR a dir with a file: rppd.rppd_config names are: ‘schema’, ‘url’, ‘bind’
    • three optional args, in any order: config schema, db connection url (starts with ‘postgres://’, binding IP and port default is ‘localhost:8881’
    • use PGPASSWORD env variable (or from file) to postgres connection
    • use PGUSER env variable (or read from file) to postgres connection
    • priority is: env (if used), if no use or not set, than app args param, than file, than default
  3. Perform insert or update to configure a callback AND fire a function

[!NOTE] Use select the row by ID on INSERT trigger would fail because requested row might not visibly yet depends on transaction isolation. The simple workaround is to sleep a second on Python function call.

Use to get more build in help:

psql> \df+ rppd*

Python samples

variables - kwargs

DB -- database connection
TOPIC -- fc.fns.topic
TABLE -- fc.fns.schema_table
TRIG -- UPDATE = 0;  INSERT = 1;  DELETE = 2;  TRUNCATE = 3;
"COLUMN_NAME" --i.e. "ID" = see trig_value (up to 3)

python example:


sql = "SELECT * FROM {} where a = %s".format(TABLE)
print(sql)

cur = DB.cursor()
cur.execute(sql, (PK))
print(cur.fetchall())

cur.execute("insert into test (b) values (%s)", ( "{}-{}".format(TOPIC, TRIG),))
DB.commit()

test example:

create table if not exists test_source (id serial primary key, input text);
create table if not exists test_sink (id serial primary key, data text);
\set code `cat test_fn.py`
insert into rppd_function (code, checksum, schema_table, topic) values (:'code', 'na', 'public.test_source', '.id');
CREATE TRIGGER test_src_event AFTER INSERT OR UPDATE OR DELETE ON test_source FOR EACH ROW EXECUTE PROCEDURE rppd_event();
insert into test_source (input) values ('test input');
select * from test_sink;

python test example

for the test above

cur = DB.cursor()
cur.execute("SELECT input FROM test_source where id = %s", ([ID]))
input = cur.fetchall()
if len(input) > 0:
    cur.execute("insert into test_sink (data) values (%s)", ( input[0] ))
    DB.commit()

TODO

Core features and improvements

Cloud integration