Implementation of an immutable database in postgres

_ImmutableDatabase

create_alternative_views.sql

-- An alternative style of creating a user view
-- that leverages group-by and aggregates
CREATE OR REPLACE VIEW users_group_by AS (
SELECT user_base.id
       ,first_not_null(email ORDER BY user_credentials.time DESC) AS email
       ,first_not_null(firstname ORDER BY user_metadata.time DESC) AS firstname
       ,first_not_null(lastname ORDER BY user_metadata.time DESC) AS lastname

FROM user_base
LEFT JOIN user_deletion ON user_deletion.user_id=user_base.id
LEFT JOIN user_credentials ON user_credentials.user_id=user_base.id
LEFT JOIN user_metadata ON user_metadata.user_id=user_base.id
WHERE user_deletion.id IS NULL

GROUP BY user_base.id
);

-- Create a view that represents a user using only
-- the most recently added data, regardless of whether
-- it is null or not
CREATE OR REPLACE VIEW users_strictly_most_recent AS (
SELECT user_base.*,
       credentials.*,
       metadata.*
FROM user_base
LEFT JOIN LATERAL (
     SELECT email
     FROM user_credentials
     WHERE user_id=user_base.id
     ORDER BY time DESC
     LIMIT 1
) AS credentials ON true
LEFT JOIN LATERAL (
     SELECT firstname, lastname
     FROM user_metadata
     WHERE user_id=user_base.id
     ORDER BY time DESC
     LIMIT 1
) AS metadata ON true
);

create_base_user.sql

DROP SEQUENCE IF EXISTS id_sequence CASCADE;
CREATE SEQUENCE id_sequence START 101;

-- Core user object that is created
-- when the user is created
DROP TABLE IF EXISTS user_base CASCADE;
CREATE TABLE user_base (
       id INTEGER PRIMARY KEY DEFAULT nextval('id_sequence'),
       time TIMESTAMP WITH TIME ZONE DEFAULT clock_timestamp() -- NOW()
);

CREATE INDEX user_base_time_idx ON user_base(time);
CREATE INDEX user_base_id_time_idx ON user_base(id, time);

create_functions.sql

CREATE OR REPLACE FUNCTION first_or_null(anyelement, anyelement) RETURNS anyelement AS
$$
SELECT coalesce($1, $2);
$$ LANGUAGE SQL;


-- Polymorphic aggregate function
DROP AGGREGATE first_not_null (anyelement);
CREATE AGGREGATE first_not_null (anyelement)
(
    sfunc = first_or_null,
    stype = anyelement
);

create_tables.sql

DROP SEQUENCE IF EXISTS id_sequence CASCADE;
CREATE SEQUENCE id_sequence START 101;


-- Core user object that is created
-- when the user is created
DROP TABLE IF EXISTS user_base CASCADE;
CREATE TABLE user_base (
       id INTEGER PRIMARY KEY DEFAULT nextval('id_sequence'),
       time TIMESTAMP WITH TIME ZONE DEFAULT clock_timestamp() -- NOW()
);

CREATE INDEX user_base_time_idx ON user_base(time);
CREATE INDEX user_base_id_time_idx ON user_base(id, time);


DROP TABLE IF EXISTS user_credentials CASCADE;
CREATE TABLE user_credentials (
       id INTEGER PRIMARY KEY DEFAULT nextval('id_sequence'),
       time TIMESTAMP WITH TIME ZONE DEFAULT clock_timestamp(), --NOW(),
       user_id INTEGER REFERENCES user_base(id),
       email TEXT,
       password_hash TEXT
);

CREATE INDEX user_credentials_time_idx ON user_credentials(time);
CREATE INDEX user_credentials_id_time_idx ON user_credentials(id, time);
CREATE INDEX user_credentials_user_id_idx ON user_credentials(user_id);
CREATE INDEX user_credentials_user_id_time_idx ON user_credentials(user_id, time);

DROP TABLE IF EXISTS user_metadata CASCADE;
CREATE TABLE user_metadata (
       id integer PRIMARY KEY DEFAULT nextval('id_sequence'),
       time TIMESTAMP WITH TIME ZONE DEFAULT clock_timestamp(), --NOW(),
       user_id integer REFERENCES user_base(id),
       firstname TEXT,
       lastname TEXT
);

CREATE INDEX user_metadata_time_idx ON user_metadata(time);
CREATE INDEX user_metadata_id_time_idx ON user_metadata(id, time);
CREATE INDEX user_metadata_user_id_idx_idx ON user_metadata(user_id);
CREATE INDEX user_metadata_user_id_idx_time_idx ON user_metadata(user_id, time);


DROP TABLE IF EXISTS user_deletion CASCADE;
CREATE TABLE user_deletion (
       id integer PRIMARY KEY DEFAULT nextval('id_sequence'),
       time TIMESTAMP WITH TIME ZONE DEFAULT clock_timestamp(),
       user_id integer REFERENCES user_base(id),
       is_deleted boolean DEFAULT TRUE
);

CREATE INDEX user_deletion_id_idx ON user_deletion(id);
CREATE INDEX user_deletion_time_idx ON user_deletion(time);
CREATE INDEX user_deletion_user_id_idx ON user_deletion(user_id);
CREATE INDEX user_deletion_is_deleted_idx ON user_deletion(is_deleted)
CREATE INDEX user_deletion_user_id_time_idx ON user_deletion(user_id, time);
CREATE INDEX user_deletion_user_id_time_is_deleted_idx ON user_deletion(user_id, time, is_deleted);

create_user_credentials.sql

DROP TABLE IF EXISTS user_credentials CASCADE;
CREATE TABLE user_credentials (
       id INTEGER PRIMARY KEY DEFAULT nextval('id_sequence'),
       time TIMESTAMP WITH TIME ZONE DEFAULT clock_timestamp(), --NOW(),
       user_id INTEGER REFERENCES user_base(id),
       email TEXT,
       password_hash TEXT
);

CREATE INDEX user_credentials_time_idx ON user_credentials(time);
CREATE INDEX user_credentials_id_time_idx ON user_credentials(id, time);
CREATE INDEX user_credentials_user_id_idx ON user_credentials(user_id);
CREATE INDEX user_credentials_user_id_time_idx ON user_credentials(user_id, time);

create_user_deletion.sql

DROP TABLE IF EXISTS user_deletion CASCADE;
CREATE TABLE user_deletion (
       id integer PRIMARY KEY DEFAULT nextval('id_sequence'),
       time TIMESTAMP WITH TIME ZONE DEFAULT clock_timestamp(),
       user_id integer REFERENCES user_base(id),
       is_deleted boolean DEFAULT TRUE
);

CREATE INDEX user_deletion_id_idx ON user_deletion(id);
CREATE INDEX user_deletion_time_idx ON user_deletion(time);
CREATE INDEX user_deletion_user_id_idx ON user_deletion(user_id);
CREATE INDEX user_deletion_is_deleted_idx ON user_deletion(is_deleted);
CREATE INDEX user_deletion_user_id_time_idx ON user_deletion(user_id, time);
CREATE INDEX user_deletion_user_id_time_is_deleted_idx ON user_deletion(user_id, time, is_deleted);

create_user_metadata.sql

DROP TABLE IF EXISTS user_metadata CASCADE;
CREATE TABLE user_metadata (
       id integer PRIMARY KEY DEFAULT nextval('id_sequence'),
       time TIMESTAMP WITH TIME ZONE DEFAULT clock_timestamp(), --NOW(),
       user_id integer REFERENCES user_base(id),
       firstname TEXT,
       lastname TEXT
);

CREATE INDEX user_metadata_time_idx ON user_metadata(time);
CREATE INDEX user_metadata_id_time_idx ON user_metadata(id, time);
CREATE INDEX user_metadata_user_id_idx_idx ON user_metadata(user_id);
CREATE INDEX user_metadata_user_id_idx_time_idx ON user_metadata(user_id, time);

create_user_view.sql

-- Create a view that represents a user such that
-- every value is the most recent non-null value
-- for each requested column (if such a value exists)
CREATE OR REPLACE VIEW users AS (
SELECT user_base.*,
       credentials.*,
       metadata.*
FROM user_base
LEFT JOIN user_deletion ON user_deletion.user_id=user_base.id
LEFT JOIN LATERAL (
     SELECT
     first_not_null(email ORDER BY time DESC) AS email
     FROM user_credentials
     WHERE user_id=user_base.id
) AS credentials ON true
LEFT JOIN LATERAL (
     SELECT
     first_not_null(firstname ORDER BY TIME DESC) AS firstname,
     first_not_null(lastname ORDER BY TIME DESC) AS lastname
     FROM user_metadata
     WHERE user_id=user_base.id
) AS metadata ON true
WHERE user_deletion.id IS NULL
);

generate_data.py

import random
import hashlib

import argparse
import psycopg2

from faker import Faker


def add_credentials(cur, user_id, fake):
    email = fake.email()
    password = hashlib.sha224(email).hexdigest()
    cur.execute("INSERT INTO user_credentials VALUES (DEFAULT, DEFAULT, %s, %s, %s) RETURNING id",
                (user_id, email, password))
    return cur.fetchone()[0]


def add_metadata(cur, user_id, fake):
    firstname, lastname = fake.first_name(), fake.last_name()
    cur.execute("INSERT INTO user_metadata VALUES (DEFAULT, DEFAULT, %s, %s, %s) RETURNING id",
                (user_id, firstname, lastname))
    return cur.fetchone()[0]


def delete_user(cur, user_id):
    cur.execute("INSERT INTO user_deletion VALUES (DEFAULT, DEFAULT, %s, DEFAULT) RETURNING id",
                (user_id,))
    return cur.fetchone()[0]


def random_true(num_true=1, num_false=1):
    return random.choice([True if i < num_true else False for i in range(num_true+num_false)])


def main():

    fake = Faker()

    parser = argparse.ArgumentParser(description='Process some integers.')
    parser.add_argument('--num_users', type=int, default=1000, help='Number of test users to create')
    parser.add_argument('--database', help="Database Name")
    parser.add_argument('--user', help="Database User")
    parser.add_argument('--host', default='localhost', help="Database User")

    args = parser.parse_args()

    with psycopg2.connect(database=args.database, user=args.user, host=args.host) as conn:
        with conn.cursor() as cur:
            for i in range(args.num_users):

                cur.execute("INSERT INTO user_base DEFAULT VALUES RETURNING id")
                user_id = cur.fetchone()[0]

                while random_true(2):
                    add_credentials(cur, user_id, fake)

                while random_true(2):
                    add_metadata(cur, user_id, fake)

                if random_true(1, 5):
                    delete_user(cur, user_id)

                conn.commit()


if __name__ == '__main__':
    main()

run.sh

psql postgres postgres -f create_database.sql

psql immutable_database immutable -f create_functions.sql

psql immutable_database immutable -f create_base_user.sql
psql immutable_database immutable -f create_user_credentials.sql
psql immutable_database immutable -f create_user_metadata.sql
psql immutable_database immutable -f create_user_deletion.sql

psql immutable_database immutable -f create_user_view.sql

python generate_data.py --num_users 1000 --database immutable_database --user immutable