Building a Complex, Real-Time Data Management Application

Let's Build a Complex, Real-Time Data Management Application JONATHAN S. KATZ PGCONF.EU 2018 OCTOBER 25, 2018 ...before the session ends!

About Crunchy Data 2 Market Leading Data Security • Crunchy Certified PostgreSQL is open source and Common Criteria EAL 2+ Certified, with essential security enhancements for enterprise deployment • Author of the DISA Secure Technology Implementation Guide for PostgreSQL and co-author of CIS PostgreSQL Benchmark. Move ATO from weeks to days! Cloud Ready Data Management • Open source, Kubernetes-based solutions proven to scale to 1000s of database instances • Cloud-agnostic technology provide flexibility on how to deploy databases to public clouds, private clouds, or on-premise technology Leader in Open Source Enterprise PostgreSQL • Developer of essential open source tools for high availability, disaster recovery, and and monitoring for PostgreSQL • Leading contributor and sponsor of features that enhance stability, security, and performance of PostgreSQL

• Director of Communications, Crunchy Data • Previously: Engineering leadership in startups • Longtime PostgreSQL community contributor • Advocacy & various committees for PGDG • @postgresql + .org content • Director, PgUS • Co-Organizer, NYCPUG • Conference organization + speaking • @jkatz05 About Me 3

• This talk introduces many different tools and techniques available in PostgreSQL for building applications • Introduces different features and where to find out more information • We have a lot of material to cover in a short time - the slides and demonstrations will be made available How to Approach This Talk 4

• Imagine we are managing the rooms at the Marriott Lisbon Hotel • We have a set of operating hours in which the rooms can be booked • Only one booking can occur in the room at a given time The Problem 5

• We need to know... • All the rooms that are available to book • When the rooms are available to be booked (operating hours) • When the rooms have been booked • And... • The system needs to be able to CRUD fast • (Create, Read, Update, Delete. Fast). Application Requirements 7

First, let's talk about how we can find availability

• Availability can be thought about in three ways: • Closed • Available • Unavailable (or "booked") • Our ultimate "calendar tuple" is (room, status, range) Managing Availability 10

• PostgreSQL 9.2 introduced "range types" that included the ability to store and efficiently search over ranges of data • Built-in: • Date, Timestamps • Integer, Numeric • Lookups (e.g. overlaps) can be sped up using GiST and SP-GiST indexes PostgreSQL Range Types 11 SELECT tstzrange('2018-10-26 09:30'::timestamptz, '2018-10-26 10:30'::timestamptz);

Availability 13 SELECT * FROM ( VALUES ('closed', tstzrange('2018-10-26 0:00', '2018-10-26 8:00')), ('available', tstzrange('2018-10-26 08:00', '2018-10-26 09:30')), ('unavailable', tstzrange('2018-10-26 09:30', '2018-10-26 10:30')), ('available', tstzrange('2018-10-26 10:30', '2018-10-26 16:30')), ('unavailable', tstzrange('2018-10-26 16:30', '2018-10-26 18:30')), ('available', tstzrange('2018-10-26 18:30', '2018-10-26 20:00')), ('closed', tstzrange('2018-10-26 20:00', '2018-10-27 0:00')) ) x(status, calendar_range) ORDER BY lower(x.calendar_range);

• Insert new ranges and dividing them up • PostgreSQL does not work well with discontiguous ranges (...yet) • Availability • Just for one day - what about other days? • What happens with data in the past? • What happens with data in the future? • Unavailability • Ensure no double-bookings • Overlapping Events? • Just one space But... 15

availability_rule id <serial> PRIMARY KEY room_id <int> REFERENCES (room) days_of_week <int[]> start_time <time> end_time <time> generate_weeks_into_future <int> DEFAULT 52 room id <serial> PRIMARY KEY name <text> availability id <serial> PRIMARY KEY room_id <int> REFERENCES (room) availability_rule_id <int> REFERENCES (availabilityrule) available_date <date> available_range <tstzrange> unavailability id <serial> PRIMARY KEY room_id <int> REFERENCES (room) unavailable_date <date> unavailable_range <tstzrange> calendar id <serial> PRIMARY KEY room_id <int> REFERENCES (room) status <text> DOMAIN: {available, unavailable, closed} calendar_date <date> calendar_range <tstzrange> Managing Availability 16

• We can now store data, but what about: • Generating initial calendar? • Generating availability based on rules? • Generating unavailability? • Sounds like we need to build an application Managing Availability 17

• To build our application, there are a few topics we will need to explore first: • generate_series • Recursive queries • SQL Functions • Set returning functions • PL/pgsql • Triggers Managing Availability 18

• Generate series is a "set returning" function, i.e. a function that can return multiple rows of data • Generate series can return: • A set of numbers (int, bigint, numeric) either incremented by 1 or some other integer interval • A set of timestamps incremented by a time interval(!!) generate_series: More than just generating test data 19 SELECT x::date FROM generate_series( '2018-01-01'::date, '2018-12-31'::date, '1 day'::interval ) x;

• PostgreSQL 8.4 introduced the "WITH" syntax and with it also introduced the ability to perform recursive queries • WITH RECURSIVE ... AS () • Base case vs. recursive case • UNION vs. UNION ALL • CAN HIT INFINITE LOOPS Recursion in my SQL? 20

Recursion in my SQL? 21 WITH RECURSIVE fac AS ( SELECT 1::numeric AS n, 1::numeric AS i UNION SELECT fac.n * (fac.i + 1), fac.i + 1 AS i FROM fac ) SELECT fac.n, fac.i FROM fac; Nope

Recursion in my SQL? 22 WITH RECURSIVE fac AS ( SELECT 1::numeric AS n, 1::numeric AS i UNION SELECT fac.n * (fac.i + 1), fac.i + 1 AS i FROM fac WHERE i + 1 <= 100 ) SELECT fac.n, fac.i FROM fac; Better

• PostgreSQL provides the ability to write functions to help encapsulate repeated behavior • PostgreSQL 11 introduces stored procedures which enables you to embed transactions! • SQL functions have many properties, including: • Input / output • Volatility (IMMUTABLE, STABLE, VOLATILE) (default VOLATILE) • Parallel safety (default PARALLEL UNSAFE) • LEAKPROOF; SECURITY DEFINER • Execution Cost • Language type (more on this later) Functions 23

Functions 24 CREATE OR REPLACE FUNCTION pgconfeu_fac(n int) RETURNS numeric AS $$ WITH RECURSIVE fac AS ( SELECT 1::numeric AS n, 1::numeric AS i UNION SELECT fac.n * (fac.i + 1), fac.i + 1 AS i FROM fac WHERE i + 1 <= $1 ) SELECT max(fac.n) FROM fac; $$ LANGUAGE SQL IMMUTABLE PARALLEL SAFE;

Functions 25 CREATE OR REPLACE FUNCTION pgconfeu_fac_set(n int) RETURNS SETOF numeric AS $$ WITH RECURSIVE fac AS ( SELECT 1::numeric AS n, 1::numeric AS i UNION SELECT fac.n * (fac.i + 1), fac.i + 1 AS i FROM fac WHERE i + 1 <= $1 ) SELECT fac.n FROM fac ORDER BY fac.n; $$ LANGUAGE SQL IMMUTABLE PARALLEL SAFE;

Functions 26 CREATE OR REPLACE FUNCTION pgopen_fac_table(n int) RETURNS TABLE(n numeric) AS $$ WITH RECURSIVE fac AS ( SELECT 1::numeric AS n, 1::numeric AS i UNION SELECT fac.n * (fac.i + 1), fac.i + 1 AS i FROM fac WHERE i + 1 <= $1 ) SELECT fac.n FROM fac ORDER BY fac.n; $$ LANGUAGE SQL IMMUTABLE PARALLEL SAFE;

• PostgreSQL has the ability to load in procedural languages and execute code in them beyond SQL • "PL" • Built-in: pgSQL, Python, Perl, Tcl • Others: Javascript, R, Java, C, JVM, Container, LOLCODE, Ruby, PHP, Lua, pgPSM, Scheme Procedural Languages 27

PL/pgSQL 28 CREATE EXTENSION IF NOT EXISTS plpgsql; CREATE OR REPLACE FUNCTION pgopen_fac_plpgsql(n int) RETURNS numeric AS $$ DECLARE fac numeric; i int; BEGIN fac := 1; FOR i IN 1..n LOOP fac := fac * i; END LOOP; RETURN fac; END; $$ LANGUAGE plpgsql IMMUTABLE PARALLEL SAFE;

• Triggers are functions that can be called before/after/instead of an operation or event • Data changes (INSERT/UPDATE/DELETE) • Events (DDL, DCL, etc. changes) • Atomic • Must return "trigger" or "event_trigger" • (Return "NULL" in a trigger if you want to skip operation) • (Gotcha: RETURN OLD [INSERT] / RETURN NEW [DELETE]) • Execute once per modified row or once per SQL statement • Multiple triggers on same event will execute in alphabetical order • Writeable in any PL language that defined trigger interface Triggers 29

Building a Synchronized System

We will scan through the application code. It will be available for download later ;-)

• [Test your live demos before running them, and you will have much success!] • availability_rule inserts took some time, > 500ms • availability: INSERT 52 • calendar: INSERT 52 from nontrivial function • Updates on individual availability / unavailability are not too painful • Lookups are faaaaaaaast Lessons of The Test 33

• Even with only 100 more rooms with a few set of rules, rule generation time increased significantly • Lookups are still lightning fast! Web Scale :( 35

• Added in PostgreSQL 9.4 • Replays all logical changes made to the database • Create a logical replication slot in your database • Only one receiver can consume changes from one slot at a time • Slot keeps track of last change that was read by a receiver • If receiver disconnects, slot will ensure database holds changes until receiver reconnects • Only changes from tables with primary keys are relayed • As of PostgreSQL 10, you can set a "REPLICA IDENTITY" on a UNIQUE, NOT NULL, non-deferrable, non-partial column(s) • Basis for Logical Replication Logical Decoding 36

• A logical replication slot has a name and an output plugin • PostgreSQL comes with the "test" output plugin • Have to write a custom parser to read changes from test output plugin • Several output plugins and libraries available • wal2json: https://github.com/eulerto/wal2json • jsoncdc: https://github.com/posix4e/jsoncdc • Debezium: http://debezium.io/ • (Test: https://www.postgresql.org/docs/11/static/test-decoding.html) • Every change in the database is streamed • Need to be aware of the logical decoding format Logical Decoding Out of the Box 37

• C: libpq • pg_recvlogical • PostgreSQL functions • Python: psycopg2 - version 2.7 • JDBC: version 42 • Go: go-pgx • JavaScript: node-postgres (pg-logical-replication) Driver Support 38

Using Logical Decoding 39 wal_level = logical max_wal_senders = 2 max_replication_slots = 2 postgresql.conf local replication jkatz trust pg_hba.conf # DEVELOPMENT ONLY SELECT * FROM pg_create_logical_replication_slot('schedule', 'wal2json'); In the database:

• We know it takes time to regenerate calendar • Want to ensure changes always propagate but want to ensure all users (managers, calendar searchers) have good experience Thoughts 40 🤔

• Will use the same data model as before as well as the same helper functions, but without the triggers • (That's a lie, we will have one set of DELETE triggers as "DELETE" in the wal2json output plugin currently does not provide enough information) Replacing Triggers 41

Replacing Triggers 42 /** * Helper function: substitute the data within the `calendar`; this can be used * for all updates that occur on `availability` and `unavailability` */ CREATE OR REPLACE FUNCTION calendar_manage(room_id int, calendar_date date) RETURNS void AS $$ WITH delete_calendar AS ( DELETE FROM calendar WHERE room_id = $1 AND calendar_date = $2 ) INSERT INTO calendar (room_id, status, calendar_date, calendar_range) SELECT $1, c.status, $2, c.calendar_range FROM calendar_generate_calendar($1, tstzrange($2, $2 + 1)) c $$ LANGUAGE SQL;

Replacing Triggers 43 /** Now, the trigger functions for availability and unavailability; needs this for DELETE */ CREATE OR REPLACE FUNCTION availability_manage() RETURNS trigger AS $trigger$ BEGIN IF TG_OP = 'DELETE' THEN PERFORM calendar_manage(OLD.room_id, OLD.available_date); RETURN OLD; END IF; END; $trigger$ LANGUAGE plpgsql; CREATE OR REPLACE FUNCTION unavailability_manage() RETURNS trigger AS $trigger$ BEGIN IF TG_OP = 'DELETE' THEN PERFORM calendar_manage(OLD.room_id, OLD.unavailable_date); RETURN OLD; END IF; END; $trigger$ LANGUAGE plpgsql; /** And the triggers, applied to everything */ CREATE TRIGGER availability_manage AFTER DELETE ON availability FOR EACH ROW EXECUTE PROCEDURE availability_manage(); CREATE TRIGGER unavailability_manage AFTER DELETE ON unavailability FOR EACH ROW EXECUTE PROCEDURE unavailability_manage();

• We will have a Python script that reads from a logical replication slot and if it detects a relevant change, take an action • Similar to what we did with triggers, but this moves the work to OUTSIDE the transaction • BUT...we can confirm whether or not the work is completed, thus if the program fails, we can restart from last acknowledged transaction ID Replacing Triggers 44

Reading the Changes 45 import json import sys import psycopg2 import psycopg2.extras SQL = { 'availability': { 'insert': """SELECT calendar_manage(%(room_id)s, %(available_date)s)""", 'update': """SELECT calendar_manage(%(room_id)s, %(available_date)s)""", }, 'availability_rule': { 'insert': True, 'update': True, }, 'room': { 'insert': """ INSERT INTO calendar (room_id, status, calendar_date, calendar_range) SELECT %(id)s, 'closed', calendar_date, tstzrange(calendar_date, calendar_date + '1 day'::interval) FROM generate_series( date_trunc('week', CURRENT_DATE), date_trunc('week', CURRENT_DATE + '52 weeks'::interval), '1 day'::interval ) calendar_date; """, }, 'unavailability': { 'insert': """SELECT calendar_manage(%(room_id)s, %(unavailable_date)s)""", 'update': """SELECT calendar_manage(%(room_id)s, %(unavailable_date)s)""", }, }

Reading the Changes 46 class StreamReader(object): def _consume_change(self, payload): connection = psycopg2.connect("dbname=realtime") cursor = connection.cursor() for data in payload['change']: sql = SQL.get(data.get('table'), {}).get(data.get('kind')) if not sql: return params = dict(zip(data['columnnames'], data['columnvalues'])) if data['table'] == 'availability_rule': self._perform_availability_rule(cursor, data['kind'], params) else: cursor.execute(sql, params) connection.commit() cursor.close() connection.close() def _perform_availability_rule(self, cursor, kind, params): if kind == 'update': cursor.execute("""DELETE FROM availability WHERE availability_rule_id = %(id)s""", params) if kind in ['insert', 'update']: days_of_week = params['days_of_week'].replace('{', '').replace('}', '').split(',') for day_of_week in days_of_week: params['day_of_week'] = day_of_week cursor.execute( """ SELECT availability_rule_bulk_insert(ar, %(day_of_week)s) FROM availability_rule ar WHERE ar.id = %(id)s """, params)

Reading the Changes 47 def __init__(self): self.connection = psycopg2.connect("dbname=schedule", connection_factory=psycopg2.extras.LogicalReplicationConnection, ) def __call__(self, msg): payload = json.loads(msg.payload, strict=False) print(payload) self._consume_change(payload) msg.cursor.send_feedback(flush_lsn=msg.data_start)

Reading the Changes 48 reader = StreamReader() cursor = reader.connection.cursor() cursor.start_replication(slot_name='schedule', decode=True) try: cursor.consume_stream(reader) except KeyboardInterrupt: print("Stopping reader...") finally: cursor.close() reader.connection.close() print("Exiting reader")

• A consumer of the logical stream can only read one change at a time • If our processing of a change takes a lot of time, it will create a backlog of changes • Backlog means the PostgreSQL server needs to retain more WAL logs • Retaining too many WAL logs can lead to running out of disk space • Running out of disk space can lead to...rough times. The Consumer Bottleneck 49 🌩 🌤 🌥 ☁

Can we move any processing to a separate part of the application?

• Can utilize a durable message queueing system to store any WAL changes that are necessary to perform post-processing on • Ensure the changes are worked on in order • "Divide-and-conquer" workload - have multiple workers acting on different "topics" • Remove WAL bloat Shifting the Workload 51

• Durable message processing and distribution system • Streams • Supports parallelization of consumers • Multiple consumers, partitions • Highly-available, distributed architecture • Acknowledgement of receiving, processing messages; can replay (sounds like WAL?) Apache Kafka 52

WAL Consumer 54 import json, sys from kafka import KafkaProducer from kafka.errors import KafkaError import psycopg2 import psycopg2.extras TABLES = set([ 'availability', 'availability_rule', 'room', 'unavailability', ]) reader = WALConsumer() cursor = reader.connection.cursor() cursor.start_replication(slot_name='schedule', decode=True) try: cursor.consume_stream(reader) except KeyboardInterrupt: print("Stopping reader...") finally: cursor.close() reader.connection.close() print("Exiting reader")

WAL Consumer 55 class WALConsumer(object): def __init__(self): self.connection = psycopg2.connect("dbname=realtime", connection_factory=psycopg2.extras.LogicalReplicationConnection, ) self.producer = producer = KafkaProducer( bootstrap_servers=['localhost:9092'], value_serializer=lambda m: json.dumps(m).encode('ascii'), ) def __call__(self, msg): payload = json.loads(msg.payload, strict=False) print(payload) # determine if the payload should be passed on to a consumer listening # to the Kafka que for data in payload['change']: if data.get('table') in TABLES: self.producer.send(data.get('table'), data) # ensure everything is sent; call flush at this point self.producer.flush() # acknowledge that the change has been read - tells PostgreSQL to stop # holding onto this log file msg.cursor.send_feedback(flush_lsn=msg.data_start)

Kafka Consumer 56 import json from kafka import KafkaConsumer from kafka.structs import OffsetAndMetadata, TopicPartition import psycopg2 class Worker(object): """Base class to work perform any post processing on changes""" OPERATIONS = set([]) # override with "insert", "update", "delete" def __init__(self, topic): # connect to the PostgreSQL database self.connection = psycopg2.connect("dbname=realtime") # connect to Kafka self.consumer = KafkaConsumer( bootstrap_servers=['localhost:9092'], value_deserializer=lambda m: json.loads(m.decode('utf8')), auto_offset_reset="earliest", group_id='1') # subscribe to the topic(s) self.consumer.subscribe(topic if isinstance(topic, list) else [topic])

Kafka Consumer 57 def run(self): """Function that runs ad-infinitum""" # loop through the payloads from the consumer # determine if there are any follow-up actions based on the kind of # operation, and if so, act upon it # always commit when done. for msg in self.consumer: print(msg) # load the data from the message data = msg.value # determine if there are any follow-up operations to perform if data['kind'] in self.OPERATIONS: # open up a cursor for interacting with PostgreSQL cursor = self.connection.cursor() # put the parameters in an easy to digest format params = dict(zip(data['columnnames'], data['columnvalues'])) # all the function getattr(self, data['kind'])(cursor, params) # commit any work that has been done, and close the cursor self.connection.commit() cursor.close() # acknowledge the message has been handled tp = TopicPartition(msg.topic, msg.partition) offsets = {tp: OffsetAndMetadata(msg.offset, None)} self.consumer.commit(offsets=offsets)

Kafka Consumer 58 # override with the appropriate post-processing code def insert(self, cursor, params): """Override with any post-processing to be done on an ``INSERT``""" raise NotImplementedError() def update(self, cursor, params): """Override with any post-processing to be done on an ``UPDATE``""" raise NotImplementedError() def delete(self, cursor, params): """Override with any post-processing to be done on an ``DELETE``""" raise NotImplementedError()

• Logical decoding allows the bulk inserts to occur significantly faster from a transactional view • DELETEs are tricky if you need to do anything other than using the PRIMARY KEY • Can bucket changes by topic • Potential bottleneck for long running execution, but bottlenecks are isolated to specific queues Lessons 60

Conclusion 61 • PostgreSQL is robust • Triggers will keep your data in sync but can have significant performance overhead • Utilizing a logical replication slot can eliminate trigger overhead and transfer the computational load elsewhere • Not a panacea: still need to use good architectural patterns!

jonathan.katz@crunchydata.com @jkatz05 Thank You! Questions?

Appendix A: Schema for Example

Managing Availability 65 CREATE TABLE room ( id int GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text NOT NULL ); CREATE TABLE availability_rule ( id int GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, room_id int NOT NULL REFERENCES room (id) ON DELETE CASCADE, days_of_week int[] NOT NULL, start_time time NOT NULL, end_time time NOT NULL, generate_weeks_into_future int NOT NULL DEFAULT 52 );

Managing Availability 66 CREATE TABLE availability ( id int GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, room_id int NOT NULL REFERENCES room (id) ON DELETE CASCADE, availability_rule_id int NOT NULL REFERENCES availability_rule (id) ON DELETE CASCADE, available_date date NOT NULL, available_range tstzrange NOT NULL ); CREATE INDEX availability_available_range_gist_idx ON availability USING gist(available_range);

Managing Availability 67 CREATE TABLE unavailability ( id int GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, room_id int NOT NULL REFERENCES room (id) ON DELETE CASCADE, unavailable_date date NOT NULL, unavailable_range tstzrange NOT NULL ); CREATE INDEX unavailability_unavailable_range_gist_idx ON unavailability USING gist(unavailable_range);

Managing Availability 68 CREATE TABLE calendar ( id int GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, room_id int NOT NULL REFERENCES room (id) ON DELETE CASCADE, status text NOT NULL, calendar_date date NOT NULL, calendar_range tstzrange NOT NULL ); CREATE INDEX calendar_room_id_calendar_date_idx ON calendar (room_id, calendar_date);

Appendix B: Finding Availability for a Room

70 /** AVAILABILITY, UNAVAILABILITY, and CALENDAR */ /** We need some lengthy functions to help generate the calendar */

71 /** Helper function: generate the available chunks of time within a block of time for a day within a calendar */ CREATE OR REPLACE FUNCTION calendar_generate_available(room_id int, calendar_range tstzrange) RETURNS TABLE(status text, calendar_range tstzrange) AS $$ WITH RECURSIVE availables AS ( SELECT 'closed' AS left_status, CASE WHEN availability.id IS NULL THEN tstzrange(calendar_date, calendar_date + '1 day'::interval) ELSE tstzrange( calendar_date, lower(availability.available_range * tstzrange(calendar_date, calendar_date + '1 day'::interval)) ) END AS left_range, CASE isempty(availability.available_range * tstzrange(calendar_date, calendar_date + '1 day'::interval)) WHEN TRUE THEN 'closed' ELSE 'available' END AS center_status, availability.available_range * tstzrange(calendar_date, calendar_date + '1 day'::interval) AS center_range, 'closed' AS right_status, CASE WHEN availability.id IS NULL THEN tstzrange(calendar_date, calendar_date + '1 day'::interval) ELSE tstzrange( upper(availability.available_range * tstzrange(calendar_date, calendar_date + '1 day'::interval)), calendar_date + '1 day'::interval ) END AS right_range FROM generate_series(lower($2), upper($2), '1 day'::interval) AS calendar_date LEFT OUTER JOIN availability ON availability.room_id = $1 AND availability.available_range && $2 UNION SELECT 'closed' AS left_status, CASE WHEN availability.available_range && availables.left_range THEN tstzrange( lower(availables.left_range), lower(availables.left_range * availability.available_range) ) ELSE tstzrange( lower(availables.right_range), lower(availables.right_range * availability.available_range) ) END AS left_range, CASE WHEN availability.available_range && availables.left_range OR availability.available_range && availables.right_range THEN 'available' ELSE 'closed' END AS center_status, CASE WHEN availability.available_range && availables.left_range THEN availability.available_range * availables.left_range ELSE availability.available_range * availables.right_range END AS center_range, 'closed' AS right_status, CASE WHEN availability.available_range && availables.left_range THEN tstzrange( upper(availables.left_range * availability.available_range), upper(availables.left_range) ) ELSE tstzrange( upper(availables.right_range * availability.available_range), upper(availables.right_range) ) END AS right_range FROM availables JOIN availability ON availability.room_id = $1 AND availability.available_range && $2 AND availability.available_range <> availables.center_range AND ( availability.available_range && availables.left_range OR availability.available_range && availables.right_range ) ) SELECT * FROM ( SELECT x.left_status AS status, x.left_range AS calendar_range FROM availables x LEFT OUTER JOIN availables y ON x.left_range <> y.left_range AND x.left_range @> y.left_range GROUP BY 1, 2 HAVING NOT bool_or(COALESCE(x.left_range @> y.left_range, FALSE)) UNION SELECT DISTINCT x.center_status AS status, x.center_range AS calendar_range FROM availables x UNION SELECT x.right_status AS status, x.right_range AS calendar_range FROM availables x LEFT OUTER JOIN availables y ON x.right_range <> y.right_range AND x.right_range @> y.right_range GROUP BY 1, 2 HAVING NOT bool_or(COALESCE(x.right_range @> y.right_range, FALSE)) ) x WHERE NOT isempty(x.calendar_range) AND NOT lower_inf(x.calendar_range) AND NOT upper_inf(x.calendar_range) AND x.calendar_range <@ $2 $$ LANGUAGE SQL STABLE; This is the first of two helpers functions...

• We will have two availability rules: • Open every day 8am - 8pm • Open every day 9pm - 10:30pm For this experiment 72

73 INSERT INTO room (name) VALUES ('Test Room'); INSERT INTO availability_rule (room_id, days_of_week, start_time, end_time) VALUES (1, ARRAY[1,2,3,4,5,6,7], '08:00', '20:00'), (1, ARRAY[1,2,3,4,5,6,7], '21:00', '22:30');

74 /** Helper function: generate the available chunks of time within a block of time for a day within a calendar */ CREATE OR REPLACE FUNCTION calendar_generate_available(room_id int, calendar_range tstzrange) RETURNS TABLE(status text, calendar_range tstzrange) AS $$

75 WITH RECURSIVE availables AS ( SELECT 'closed' AS left_status, CASE WHEN availability.id IS NULL THEN tstzrange(calendar_date, calendar_date + '1 day'::interval) ELSE tstzrange( calendar_date, lower(availability.available_range * tstzrange(calendar_date, calendar_date + '1 day'::interval)) ) END AS left_range, CASE isempty(availability.available_range * tstzrange(calendar_date, calendar_date + '1 day'::interval)) WHEN TRUE THEN 'closed' ELSE 'available' END AS center_status, availability.available_range * tstzrange(calendar_date, calendar_date + '1 day'::interval) AS center_range, 'closed' AS right_status, CASE WHEN availability.id IS NULL THEN tstzrange(calendar_date, calendar_date + '1 day'::interval) ELSE tstzrange( upper(availability.available_range * tstzrange(calendar_date, calendar_date + '1 day'::interval)), calendar_date + '1 day'::interval ) END AS right_range FROM generate_series(lower($2), upper($2), '1 day'::interval) AS calendar_date LEFT OUTER JOIN availability ON availability.room_id = $1 AND availability.available_range && $2

77 UNION SELECT 'closed' AS left_status, CASE WHEN availability.available_range && availables.left_range THEN tstzrange( lower(availables.left_range), lower(availables.left_range * availability.available_range) ) ELSE tstzrange( lower(availables.right_range), lower(availables.right_range * availability.available_range) ) END AS left_range, CASE WHEN availability.available_range && availables.left_range OR availability.available_range && availables.right_range THEN 'available' ELSE 'closed' END AS center_status, CASE WHEN availability.available_range && availables.left_range THEN availability.available_range * availables.left_range ELSE availability.available_range * availables.right_range END AS center_range, 'closed' AS right_status, CASE WHEN availability.available_range && availables.left_range THEN tstzrange( upper(availables.left_range * availability.available_range), upper(availables.left_range) ) ELSE tstzrange( upper(availables.right_range * availability.available_range), upper(availables.right_range) ) END AS right_range FROM availables JOIN availability ON availability.room_id = $1 AND availability.available_range && $2 AND availability.available_range <> availables.center_range AND ( availability.available_range && availables.left_range OR availability.available_range && availables.right_range ))

78 UNION SELECT ... FROM availables JOIN availability ON availability.room_id = $1 AND availability.available_range && $2 AND availability.available_range <> availables.center_range AND ( availability.available_range && availables.left_range OR availability.available_range && availables.right_range )

80 'closed' AS left_status, CASE WHEN availability.available_range && availables.left_range THEN tstzrange( lower(availables.left_range), lower(availables.left_range * availability.available_range) ) ELSE tstzrange( lower(availables.right_range), lower(availables.right_range * availability.available_range) ) END AS left_range, CASE WHEN availability.available_range && availables.left_range OR availability.available_range && availables.right_range THEN 'available' ELSE 'closed' END AS center_status, CASE WHEN availability.available_range && availables.left_range THEN availability.available_range * availables.left_range ELSE availability.available_range * availables.right_range END AS center_range, 'closed' AS right_status, CASE WHEN availability.available_range && availables.left_range THEN tstzrange( upper(availables.left_range * availability.available_range), upper(availables.left_range) ) ELSE tstzrange( upper(availables.right_range * availability.available_range), upper(availables.right_range) ) END AS right_range

83 SELECT * FROM ( SELECT x.left_status AS status, x.left_range AS calendar_range FROM availables x LEFT OUTER JOIN availables y ON x.left_range <> y.left_range AND x.left_range @> y.left_range GROUP BY 1, 2 HAVING NOT bool_or(COALESCE(x.left_range @> y.left_range, FALSE)) UNION SELECT DISTINCT x.center_status AS status, x.center_range AS calendar_range FROM availables x UNION SELECT x.right_status AS status, x.right_range AS calendar_range FROM availables x LEFT OUTER JOIN availables y ON x.right_range <> y.right_range AND x.right_range @> y.right_range GROUP BY 1, 2 HAVING NOT bool_or(COALESCE(x.right_range @> y.right_range, FALSE)) ) x WHERE NOT isempty(x.calendar_range) AND NOT lower_inf(x.calendar_range) AND NOT upper_inf(x.calendar_range) AND x.calendar_range <@ $2 $$ LANGUAGE SQL STABLE;

Building a Complex, Real-Time Data Management Application

More Related Content

What's hot

Similar to Building a Complex, Real-Time Data Management Application

More from Jonathan Katz

Recently uploaded

In this document

Building a Complex, Real-Time Data Management Application