#!/usr/bin/env bash

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PG_BIN="${PG_BIN:-/Applications/Postgres.app/Contents/Versions/latest/bin}"
PUB_PORT="${PUB_PORT:-56432}"
SUB_PORT="${SUB_PORT:-56433}"
CLIENTS="${CLIENTS:-8}"
JOBS="${JOBS:-4}"
TRANSACTIONS_PER_CLIENT="${TRANSACTIONS_PER_CLIENT:-2500}"
ROWS_PER_TRANSACTION=25
FULL_IDENTITY_ROWS="${FULL_IDENTITY_ROWS:-50000}"
FULL_IDENTITY_UPDATES="${FULL_IDENTITY_UPDATES:-2000}"
RUN_IDENTITY_BENCHMARK="${RUN_IDENTITY_BENCHMARK:-true}"

for program in initdb pg_ctl psql pgbench pg_recvlogical pg_isready; do
    if [[ ! -x "$PG_BIN/$program" ]]; then
        printf 'Missing required PostgreSQL program: %s/%s\n' "$PG_BIN" "$program" >&2
        exit 1
    fi
done

for port in "$PUB_PORT" "$SUB_PORT"; do
    if "$PG_BIN/pg_isready" -h 127.0.0.1 -p "$port" -q; then
        printf 'Port %s is already in use. Set PUB_PORT and SUB_PORT to unused ports.\n' "$port" >&2
        exit 1
    fi
done

BENCH_ROOT="$(mktemp -d /private/tmp/logical-repl-bench.XXXXXX)"
PUB_DATA="$BENCH_ROOT/publisher"
SUB_DATA="$BENCH_ROOT/subscriber"
PUB_LOG="$BENCH_ROOT/publisher.log"
SUB_LOG="$BENCH_ROOT/subscriber.log"
PUB_CONN="host=127.0.0.1 port=$PUB_PORT dbname=bench user=postgres"
SUB_CONN="host=127.0.0.1 port=$SUB_PORT dbname=bench user=postgres"
PUB_ADMIN="host=127.0.0.1 port=$PUB_PORT dbname=postgres user=postgres"
SUB_ADMIN="host=127.0.0.1 port=$SUB_PORT dbname=postgres user=postgres"
PUB_STARTED=0
SUB_STARTED=0
RECEIVER_PID=""
SAMPLER_PID=""
SAMPLER_PIDS=()

cleanup() {
    local sampler_pid
    for sampler_pid in "${SAMPLER_PIDS[@]-}"; do
        if [[ -n "$sampler_pid" ]]; then
            kill -TERM "$sampler_pid" 2>/dev/null || true
        fi
    done
    if [[ -n "$RECEIVER_PID" ]]; then
        kill -TERM "$RECEIVER_PID" 2>/dev/null || true
        wait "$RECEIVER_PID" 2>/dev/null || true
    fi
    if [[ "$SUB_STARTED" -eq 1 ]]; then
        "$PG_BIN/pg_ctl" -D "$SUB_DATA" -m fast -w stop >/dev/null 2>&1 || true
    fi
    if [[ "$PUB_STARTED" -eq 1 ]]; then
        "$PG_BIN/pg_ctl" -D "$PUB_DATA" -m fast -w stop >/dev/null 2>&1 || true
    fi
    case "$BENCH_ROOT" in
        /private/tmp/logical-repl-bench.*) rm -rf -- "$BENCH_ROOT" ;;
        *) printf 'Refusing to remove unexpected temporary path: %s\n' "$BENCH_ROOT" >&2 ;;
    esac
}

report_error() {
    local status="$?"
    printf 'Benchmark failed at line %s with exit status %s.\n' "${BASH_LINENO[0]}" "$status" >&2
    return "$status"
}

trap report_error ERR
trap cleanup EXIT INT TERM

now() {
    perl -MTime::HiRes=time -e 'printf "%.6f\n", time'
}

elapsed() {
    awk -v start="$1" -v finish="$2" 'BEGIN { printf "%.3f", finish - start }'
}

start_cpu_sampler() {
    local root_pid="$1"
    local stop_file="$2"
    local result_file="$3"

    (
        while [[ ! -e "$stop_file" ]]; do
            ps -axo pid=,ppid=,time=
            printf '%s\n' '--sample--'
            sleep 0.1
        done
        ps -axo pid=,ppid=,time=
    ) | awk -v root="$root_pid" '
        function seconds(value, parts, count, days) {
            days = 0
            if (index(value, "-") > 0) {
                split(value, parts, "-")
                days = parts[1]
                value = parts[2]
            }
            count = split(value, parts, ":")
            if (count == 3) return days * 86400 + parts[1] * 3600 + parts[2] * 60 + parts[3]
            if (count == 2) return days * 86400 + parts[1] * 60 + parts[2]
            return days * 86400 + value
        }
        $1 == "--sample--" { sample++; next }
        $1 == root || $2 == root {
            current = seconds($3)
            if (!($1 in seen)) {
                baseline[$1] = sample == 0 ? current : 0
                seen[$1] = 1
            }
            if (current > maximum[$1]) maximum[$1] = current
        }
        END {
            total = 0
            for (pid in maximum) total += maximum[pid] - baseline[pid]
            printf "%.3f\n", total
        }
    ' > "$result_file" &
    SAMPLER_PID="$!"
    SAMPLER_PIDS+=("$SAMPLER_PID")
}

wait_for_slot() {
    local slot_name="$1"
    local target_lsn="$2"
    local deadline=$((SECONDS + 300))
    local caught_up="f"

    while [[ "$caught_up" != "t" ]]; do
        if (( SECONDS > deadline )); then
            printf 'Timed out waiting for slot %s to reach %s\n' "$slot_name" "$target_lsn" >&2
            exit 1
        fi
        caught_up="$("$PG_BIN/psql" "$PUB_CONN" -XAtq -c \
            "SELECT coalesce(confirmed_flush_lsn >= '$target_lsn'::pg_lsn, false) FROM pg_replication_slots WHERE slot_name = '$slot_name';")"
        sleep 0.05
    done
}

wait_for_subscription() {
    local deadline=$((SECONDS + 60))
    local active="f"

    while [[ "$active" != "t" ]]; do
        if (( SECONDS > deadline )); then
            printf 'Timed out waiting for the logical replication subscription to start.\n' >&2
            exit 1
        fi
        active="$("$PG_BIN/psql" "$PUB_CONN" -XAtq -c \
            "SELECT coalesce(bool_or(active), false) FROM pg_replication_slots WHERE slot_name = 'bench_sub';")"
        sleep 0.05
    done
}

wait_for_slot_inactive() {
    local slot_name="$1"
    local deadline=$((SECONDS + 60))
    local active="t"

    while [[ "$active" != "f" ]]; do
        if (( SECONDS > deadline )); then
            printf 'Timed out waiting for slot %s to become inactive.\n' "$slot_name" >&2
            exit 1
        fi
        active="$("$PG_BIN/psql" "$PUB_CONN" -XAtq -c \
            "SELECT coalesce(active, false) FROM pg_replication_slots WHERE slot_name = '$slot_name';")"
        sleep 0.05
    done
}

wait_for_event_marker() {
    local marker_id="$1"
    local deadline=$((SECONDS + 300))
    local applied="f"

    while [[ "$applied" != "t" ]]; do
        if (( SECONDS > deadline )); then
            printf 'Timed out waiting for replicated event marker %s.\n' "$marker_id" >&2
            exit 1
        fi
        applied="$("$PG_BIN/psql" "$SUB_CONN" -XAtq -c \
            "SELECT EXISTS (SELECT 1 FROM events WHERE id = $marker_id);")"
        sleep 0.05
    done
}

wait_for_apply_marker() {
    local deadline=$((SECONDS + 300))
    local applied="f"

    while [[ "$applied" != "t" ]]; do
        if (( SECONDS > deadline )); then
            printf 'Timed out waiting for the replica-identity apply marker.\n' >&2
            exit 1
        fi
        applied="$("$PG_BIN/psql" "$SUB_CONN" -XAtq -c \
            'SELECT applied FROM apply_marker WHERE id = 1;')"
        sleep 0.05
    done
}

drop_subscription_and_slot() {
    "$PG_BIN/psql" "$SUB_CONN" -Xq -c 'DROP SUBSCRIPTION IF EXISTS bench_sub;' >/dev/null
    "$PG_BIN/psql" "$PUB_CONN" -Xq -c \
        "SELECT pg_drop_replication_slot('decode_slot') FROM pg_replication_slots WHERE slot_name = 'decode_slot';" >/dev/null
}

reset_tables() {
    drop_subscription_and_slot
    "$PG_BIN/psql" "$PUB_CONN" -Xq -v ON_ERROR_STOP=1 <<'SQL' >/dev/null
TRUNCATE events;
ALTER SEQUENCE events_id_seq RESTART WITH 1;
SELECT pg_stat_reset_shared('wal');
SELECT pg_stat_reset_replication_slot(NULL);
SQL
    "$PG_BIN/psql" "$SUB_CONN" -Xq -v ON_ERROR_STOP=1 <<'SQL' >/dev/null
TRUNCATE events;
DROP INDEX IF EXISTS events_account_idx;
DROP INDEX IF EXISTS events_status_idx;
DROP INDEX IF EXISTS events_updated_idx;
DROP INDEX IF EXISTS events_account_status_idx;
SELECT pg_stat_reset();
SQL
    "$PG_BIN/psql" "$PUB_CONN" -Xq -c 'CHECKPOINT;' >/dev/null
    "$PG_BIN/psql" "$SUB_CONN" -Xq -c 'CHECKPOINT;' >/dev/null
}

create_subscription() {
    local binary="$1"
    "$PG_BIN/psql" "$SUB_CONN" -Xq -v ON_ERROR_STOP=1 -c \
        "CREATE SUBSCRIPTION bench_sub CONNECTION '$PUB_CONN' PUBLICATION bench_pub WITH (copy_data = false, binary = $binary, streaming = parallel);" >/dev/null
    wait_for_subscription
}

start_decoder() {
    "$PG_BIN/pg_recvlogical" -d "$PUB_CONN" -S decode_slot --create-slot -P pgoutput
    "$PG_BIN/pg_recvlogical" -d "$PUB_CONN" -S decode_slot --start -f /dev/null -F 0 -s 1 \
        -o proto_version=4 -o publication_names=bench_pub -o binary=false -o streaming=parallel &
    RECEIVER_PID="$!"

    local deadline=$((SECONDS + 60))
    local active="f"
    while [[ "$active" != "t" ]]; do
        if (( SECONDS > deadline )); then
            printf 'Timed out waiting for pg_recvlogical to start.\n' >&2
            exit 1
        fi
        active="$("$PG_BIN/psql" "$PUB_CONN" -XAtq -c \
            "SELECT coalesce(active, false) FROM pg_replication_slots WHERE slot_name = 'decode_slot';")"
        sleep 0.05
    done
}

stop_decoder() {
    if [[ -n "$RECEIVER_PID" ]]; then
        kill -TERM "$RECEIVER_PID" 2>/dev/null || true
        wait "$RECEIVER_PID" 2>/dev/null || true
        RECEIVER_PID=""
    fi
}

run_scenario() {
    local name="$1"
    local slot_name="$2"
    local pub_pid sub_pid start_lsn target_lsn start_time workload_end end_time
    local pub_stop sub_stop pub_cpu_file sub_cpu_file pub_sampler sub_sampler
    local pgbench_output tps transactions marker_id wal_bytes subscriber_wal_bytes
    local pub_cpu sub_cpu workload_seconds drain_seconds total_seconds

    pub_pid="$(sed -n '1p' "$PUB_DATA/postmaster.pid")"
    sub_pid="$(sed -n '1p' "$SUB_DATA/postmaster.pid")"
    pub_stop="$BENCH_ROOT/${name}.publisher.stop"
    sub_stop="$BENCH_ROOT/${name}.subscriber.stop"
    pub_cpu_file="$BENCH_ROOT/${name}.publisher.cpu"
    sub_cpu_file="$BENCH_ROOT/${name}.subscriber.cpu"
    "$PG_BIN/psql" "$SUB_CONN" -Xq -c "SELECT pg_stat_reset_shared('wal');" >/dev/null
    start_cpu_sampler "$pub_pid" "$pub_stop" "$pub_cpu_file"
    pub_sampler="$SAMPLER_PID"
    start_cpu_sampler "$sub_pid" "$sub_stop" "$sub_cpu_file"
    sub_sampler="$SAMPLER_PID"
    sleep 0.25

    start_lsn="$("$PG_BIN/psql" "$PUB_CONN" -XAtq -c 'SELECT pg_current_wal_lsn();')"
    start_time="$(now)"
    pgbench_output="$("$PG_BIN/pgbench" "$PUB_CONN" -n -c "$CLIENTS" -j "$JOBS" \
        -t "$TRANSACTIONS_PER_CLIENT" -f "$SCRIPT_DIR/logical-replication-insert.sql")"
    workload_end="$(now)"
    marker_id="$("$PG_BIN/psql" "$PUB_CONN" -XAtq -c \
        "INSERT INTO events (account_id, status, payload, updated_at) VALUES (-1, -1, 'apply-marker', clock_timestamp()) RETURNING id;")"
    target_lsn="$("$PG_BIN/psql" "$PUB_CONN" -XAtq -c 'SELECT pg_current_wal_flush_lsn();')"

    if [[ "$slot_name" == "bench_sub" ]]; then
        wait_for_event_marker "$marker_id"
    elif [[ -n "$slot_name" ]]; then
        wait_for_slot "$slot_name" "$target_lsn"
    fi
    end_time="$(now)"

    touch "$pub_stop" "$sub_stop"
    wait "$pub_sampler"
    wait "$sub_sampler"
    SAMPLER_PIDS=()

    tps="$(awk '/^tps =/ { value=$3 } END { print value }' <<< "$pgbench_output")"
    transactions="$(awk '/^number of transactions actually processed:/ { print $6 }' <<< "$pgbench_output")"
    transactions="${transactions%%/*}"
    wal_bytes="$("$PG_BIN/psql" "$PUB_CONN" -XAtq -c \
        "SELECT pg_wal_lsn_diff('$target_lsn'::pg_lsn, '$start_lsn'::pg_lsn)::bigint;")"
    subscriber_wal_bytes="$("$PG_BIN/psql" "$SUB_CONN" -XAtq -c \
        'SELECT wal_bytes::bigint FROM pg_stat_wal;')"
    pub_cpu="$(sed -n '1p' "$pub_cpu_file")"
    sub_cpu="$(sed -n '1p' "$sub_cpu_file")"
    workload_seconds="$(elapsed "$start_time" "$workload_end")"
    drain_seconds="$(elapsed "$workload_end" "$end_time")"
    total_seconds="$(elapsed "$start_time" "$end_time")"

    printf '%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n' \
        "$name" "$transactions" "$((transactions * ROWS_PER_TRANSACTION))" "$tps" \
        "$workload_seconds" "$drain_seconds" "$total_seconds" "$pub_cpu" "$sub_cpu" \
        "$wal_bytes" "$subscriber_wal_bytes"
}

run_full_identity_scenario() {
    local name="$1"
    local add_index="$2"
    local pub_pid sub_pid pub_stop sub_stop pub_cpu_file sub_cpu_file pub_sampler sub_sampler
    local first_updated_id start_time end_time apply_seconds pub_cpu sub_cpu verified_rows

    drop_subscription_and_slot
    "$PG_BIN/psql" "$PUB_CONN" -Xq -v ON_ERROR_STOP=1 -v rows="$FULL_IDENTITY_ROWS" <<'SQL' >/dev/null
TRUNCATE full_events;
UPDATE apply_marker SET applied = false WHERE id = 1;
INSERT INTO full_events (id, status, payload, updated_at)
SELECT g, 0, repeat(md5(g::text), 4), '2026-01-01 00:00:00+00'::timestamptz
FROM generate_series(1, :rows) AS g;
SQL
    "$PG_BIN/psql" "$SUB_CONN" -Xq -v ON_ERROR_STOP=1 -v rows="$FULL_IDENTITY_ROWS" <<'SQL' >/dev/null
TRUNCATE full_events;
DROP INDEX IF EXISTS full_events_lookup_idx;
UPDATE apply_marker SET applied = false WHERE id = 1;
INSERT INTO full_events (id, status, payload, updated_at)
SELECT g, 0, repeat(md5(g::text), 4), '2026-01-01 00:00:00+00'::timestamptz
FROM generate_series(1, :rows) AS g;
SQL
    if [[ "$add_index" == "true" ]]; then
        "$PG_BIN/psql" "$SUB_CONN" -Xq -v ON_ERROR_STOP=1 -c \
            'CREATE INDEX full_events_lookup_idx ON full_events (id);' >/dev/null
    fi
    "$PG_BIN/psql" "$PUB_CONN" -Xq -c 'CHECKPOINT;' >/dev/null
    "$PG_BIN/psql" "$SUB_CONN" -Xq -c 'CHECKPOINT;' >/dev/null

    "$PG_BIN/psql" "$SUB_CONN" -Xq -v ON_ERROR_STOP=1 -c \
        "CREATE SUBSCRIPTION bench_sub CONNECTION '$PUB_CONN' PUBLICATION update_pub WITH (copy_data = false, binary = false, streaming = parallel);" >/dev/null
    wait_for_subscription
    "$PG_BIN/psql" "$SUB_CONN" -Xq -v ON_ERROR_STOP=1 -c \
        'ALTER SUBSCRIPTION bench_sub DISABLE;' >/dev/null
    wait_for_slot_inactive bench_sub

    first_updated_id=$((FULL_IDENTITY_ROWS - FULL_IDENTITY_UPDATES))
    "$PG_BIN/psql" "$PUB_CONN" -Xq -v ON_ERROR_STOP=1 -c \
        "UPDATE full_events SET status = status + 1, updated_at = clock_timestamp() WHERE id > $first_updated_id;" >/dev/null
    "$PG_BIN/psql" "$PUB_CONN" -Xq -v ON_ERROR_STOP=1 -c \
        'UPDATE apply_marker SET applied = true WHERE id = 1;' >/dev/null

    pub_pid="$(sed -n '1p' "$PUB_DATA/postmaster.pid")"
    sub_pid="$(sed -n '1p' "$SUB_DATA/postmaster.pid")"
    pub_stop="$BENCH_ROOT/${name}.publisher.stop"
    sub_stop="$BENCH_ROOT/${name}.subscriber.stop"
    pub_cpu_file="$BENCH_ROOT/${name}.publisher.cpu"
    sub_cpu_file="$BENCH_ROOT/${name}.subscriber.cpu"
    start_cpu_sampler "$pub_pid" "$pub_stop" "$pub_cpu_file"
    pub_sampler="$SAMPLER_PID"
    start_cpu_sampler "$sub_pid" "$sub_stop" "$sub_cpu_file"
    sub_sampler="$SAMPLER_PID"
    sleep 0.25

    start_time="$(now)"
    "$PG_BIN/psql" "$SUB_CONN" -Xq -v ON_ERROR_STOP=1 -c \
        'ALTER SUBSCRIPTION bench_sub ENABLE;' >/dev/null
    wait_for_apply_marker
    end_time="$(now)"

    touch "$pub_stop" "$sub_stop"
    wait "$pub_sampler"
    wait "$sub_sampler"
    SAMPLER_PIDS=()

    apply_seconds="$(elapsed "$start_time" "$end_time")"
    pub_cpu="$(sed -n '1p' "$pub_cpu_file")"
    sub_cpu="$(sed -n '1p' "$sub_cpu_file")"
    verified_rows="$("$PG_BIN/psql" "$SUB_CONN" -XAtq -c \
        "SELECT count(*) FROM full_events WHERE status = 1;")"
    if [[ "$verified_rows" != "$FULL_IDENTITY_UPDATES" ]]; then
        printf 'Expected %s updated rows on subscriber, found %s.\n' \
            "$FULL_IDENTITY_UPDATES" "$verified_rows" >&2
        exit 1
    fi

    printf '%s,%s,%s,%s,%s,%s\n' \
        "$name" "$FULL_IDENTITY_ROWS" "$FULL_IDENTITY_UPDATES" \
        "$apply_seconds" "$pub_cpu" "$sub_cpu"
}

printf 'Temporary benchmark directory: %s\n' "$BENCH_ROOT" >&2
printf 'Initialising two isolated PostgreSQL clusters...\n' >&2
"$PG_BIN/initdb" -D "$PUB_DATA" --username=postgres --no-locale --encoding=UTF8 --auth=trust >/dev/null
"$PG_BIN/initdb" -D "$SUB_DATA" --username=postgres --no-locale --encoding=UTF8 --auth=trust >/dev/null

COMMON_OPTIONS="-c shared_buffers=256MB -c fsync=off -c synchronous_commit=off -c full_page_writes=off -c autovacuum=off -c checkpoint_timeout=30min -c max_wal_size=4GB -c min_wal_size=512MB"
"$PG_BIN/pg_ctl" -D "$PUB_DATA" -l "$PUB_LOG" -w start -o \
    "-p $PUB_PORT -k $BENCH_ROOT -c listen_addresses=127.0.0.1 -c wal_level=logical -c max_replication_slots=10 -c max_wal_senders=10 $COMMON_OPTIONS" >/dev/null
PUB_STARTED=1
"$PG_BIN/pg_ctl" -D "$SUB_DATA" -l "$SUB_LOG" -w start -o \
    "-p $SUB_PORT -k $BENCH_ROOT -c listen_addresses=127.0.0.1 -c max_worker_processes=16 -c max_logical_replication_workers=12 -c max_parallel_apply_workers_per_subscription=8 $COMMON_OPTIONS" >/dev/null
SUB_STARTED=1

"$PG_BIN/psql" "$PUB_ADMIN" -Xq -v ON_ERROR_STOP=1 -c 'CREATE DATABASE bench;' >/dev/null
"$PG_BIN/psql" "$SUB_ADMIN" -Xq -v ON_ERROR_STOP=1 -c 'CREATE DATABASE bench;' >/dev/null

"$PG_BIN/psql" "$PUB_CONN" -Xq -v ON_ERROR_STOP=1 <<'SQL' >/dev/null
CREATE SEQUENCE events_id_seq;
CREATE TABLE events (
    id bigint PRIMARY KEY DEFAULT nextval('events_id_seq'),
    account_id integer NOT NULL,
    status smallint NOT NULL,
    payload text NOT NULL,
    updated_at timestamptz NOT NULL
);
CREATE PUBLICATION bench_pub FOR TABLE events;

CREATE TABLE full_events (
    id bigint NOT NULL,
    status integer NOT NULL,
    payload text NOT NULL,
    updated_at timestamptz NOT NULL
);
CREATE INDEX full_events_pub_id_idx ON full_events (id);
ALTER TABLE full_events REPLICA IDENTITY FULL;
CREATE TABLE apply_marker (
    id integer PRIMARY KEY,
    applied boolean NOT NULL
);
INSERT INTO apply_marker VALUES (1, false);
CREATE PUBLICATION update_pub FOR TABLE full_events, apply_marker WITH (publish = 'update');
SQL

"$PG_BIN/psql" "$SUB_CONN" -Xq -v ON_ERROR_STOP=1 <<'SQL' >/dev/null
CREATE TABLE events (
    id bigint PRIMARY KEY,
    account_id integer NOT NULL,
    status smallint NOT NULL,
    payload text NOT NULL,
    updated_at timestamptz NOT NULL
);
CREATE TABLE full_events (
    id bigint NOT NULL,
    status integer NOT NULL,
    payload text NOT NULL,
    updated_at timestamptz NOT NULL
);
CREATE TABLE apply_marker (
    id integer PRIMARY KEY,
    applied boolean NOT NULL
);
INSERT INTO apply_marker VALUES (1, false);
SQL

printf 'scenario,transactions,rows,tps,workload_seconds,drain_seconds,total_seconds,publisher_cpu_seconds,subscriber_cpu_seconds,publisher_wal_bytes,subscriber_wal_bytes\n'

printf 'Running baseline...\n' >&2
reset_tables
run_scenario baseline ''

printf 'Running decode-only consumer...\n' >&2
reset_tables
start_decoder
run_scenario decode_only decode_slot
stop_decoder

printf 'Running logical replication with text transfer...\n' >&2
reset_tables
create_subscription false
run_scenario logical_text bench_sub

printf 'Running logical replication with binary transfer...\n' >&2
reset_tables
create_subscription true
run_scenario logical_binary bench_sub

printf 'Running logical replication with four extra subscriber indexes...\n' >&2
reset_tables
"$PG_BIN/psql" "$SUB_CONN" -Xq -v ON_ERROR_STOP=1 <<'SQL' >/dev/null
CREATE INDEX events_account_idx ON events (account_id);
CREATE INDEX events_status_idx ON events (status);
CREATE INDEX events_updated_idx ON events (updated_at);
CREATE INDEX events_account_status_idx ON events (account_id, status);
SQL
create_subscription false
run_scenario logical_four_indexes bench_sub

drop_subscription_and_slot

if [[ "$RUN_IDENTITY_BENCHMARK" == "true" ]]; then
    printf '\nreplica_identity_scenario,table_rows,updates,apply_seconds,publisher_cpu_seconds,subscriber_cpu_seconds\n'
    printf 'Running REPLICA IDENTITY FULL without a subscriber lookup index...\n' >&2
    run_full_identity_scenario full_identity_no_index false
    printf 'Running REPLICA IDENTITY FULL with a subscriber lookup index...\n' >&2
    run_full_identity_scenario full_identity_id_index true
fi

drop_subscription_and_slot
printf 'Benchmark complete.\n' >&2