duckdb skeleton

duckdb/.gitignore

@@ -0,0 +1,2 @@
+*.duckdb*
+ldbc_parquets/

duckdb/README.md

@@ -0,0 +1,111 @@
+# LDBC SNB BI Umbra implementation
+
+[Umbra](https://umbra-db.com/) implementation of the [LDBC Social Network Benchmark's BI workload](https://github.com/ldbc/ldbc_snb_docs).
+
+:warning: Note that while this implementation uses containerized execution for Umbra, it does not work on MacOS (neither on x86, nor on Apple Silicon).
+
+## Getting the container
+
+The Umbra container is currently available upon request.
+
+## Generating the data set
+
+The Umbra implementation expects the data to be in `composite-merged-fk` CSV layout, with headers and without quoted fields.
+To generate data that confirms this requirement, run Datagen without any layout or formatting arguments (`--explode-*` or `--format-options`).
+
+In Datagen's directory (`ldbc_snb_datagen_spark`), issue the following commands. We assume that the Datagen project is built and the `${PLATFORM_VERSION}`, `${DATAGEN_VERSION}` environment variables are set correctly.
+
+```bash
+export SF=desired_scale_factor
+export LDBC_SNB_DATAGEN_MAX_MEM=available_memory
+export LDBC_SNB_DATAGEN_JAR=$(sbt -batch -error 'print assembly / assemblyOutputPath')
+```
+
+```bash
+rm -rf out-sf${SF}/
+tools/run.py \
+    --cores $(nproc) \
+    --memory ${LDBC_SNB_DATAGEN_MAX_MEM} \
+    -- \
+    --format csv \
+    --scale-factor ${SF} \
+    --mode bi \
+    --output-dir out-sf${SF}
+```
+
+## Loading the data set
+
+Note that unlike Postgres, Umbra does not support directly loading from compressed files (`.csv.gz`).
+
+1. Set the `${UMBRA_CSV_DIR}` environment variable to point to the data set.
+
+    * To use a locally generated data set, set the `${LDBC_SNB_DATAGEN_DIR}` and `${SF}` environment variables and run:
+
+        ```bash
+        export UMBRA_CSV_DIR=${LDBC_SNB_DATAGEN_DIR}/out-sf${SF}/graphs/csv/bi/composite-merged-fk/
+        ```
+
+        Or, simply run:
+
+        ```bash
+        . scripts/use-datagen-data-set.sh
+        ```
+
+    * To download and use the sample data set, run:
+
+        ```bash
+        wget -q https://ldbcouncil.org/ldbc_snb_datagen_spark/social-network-sf0.003-bi-composite-merged-fk.zip
+        unzip -q social-network-sf0.003-bi-composite-merged-fk.zip
+        export UMBRA_CSV_DIR=`pwd`/social-network-sf0.003-bi-composite-merged-fk/graphs/csv/bi/composite-merged-fk/
+        ```
+
+        Or, simply run:
+
+        ```
+        scripts/get-sample-data-set.sh
+        . scripts/use-sample-data-set.sh
+        ```
+
+1. The data set should consist of uncompressed CSVs. If you retrieved a compressed data set (`.csv.gz` files), set the `${UMBRA_CSV_DIR}` environment variable and uncompress the files (note that doing so deletes the original compressed files):
+
+    ```bash
+    scripts/decompress-data-set.sh
+    ```
+
+1. To start the DBMS, create a database and load the data, run:
+
+    ```bash
+    scripts/load-in-one-step.sh
+    ```
+
+1. The substitution parameters should be generated using the [`paramgen`](../paramgen).
+
+## Queries
+
+To run the queries, issue:
+
+```bash
+scripts/queries.sh
+```
+
+For a test run, use:
+
+```bash
+scripts/queries.sh ${SF} --test
+```
+
+## Benchmark
+
+To run the queries and the batches alternately, as specified by the benchmark, run:
+
+```bash
+scripts/benchmark.sh
+```
+
+## Running queries interactively
+
+To connect to the database through the SQL console, use:
+
+```bash
+scripts/connect.sh
+```

duckdb/benchmark.py

@@ -0,0 +1,223 @@
+import csv
+import datetime
+from dateutil.relativedelta import relativedelta
+import os
+import psycopg2
+import time
+from queries import run_script, run_queries, run_precomputations, load_mht, load_plm, load_post
+from pathlib import Path
+from itertools import cycle
+import argparse
+
+
+def execute(cur, query):
+    cur.execute(query)
+
+
+query_variants = ["1", "2a", "2b", "3", "4", "5", "6", "7", "8a", "8b", "9", "10a", "10b", "11", "12", "13", "14a", "14b", "15a", "15b", "16a", "16b", "17", "18", "19a", "19b", "20a", "20b"]
+
+def run_batch_updates(pg_con, data_dir, batch_date, batch_type, timings_file):
+    batch_id = batch_date.strftime('%Y-%m-%d')
+    batch_dir = f"batch_id={batch_id}"
+    print(f"#################### {batch_dir} ####################")
+
+    start = time.time()
+
+    print("## Inserts")
+    for entity in insert_entities:
+        batch_path = f"{data_dir}/inserts/dynamic/{entity}/{batch_dir}"
+        if not os.path.exists(batch_path):
+            continue
+
+        for csv_file in [f for f in os.listdir(batch_path) if f.endswith(".csv")]:
+            csv_path = f"{batch_path}/{csv_file}"
+            print(f"- {csv_path}")
+            execute(cur, "BEGIN BULK WRITE;")
+            execute(cur, f"COPY {entity} FROM '{dbs_data_dir}/inserts/dynamic/{entity}/{batch_dir}/{csv_file}' (DELIMITER '|', HEADER, NULL '', FORMAT text)")
+            execute(cur, "COMMIT;")
+            if entity == "Person_knows_Person":
+                execute(cur, "BEGIN BULK WRITE;")
+                execute(cur, f"COPY {entity} (creationDate, Person2id, Person1id) FROM '{dbs_data_dir}/inserts/dynamic/{entity}/{batch_dir}/{csv_file}' (DELIMITER '|', HEADER, NULL '', FORMAT text)")
+                execute(cur, "COMMIT;")
+
+    for entity in ["Comment_hasTag_Tag", "Post_hasTag_Tag"]:
+        batch_path = f"{data_dir}/inserts/dynamic/{entity}/{batch_dir}"
+        if not os.path.exists(batch_path):
+            continue
+        for csv_file in [f for f in os.listdir(batch_path) if f.endswith(".csv")]:
+            csvpath = f"{dbs_data_dir}/inserts/dynamic/{entity}/{batch_dir}/{csv_file}"
+            load_mht(cur, csvpath)
+
+    for entity in ["Person_likes_Comment", "Person_likes_Post"]:
+        batch_path = f"{data_dir}/inserts/dynamic/{entity}/{batch_dir}"
+        if not os.path.exists(batch_path):
+            continue
+        for csv_file in [f for f in os.listdir(batch_path) if f.endswith(".csv")]:
+            csvpath = f"{dbs_data_dir}/inserts/dynamic/{entity}/{batch_dir}/{csv_file}"
+            load_plm(cur, csvpath)
+
+    for entity in ["Post"]:
+        batch_path = f"{data_dir}/inserts/dynamic/{entity}/{batch_dir}"
+        if not os.path.exists(batch_path):
+            continue
+        for csv_file in [f for f in os.listdir(batch_path) if f.endswith(".csv")]:
+            csvpath = f"{dbs_data_dir}/inserts/dynamic/{entity}/{batch_dir}/{csv_file}"
+            load_post(cur, csvpath)
+
+
+    print("## Deletes")
+    # Deletes are implemented using a SQL script which use auxiliary tables.
+    # Entities to be deleted are first put into {entity}_Delete_candidate tables.
+    # These are cleaned up before running the delete script.
+    for entity in delete_entities:
+        execute(cur, "BEGIN BULK WRITE;")
+        execute(cur, f"DELETE FROM {entity}_Delete_candidates")
+        execute(cur, "COMMIT;")
+
+        batch_path = f"{data_dir}/deletes/dynamic/{entity}/{batch_dir}"
+        if not os.path.exists(batch_path):
+            continue
+
+        for csv_file in [f for f in os.listdir(batch_path) if f.endswith(".csv")]:
+            csv_path = f"{batch_path}/{csv_file}"
+            print(f"- {csv_path}")
+            execute(cur, "BEGIN BULK WRITE;")
+            execute(cur, f"COPY {entity}_Delete_candidates FROM '{dbs_data_dir}/deletes/dynamic/{entity}/{batch_dir}/{csv_file}' (DELIMITER '|', HEADER, NULL '', FORMAT text)")
+            execute(cur, "COMMIT;")
+
+    print("Maintain materialized views . . .")
+    run_script(pg_con, cur, "dml/maintain-views.sql")
+    print("Done.")
+    print()
+
+    print("Apply deletes . . .")
+    # Invoke delete script which makes use of the {entity}_Delete_candidates tables
+    run_script(pg_con, cur, "dml/apply-deletes.sql")
+    print("Done.")
+    print()
+
+    print("Apply precomp . . .")
+    run_precomputations(query_variants, pg_con, cur, batch_id, batch_type, sf, timings_file)
+    print("Done.")
+    print()
+
+    end = time.time()
+    duration = end - start
+    timings_file.write(f"Umbra|{sf}|{batch_id}|{batch_type}|writes||{duration}\n")
+
+
+parser = argparse.ArgumentParser()
+parser.add_argument('--scale_factor', type=str, help='Scale factor', required=True)
+parser.add_argument('--test', action='store_true', help='Test execution: 1 query/batch', required=False)
+parser.add_argument('--validate', action='store_true', help='Validation mode', required=False)
+parser.add_argument('--pgtuning', action='store_true', help='Paramgen tuning execution: 100 queries/batch', required=False)
+parser.add_argument('--local', action='store_true', help='Local run (outside of a container)', required=False)
+parser.add_argument('--data_dir', type=str, help='Directory with the initial_snapshot, insert, and delete directories', required=True)
+parser.add_argument('--param_dir', type=str, help='Directory with the initial_snapshot, insert, and delete directories')
+parser.add_argument('--queries', action='store_true', help='Only run queries', required=False)
+args = parser.parse_args()
+sf = args.scale_factor
+test = args.test
+pgtuning = args.pgtuning
+local = args.local
+data_dir = args.data_dir
+queries_only = args.queries
+validate = args.validate
+if args.param_dir is not None:
+    param_dir = args.param_dir
+else:
+    param_dir = f'../parameters/parameters-sf{sf}'
+
+
+if local:
+    dbs_data_dir = data_dir
+else:
+    dbs_data_dir = '/data'
+
+parameter_csvs = {}
+for query_variant in query_variants:
+    # wrap parameters into infinite loop iterator
+    parameter_csvs[query_variant] = cycle(csv.DictReader(open(f'{param_dir}/bi-{query_variant}.csv'), delimiter='|'))
+
+print(f"- Input data directory, ${{UMBRA_CSV_DIR}}: {data_dir}")
+
+insert_nodes = ["Forum", "Person", "Comment"]
+insert_edges = ["Forum_hasMember_Person", "Forum_hasTag_Tag", "Person_hasInterest_Tag", "Person_knows_Person", "Person_studyAt_University", "Person_workAt_Company"]
+insert_entities = insert_nodes + insert_edges
+
+# set the order of deletions to reflect the dependencies between node labels (:Comment)-[:REPLY_OF]->(:Post)<-[:CONTAINER_OF]-(:Forum)-[:HAS_MODERATOR]->(:Person)
+delete_nodes = ["Comment", "Post", "Forum", "Person"]
+delete_edges = ["Forum_hasMember_Person", "Person_knows_Person", "Person_likes_Comment", "Person_likes_Post"]
+delete_entities = delete_nodes + delete_edges
+
+output = Path(f"output/output-sf{sf}")
+output.mkdir(parents=True, exist_ok=True)
+open(f"output/output-sf{sf}/results.csv", "w").close()
+open(f"output/output-sf{sf}/timings.csv", "w").close()
+
+timings_file = open(f"output/output-sf{sf}/timings.csv", "a")
+timings_file.write(f"tool|sf|day|batch_type|q|parameters|time\n")
+results_file = open(f"output/output-sf{sf}/results.csv", "a")
+
+pg_con = psycopg2.connect(host="localhost", user="postgres", password="mysecretpassword", port=8000)
+pg_con.autocommit = True
+cur = pg_con.cursor()
+
+network_start_date = datetime.date(2012, 11, 29)
+network_end_date = datetime.date(2013, 1, 1)
+test_end_date = datetime.date(2012, 12, 2)
+batch_size = relativedelta(days=1)
+batch_date = network_start_date
+
+benchmark_start = time.time()
+
+run_script(pg_con, cur, f"ddl/schema-delete-candidates.sql")
+
+if queries_only:
+    run_queries(query_variants, parameter_csvs, pg_con, sf, test, pgtuning, batch_date, "power", timings_file, results_file)
+else:
+    # Run alternating write-read blocks.
+    # The first write-read block is the power batch, while the rest are the throughput batches.
+    current_batch = 1
+    while batch_date < network_end_date and \
+          (not test or batch_date < test_end_date) and \
+          (not validate or batch_date == network_start_date):
+        if current_batch == 1:
+            batch_type = "power"
+        else:
+            batch_type = "throughput"
+        print()
+        print(f"----------------> Batch date: {batch_date}, batch type: {batch_type} <---------------")
+
+        if current_batch == 2:
+            start = time.time()
+
+        run_batch_updates(pg_con, data_dir, batch_date, batch_type, timings_file)
+        reads_time = run_queries(query_variants, parameter_csvs, pg_con, sf, test, pgtuning, batch_date, batch_type, timings_file, results_file)
+        timings_file.write(f"Umbra|{sf}|{batch_date}|{batch_type}|reads||{reads_time:.6f}\n")
+
+        # checking if 1 hour (and a bit) has elapsed for the throughput batches
+        if current_batch >= 2:
+            end = time.time()
+            duration = end - start
+            if duration > 3605:
+                print("""Throughput batches finished successfully. Termination criteria met:
+                    - At least 1 throughput batch was executed
+                    - The total execution time of the throughput batch(es) was at least 1h""")
+                break
+
+        current_batch = current_batch + 1
+        batch_date = batch_date + batch_size
+
+cur.close()
+pg_con.close()
+
+benchmark_end = time.time()
+benchmark_duration = benchmark_end - benchmark_start
+benchmark_file = open(f"output/output-sf{sf}/benchmark.csv", "w")
+benchmark_file.write(f"time\n")
+benchmark_file.write(f"{benchmark_duration:.6f}\n")
+benchmark_file.close()
+
+timings_file.close()
+results_file.close()

duckdb/ddl/drop-tables.sql

@@ -0,0 +1,35 @@
+DROP VIEW IF EXISTS Comment_View;
+DROP VIEW IF EXISTS Post_View;
+
+DROP TABLE IF EXISTS Organisation;
+DROP TABLE IF EXISTS Place;
+DROP TABLE IF EXISTS Tag;
+DROP TABLE IF EXISTS TagClass;
+DROP TABLE IF EXISTS Company;
+DROP TABLE IF EXISTS University;
+DROP TABLE IF EXISTS Continent;
+DROP TABLE IF EXISTS Country;
+DROP TABLE IF EXISTS City;
+DROP TABLE IF EXISTS Comment;
+DROP TABLE IF EXISTS Forum;
+DROP TABLE IF EXISTS Post;
+DROP TABLE IF EXISTS Person;
+DROP TABLE IF EXISTS Comment_hasTag_Tag;
+DROP TABLE IF EXISTS Post_hasTag_Tag;
+DROP TABLE IF EXISTS Forum_hasMember_Person;
+DROP TABLE IF EXISTS Forum_hasTag_Tag;
+DROP TABLE IF EXISTS Person_hasInterest_Tag;
+DROP TABLE IF EXISTS Person_likes_Comment;
+DROP TABLE IF EXISTS Person_likes_Post;
+DROP TABLE IF EXISTS Person_studyAt_University;
+DROP TABLE IF EXISTS Person_workAt_Company;
+DROP TABLE IF EXISTS Person_knows_Person;
+
+DROP TABLE IF EXISTS Message;
+DROP TABLE IF EXISTS Message_hasTag_Tag;
+DROP TABLE IF EXISTS Person_likes_Message;
+
+DROP TABLE IF EXISTS Top100PopularForumsQ04;
+DROP TABLE IF EXISTS PopularityScoreQ06;
+DROP TABLE IF EXISTS PathQ19;
+DROP TABLE IF EXISTS PathQ20;

duckdb/ddl/output-env.sql

@@ -0,0 +1,3 @@
+
+show debug.buffersize;
+show debug.usedirectio;