diff --git a/bench/storage_rewrite.md b/bench/storage_rewrite.md new file mode 100644 index 0000000..ca7eba1 --- /dev/null +++ b/bench/storage_rewrite.md @@ -0,0 +1,136 @@ +# Storage rewrite profiling + +This note records the reproducible baseline and the implemented +shadow-generation rewrite for `ALTER TABLE ... ADD PRIMARY KEY`. Run the +workload with: + +```sh +python3 bench/storage_rewrite.py --port 3307 --label ElyraSQL \ + --rows 20000 100000 500000 --indexes 1 --repeats 3 \ + --server-pid $(pgrep -n elyrasql) +``` + +## Baseline + +Release build, default full durability, Apple Silicon host, one secondary +index. Setup is excluded; every sample creates and populates a fresh table. + +| Rows | Median | p95 | Samples | +|---:|---:|---:|---:| +| 20,000 | 202.51 ms | 208.17 ms | 3 | +| 100,000 | 1,157.44 ms | 1,167.75 ms | 3 | +| 500,000 | 6,362.16 ms | 6,668.53 ms | 3 | + +At 20,000 rows, index count exposes the expected mutation amplification: + +| Secondary indexes | Median | p95 | Samples | +|---:|---:|---:|---:| +| 0 | 75.09 ms | 76.30 ms | 5 | +| 1 | 182.45 ms | 209.96 ms | 5 | +| 2 | 352.78 ms | 355.71 ms | 5 | + +The difference between separate 20,000-row runs is normal host/filesystem +variance; the shadow comparison below uses matched fresh-process runs. + +## Profile + +A debug timing pass attributed about 82% of the operation to redb's validated +write transaction. Scan/decode/key construction, transaction staging, point +validation, and serializable range gathering were individually smaller. The +baseline transaction performs an ordered point removal and insertion for every +old/new data and secondary-index entry. + +Two redb 2.6 range-removal prototypes were measured and rejected: + +| 20,000 rows, one index | Median | Versus 182.45 ms baseline | +|---|---:|---:| +| Ordered point deletes (baseline) | 182.45 ms | 1.00x | +| `Table::retain_in` | 1,563.69 ms | 8.57x slower | +| `Table::extract_from_if` | 1,565.52 ms | 8.58x slower | + +Both APIs still walk and remove entries individually inside redb; they are not +subtree-drop primitives. Streaming serializable-range comparison was also +rejected after a 10-sample median of 218.60 ms, because it did not satisfy the +no-regression gate. + +## Shadow-generation implementation + +The engine now builds rows and indexes in a new physical generation using +bounded commits, validates the source table's write sequence and catalog, +atomically switches a small generation pointer, and reclaims the old generation +asynchronously. Cleanup markers are durable and resumed during startup. ALTERs +inside an explicit transaction and multi-operation ALTERs retain the original +atomic path. + +The default build batch is 100,000 rows and can be changed with +`ELYRASQL_REWRITE_BATCH_ROWS` (clamped to 1-100,000). A 20,000-row batch reduced +memory further, but its extra durable commits made the 500,000-row workload +slightly slower than baseline. The 100,000-row default was the best measured +latency/memory balance. + +Matched release builds, full durability, one secondary index, five samples per +cell. The baseline is commit `c7cdebc`, immediately before the shadow-generation +work, and the comparison uses fresh server processes on the same Apple Silicon +host: + +| Rows | Baseline median / p95 | Shadow median / p95 | Change | Baseline RSS growth | Shadow RSS growth | +|---:|---:|---:|---:|---:|---:| +| 20,000 | 203.01 / 206.95 ms | 100.57 / 126.58 ms | 50.5% faster | 6.6 MiB | 5.5 MiB | +| 100,000 | 1,219.81 / 1,545.40 ms | 525.75 / 736.08 ms | 56.9% faster | 44.8 MiB | 24.3 MiB | +| 500,000 | 6,287.03 / 6,809.33 ms | 3,930.00 / 6,086.40 ms | 37.5% faster | 269.7 MiB | 148.6 MiB | + +RSS is sampled from the server every 10 ms during the foreground ALTER and is +reported as growth from immediately before that ALTER. It excludes deferred +old-generation cleanup. Absolute median peaks were 118.2/91.6 MiB, +382.6/280.5 MiB, and 1,374.3/739.5 MiB for baseline/shadow respectively, though +allocator retention makes the per-operation growth the more useful comparison. + +The improvement comes from replacing one full-table validated mutation set +with bounded generation-build commits and a small validated cutover. The cost +is temporary disk space for both generations and extra total I/O while the old +generation is reclaimed. + +## Native MySQL comparison + +The same five-sample workload was also run against native MySQL 8.0.33 on the +same host. This is an engine comparison, not a claim of identical DDL semantics +or durability implementation. + +| Rows | MySQL median / p95 | ElyraSQL baseline / MySQL | ElyraSQL shadow / MySQL | +|---:|---:|---:|---:| +| 20,000 | 40.05 / 47.58 ms | 5.07x | 2.51x | +| 100,000 | 127.17 / 135.26 ms | 9.59x | 4.13x | +| 500,000 | 563.86 / 699.20 ms | 11.15x | 6.97x | + +The shadow rewrite closes roughly half the 20,000- and 100,000-row latency gap, +but MySQL remains substantially faster, especially as the table grows. + +## Crash and sustained-load campaign + +`storage_rewrite_stress.py` drives valid and invalid rewrites concurrently with +indexed reads, account transfers, and an atomic commit oracle. Its aggressive +mode adds randomized `SIGKILL`, `SIGSTOP` followed by `SIGKILL`, and repeated +kills during startup recovery. The five-minute campaign used: + +```sh +python3 bench/storage_rewrite_stress.py \ + --data /tmp/elyra-shadow-chaos.edb \ + --log /tmp/elyra-shadow-chaos.log \ + --duration 300 --crash-min-ms 50 --crash-max-ms 300 \ + --crash-long-probability 0.15 --crash-long-max-ms 5000 \ + --startup-crash-probability 0.5 --max-startup-crashes 2 \ + --stop-before-kill-probability 0.25 --restart-attempts 3 +``` + +It completed 327 forced crashes, including 237 additional recovery-time kills +and 90 stop-then-kill cycles. All 327 successful restarts passed the durable +invariants. Concurrent work included 1,272,209 indexed reads, 28,610 transfers, +7,184 atomic epochs, and 48 completed shadow rewrites. Peak RSS was 253.9 MiB +and the database peaked at 289.6 MiB. + +An earlier run exposed one transient redb 2.6.3 startup panic after repeatedly +killing recovery (`assertion failed: !self.needs_recovery`). The preserved file +opened successfully on the next launch and no persistent corruption was found. +The retry-capable campaign records such startup failures rather than hiding +them. Torn-sector, reordered-write, and individual I/O-failure testing still +requires a lower-level fault-injection interface that redb does not expose. diff --git a/bench/storage_rewrite.py b/bench/storage_rewrite.py new file mode 100644 index 0000000..802bef5 --- /dev/null +++ b/bench/storage_rewrite.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +"""Benchmark storage-heavy ALTER TABLE ADD PRIMARY KEY rewrites. + +Run against ElyraSQL or MySQL through their shared wire protocol. Each sample +uses a fresh table so setup and ALTER timing remain separate. +""" + +import argparse +import math +import subprocess +import statistics +import threading +import time + +import pymysql + + +def percentile(samples, fraction): + ordered = sorted(samples) + return ordered[min(len(ordered) - 1, math.ceil(len(ordered) * fraction) - 1)] + + +def insert_rows(cursor, table, rows, batch_rows): + for start in range(0, rows, batch_rows): + values = ",".join( + f"({row},{row % 1000},'payload-{row % 10000}')" + for row in range(start, min(rows, start + batch_rows)) + ) + cursor.execute(f"INSERT INTO {table} VALUES {values}") + + +def process_rss_mib(pid): + output = subprocess.check_output( + ["ps", "-o", "rss=", "-p", str(pid)], text=True + ).strip() + return int(output) / 1024 + + +def sample(cursor, rows, indexes, batch_rows, repeats, server_pid): + timings = [] + peak_rss = [] + rss_growth = [] + for repetition in range(repeats): + table = f"storage_rewrite_{repetition}" + cursor.execute(f"DROP TABLE IF EXISTS {table}") + index_sql = "" + if indexes >= 1: + index_sql += ", INDEX grp_idx(grp)" + if indexes >= 2: + index_sql += ", INDEX payload_idx(payload)" + cursor.execute( + f"CREATE TABLE {table} (id BIGINT, grp BIGINT, payload VARCHAR(32){index_sql})" + ) + insert_rows(cursor, table, rows, batch_rows) + baseline_rss = process_rss_mib(server_pid) if server_pid else None + stop_sampling = threading.Event() + observed_rss = [baseline_rss] if baseline_rss is not None else [] + + def sample_rss(): + while not stop_sampling.wait(0.01): + try: + observed_rss.append(process_rss_mib(server_pid)) + except (OSError, subprocess.SubprocessError, ValueError): + return + + sampler = None + if server_pid: + sampler = threading.Thread(target=sample_rss, daemon=True) + sampler.start() + started = time.perf_counter_ns() + try: + cursor.execute(f"ALTER TABLE {table} ADD PRIMARY KEY (id)") + timings.append((time.perf_counter_ns() - started) / 1_000_000) + finally: + stop_sampling.set() + if sampler: + sampler.join() + if observed_rss: + peak = max(observed_rss) + peak_rss.append(peak) + rss_growth.append(peak - baseline_rss) + cursor.execute(f"SELECT COUNT(*) FROM {table}") + stored = cursor.fetchone()[0] + if stored != rows: + raise RuntimeError(f"rewrite retained {stored} of {rows} rows") + cursor.execute(f"DROP TABLE {table}") + return timings, peak_rss, rss_growth + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--port", type=int, required=True) + parser.add_argument("--label", required=True) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--user", default="root") + parser.add_argument("--password", default="") + parser.add_argument("--database", default="") + parser.add_argument("--rows", type=int, nargs="+", default=[20_000, 100_000, 500_000]) + parser.add_argument("--indexes", type=int, nargs="+", default=[0, 1, 2]) + parser.add_argument("--batch-rows", type=int, default=1_000) + parser.add_argument("--repeats", type=int, default=5) + parser.add_argument( + "--server-pid", + type=int, + help="sample this server process's foreground peak RSS during ALTER", + ) + args = parser.parse_args() + if min(args.rows) <= 0 or args.batch_rows <= 0 or args.repeats <= 0: + parser.error("row counts, batch size, and repeats must be positive") + if min(args.indexes) < 0 or max(args.indexes) > 2: + parser.error("--indexes values must be between 0 and 2") + + connection = pymysql.connect( + host=args.host, + port=args.port, + user=args.user, + password=args.password, + autocommit=True, + ) + cursor = connection.cursor() + if args.database: + cursor.execute(f"CREATE DATABASE IF NOT EXISTS {args.database}") + cursor.execute(f"USE {args.database}") + + print(f"\n{args.label}: ALTER TABLE ADD PRIMARY KEY") + print( + f"{'rows':>10} {'indexes':>8} {'median ms':>12} {'p95 ms':>12} " + f"{'peak MiB':>10} {'RSS +MiB':>10} {'samples':>8}" + ) + print("-" * 80) + for rows in args.rows: + for indexes in args.indexes: + timings, peak_rss, rss_growth = sample( + cursor, + rows, + indexes, + args.batch_rows, + args.repeats, + args.server_pid, + ) + peak = f"{statistics.median(peak_rss):.1f}" if peak_rss else "-" + growth = f"{statistics.median(rss_growth):.1f}" if rss_growth else "-" + print( + f"{rows:>10,} {indexes:>8} {statistics.median(timings):>12.2f} " + f"{percentile(timings, 0.95):>12.2f} {peak:>10} {growth:>10} " + f"{len(timings):>8}" + ) + + cursor.close() + connection.close() + + +if __name__ == "__main__": + main() diff --git a/bench/storage_rewrite_stress.py b/bench/storage_rewrite_stress.py new file mode 100644 index 0000000..2d26d5b --- /dev/null +++ b/bench/storage_rewrite_stress.py @@ -0,0 +1,617 @@ +#!/usr/bin/env python3 +"""Crash/recovery and sustained-load stress test for shadow table rewrites.""" + +import argparse +import collections +import os +import random +import signal +import statistics +import subprocess +import threading +import time + +import pymysql + + +def is_connection_error(error): + return isinstance(error, pymysql.InterfaceError) or ( + bool(error.args) and error.args[0] in {0, 2002, 2003, 2006, 2013, 2055} + ) + + +class Campaign: + def __init__(self, args): + self.args = args + self.deadline = time.monotonic() + args.duration + self.stop = threading.Event() + self.lock = threading.Lock() + self.server_lock = threading.Lock() + self.server = None + self.server_generation = 0 + self.metrics = collections.Counter() + self.latencies = collections.defaultdict(list) + self.errors = [] + self.samples = [] + self.started_at = None + self.crash_lock = threading.Lock() + + def record(self, name, started): + elapsed_ms = (time.perf_counter() - started) * 1000 + with self.lock: + self.metrics[name] += 1 + self.latencies[name].append(elapsed_ms) + + def expected_error(self, name): + with self.lock: + self.metrics[name] += 1 + + def fail(self, where, error): + with self.lock: + self.errors.append(f"{where}: {error}") + self.stop.set() + + def connect(self): + return pymysql.connect( + host="127.0.0.1", + port=self.args.port, + user="root", + autocommit=True, + connect_timeout=1, + read_timeout=30, + write_timeout=30, + ) + + def spawn_server(self): + with self.server_lock: + log = open(self.args.log, "ab", buffering=0) + self.server = subprocess.Popen( + [ + self.args.server, + "serve", + "--data", + self.args.data, + "--listen", + f"127.0.0.1:{self.args.port}", + ], + stdout=log, + stderr=subprocess.STDOUT, + ) + self.server_generation += 1 + + def start_server(self): + last_error = None + for attempt in range(self.args.restart_attempts): + self.spawn_server() + for _ in range(100): + if self.server.poll() is not None: + last_error = RuntimeError(f"server exited with {self.server.returncode}") + self.server.wait(timeout=10) + self.expected_error("startup_open_failures") + break + try: + connection = self.connect() + connection.close() + return + except pymysql.MySQLError: + time.sleep(0.05) + else: + last_error = RuntimeError("server did not accept connections") + with self.server_lock: + process = self.server + if process.poll() is None: + process.kill() + process.wait(timeout=10) + self.expected_error("startup_open_failures") + if attempt + 1 < self.args.restart_attempts: + time.sleep(0.05) + raise last_error + + def crash_and_restart(self): + with self.crash_lock: + with self.server_lock: + process = self.server + if process and process.poll() is None: + if random.random() < self.args.stop_before_kill_probability: + os.kill(process.pid, signal.SIGSTOP) + time.sleep(random.uniform(0.001, 0.05)) + self.expected_error("stop_then_kill_crashes") + os.kill(process.pid, signal.SIGKILL) + process.wait(timeout=10) + self.expected_error("forced_crashes") + time.sleep(random.uniform(0.001, 0.05)) + + startup_crashes = 0 + while ( + startup_crashes < self.args.max_startup_crashes + and random.random() < self.args.startup_crash_probability + ): + self.spawn_server() + time.sleep(random.uniform(0.001, 0.05)) + with self.server_lock: + startup_process = self.server + if startup_process.poll() is None: + os.kill(startup_process.pid, signal.SIGKILL) + startup_process.wait(timeout=10) + startup_crashes += 1 + self.expected_error("startup_recovery_crashes") + + self.start_server() + self.verify_durable_invariants("post-crash") + self.expected_error("successful_restarts") + + def stop_server(self): + with self.server_lock: + process = self.server + if not process or process.poll() is not None: + return + process.send_signal(signal.SIGINT) + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=10) + + def setup(self): + connection = self.connect() + cursor = connection.cursor() + cursor.execute("DROP TABLE IF EXISTS stress_accounts") + cursor.execute("DROP TABLE IF EXISTS stress_sentinel") + cursor.execute("DROP TABLE IF EXISTS stress_atomic") + cursor.execute("DROP TABLE IF EXISTS stress_atomic_meta") + cursor.execute( + "CREATE TABLE stress_accounts " + "(id BIGINT PRIMARY KEY, balance BIGINT, INDEX balance_idx(balance))" + ) + values = ",".join(f"({row},1000)" for row in range(1000)) + cursor.execute(f"INSERT INTO stress_accounts VALUES {values}") + cursor.execute( + "CREATE TABLE stress_sentinel " + "(id BIGINT PRIMARY KEY, grp BIGINT, checksum BIGINT, payload VARCHAR(64), " + " INDEX grp_idx(grp), UNIQUE checksum_idx(checksum))" + ) + for start in range(0, 20_000, 1000): + values = ",".join( + f"({row},{row % 97},{row * 17 + 3},'sentinel-{row}')" + for row in range(start, start + 1000) + ) + cursor.execute(f"INSERT INTO stress_sentinel VALUES {values}") + cursor.execute( + "CREATE TABLE stress_atomic (epoch BIGINT PRIMARY KEY, checksum BIGINT)" + ) + cursor.execute( + "CREATE TABLE stress_atomic_meta (id BIGINT PRIMARY KEY, committed_epoch BIGINT)" + ) + cursor.execute("INSERT INTO stress_atomic_meta VALUES (1,0)") + cursor.close() + connection.close() + self.verify_durable_invariants("setup") + + def verify_durable_invariants(self, where): + connection = self.connect() + cursor = connection.cursor() + connection.begin() + cursor.execute("SELECT COUNT(*), SUM(balance) FROM stress_accounts") + count, balance = cursor.fetchone() + if (count, balance) != (1000, 1_000_000): + raise AssertionError(f"account invariant {(count, balance)}") + cursor.execute( + "SELECT COUNT(*), SUM(id), SUM(checksum), MIN(id), MAX(id) " + "FROM stress_sentinel" + ) + expected_sum = 19_999 * 20_000 // 2 + expected_checksum = expected_sum * 17 + 3 * 20_000 + actual = cursor.fetchone() + expected = (20_000, expected_sum, expected_checksum, 0, 19_999) + if actual != expected: + raise AssertionError(f"sentinel invariant {actual} != {expected}") + group = random.randrange(97) + cursor.execute("SELECT COUNT(*) FROM stress_sentinel WHERE grp=%s", (group,)) + indexed = cursor.fetchone()[0] + cursor.execute( + "SELECT COUNT(*) FROM stress_sentinel WHERE grp + 0=%s", (group,) + ) + scanned = cursor.fetchone()[0] + if indexed != scanned: + raise AssertionError(f"index mismatch for grp {group}: {indexed} != {scanned}") + cursor.execute("SELECT committed_epoch FROM stress_atomic_meta WHERE id=1") + epoch = cursor.fetchone()[0] + cursor.execute( + "SELECT COUNT(*), COALESCE(SUM(checksum),0), COALESCE(MAX(epoch),0) " + "FROM stress_atomic" + ) + atomic = tuple(map(int, cursor.fetchone())) + expected_atomic = (epoch, epoch * (epoch + 1) // 2 * 31 + epoch * 7, epoch) + if atomic != expected_atomic: + raise AssertionError(f"atomicity oracle {atomic} != {expected_atomic}") + connection.commit() + cursor.close() + connection.close() + self.expected_error(f"invariant_checks_{where}") + + def reconnecting_worker(self, name, operation): + connection = None + while not self.stop.is_set() and time.monotonic() < self.deadline: + try: + if connection is None: + connection = self.connect() + operation(connection) + except (pymysql.OperationalError, pymysql.InterfaceError) as error: + if not is_connection_error(error): + self.fail(name, error) + break + self.expected_error(f"{name}_reconnects") + if connection is not None: + try: + connection.close() + except Exception: + pass + connection = None + time.sleep(0.02) + except Exception as error: + self.fail(name, error) + if connection is not None: + connection.close() + + def transfer(self, connection): + left = random.randrange(1000) + right = (left + random.randrange(1, 1000)) % 1000 + started = time.perf_counter() + cursor = connection.cursor() + try: + connection.begin() + cursor.execute( + "UPDATE stress_accounts SET balance=balance-1 WHERE id=%s", (left,) + ) + cursor.execute( + "UPDATE stress_accounts SET balance=balance+1 WHERE id=%s", (right,) + ) + connection.commit() + self.record("transfers", started) + except pymysql.MySQLError as error: + if is_connection_error(error): + raise + connection.rollback() + if error.args and error.args[0] == 1213: + self.expected_error("transfer_conflicts") + else: + raise + finally: + cursor.close() + + def indexed_read(self, connection): + group = random.randrange(97) + started = time.perf_counter() + cursor = connection.cursor() + cursor.execute( + "SELECT COUNT(*), SUM(checksum) FROM stress_sentinel WHERE grp=%s", (group,) + ) + indexed = cursor.fetchone() + if random.randrange(100) == 0: + cursor.execute( + "SELECT COUNT(*), SUM(checksum) FROM stress_sentinel WHERE grp + 0=%s", + (group,), + ) + if indexed != cursor.fetchone(): + raise AssertionError(f"indexed/scanned result mismatch for {group}") + self.expected_error("online_index_cross_checks") + cursor.close() + self.record("indexed_reads", started) + + def atomic_epoch(self, connection): + started = time.perf_counter() + cursor = connection.cursor() + try: + connection.begin() + cursor.execute("SELECT committed_epoch FROM stress_atomic_meta WHERE id=1") + epoch = cursor.fetchone()[0] + 1 + cursor.execute( + "INSERT INTO stress_atomic VALUES (%s,%s)", (epoch, epoch * 31 + 7) + ) + cursor.execute( + "UPDATE stress_atomic_meta SET committed_epoch=%s WHERE id=1", (epoch,) + ) + connection.commit() + self.record("atomic_epochs", started) + except pymysql.MySQLError as error: + if is_connection_error(error): + raise + connection.rollback() + if error.args and error.args[0] == 1213: + self.expected_error("atomic_epoch_conflicts") + else: + raise + finally: + cursor.close() + + def rewrite_loop(self): + cycle = 0 + while not self.stop.is_set() and time.monotonic() < self.deadline: + table = f"stress_rewrite_{cycle % 4}" + cycle += 1 + connection = None + try: + connection = self.connect() + cursor = connection.cursor() + cursor.execute(f"DROP TABLE IF EXISTS {table}") + cursor.execute( + f"CREATE TABLE {table} " + "(id BIGINT, grp BIGINT, payload VARCHAR(64), INDEX grp_idx(grp))" + ) + rows = random.choice((20_000, 50_000, 100_000)) + expected_sum = rows * (rows - 1) // 2 + for start in range(0, rows, 1000): + values = ",".join( + f"({row},{row % 251},'rewrite-{cycle}-{row}')" + for row in range(start, min(rows, start + 1000)) + ) + cursor.execute(f"INSERT INTO {table} VALUES {values}") + started = time.perf_counter() + cursor.execute(f"ALTER TABLE {table} ADD PRIMARY KEY (id)") + self.record("shadow_rewrites", started) + cursor.execute(f"SELECT COUNT(*), SUM(id), MIN(id), MAX(id) FROM {table}") + actual = cursor.fetchone() + expected = (rows, expected_sum, 0, rows - 1) + if actual != expected: + raise AssertionError(f"{table} aggregate {actual} != {expected}") + probe = random.randrange(251) + cursor.execute(f"SELECT COUNT(*) FROM {table} WHERE grp=%s", (probe,)) + indexed = cursor.fetchone()[0] + cursor.execute(f"SELECT COUNT(*) FROM {table} WHERE grp + 0=%s", (probe,)) + if indexed != cursor.fetchone()[0]: + raise AssertionError(f"{table} secondary index mismatch") + cursor.execute(f"UPDATE {table} SET payload='changed' WHERE id=%s", (rows // 2,)) + cursor.execute(f"DELETE FROM {table} WHERE id=%s", (rows // 3,)) + cursor.execute(f"INSERT INTO {table} VALUES (%s,7,'replacement')", (rows + 1,)) + cursor.execute(f"SELECT COUNT(*) FROM {table}") + if cursor.fetchone()[0] != rows: + raise AssertionError(f"{table} post-rewrite DML count changed") + cursor.execute(f"DROP TABLE {table}") + cursor.close() + except (pymysql.OperationalError, pymysql.InterfaceError) as error: + if is_connection_error(error): + self.expected_error("rewrite_crash_interrupts") + else: + self.fail("rewrite", error) + except Exception as error: + self.fail("rewrite", error) + finally: + if connection is not None: + try: + connection.close() + except Exception: + pass + + def invalid_rewrite_loop(self): + cycle = 0 + while not self.stop.is_set() and time.monotonic() < self.deadline: + table = f"stress_invalid_{cycle % 2}" + cycle += 1 + try: + connection = self.connect() + cursor = connection.cursor() + cursor.execute(f"DROP TABLE IF EXISTS {table}") + cursor.execute(f"CREATE TABLE {table} (id BIGINT, payload VARCHAR(32))") + cursor.execute( + f"INSERT INTO {table} VALUES (1,'a'),(1,'duplicate'),(NULL,'null')" + ) + try: + cursor.execute(f"ALTER TABLE {table} ADD PRIMARY KEY (id)") + raise AssertionError("invalid primary key rewrite succeeded") + except pymysql.MySQLError: + self.expected_error("rejected_invalid_rewrites") + cursor.execute(f"SELECT COUNT(*) FROM {table}") + if cursor.fetchone()[0] != 3: + raise AssertionError("failed rewrite lost source rows") + cursor.execute(f"INSERT INTO {table} VALUES (2,'still-rowid')") + cursor.execute(f"DROP TABLE {table}") + cursor.close() + connection.close() + except (pymysql.OperationalError, pymysql.InterfaceError) as error: + if is_connection_error(error): + self.expected_error("invalid_rewrite_crash_interrupts") + else: + self.fail("invalid-rewrite", error) + except Exception as error: + self.fail("invalid-rewrite", error) + + def monitor(self): + while not self.stop.is_set() and time.monotonic() < self.deadline: + with self.server_lock: + process = self.server + generation = self.server_generation + if process and process.poll() is None: + try: + output = subprocess.check_output( + [ + "ps", + "-o", + "rss=,%cpu=,inblock=,oublock=", + "-p", + str(process.pid), + ], + text=True, + ).split() + rss_mib = int(output[0]) / 1024 + cpu = float(output[1]) + reads = int(output[2]) if output[2] != "-" else -1 + writes = int(output[3]) if output[3] != "-" else -1 + size_mib = os.path.getsize(self.args.data) / (1024 * 1024) + if rss_mib > 0: + with self.lock: + self.samples.append( + ( + time.monotonic(), + generation, + rss_mib, + cpu, + reads, + writes, + size_mib, + ) + ) + except (OSError, subprocess.SubprocessError, ValueError): + pass + time.sleep(1) + + def random_crash_loop(self): + while not self.stop.is_set() and time.monotonic() < self.deadline: + if random.random() < self.args.crash_long_probability: + delay_ms = random.uniform(self.args.crash_max_ms, self.args.crash_long_max_ms) + else: + delay_ms = random.uniform(self.args.crash_min_ms, self.args.crash_max_ms) + delay = delay_ms / 1000 + if self.stop.wait(delay): + return + try: + self.crash_and_restart() + except Exception as error: + self.fail("random-crash-recovery", error) + return + + def run(self): + self.started_at = time.monotonic() + self.start_server() + self.setup() + threads = [ + threading.Thread( + target=self.reconnecting_worker, + args=(f"transfer-{worker}", self.transfer), + daemon=True, + ) + for worker in range(4) + ] + threads += [ + threading.Thread( + target=self.reconnecting_worker, + args=(f"reader-{worker}", self.indexed_read), + daemon=True, + ) + for worker in range(3) + ] + threads += [ + threading.Thread( + target=self.reconnecting_worker, + args=("atomic-epoch", self.atomic_epoch), + daemon=True, + ), + threading.Thread(target=self.rewrite_loop, daemon=True), + threading.Thread(target=self.invalid_rewrite_loop, daemon=True), + threading.Thread(target=self.monitor, daemon=True), + ] + for thread in threads: + thread.start() + + if self.args.crash_max_ms > 0: + crash_thread = threading.Thread(target=self.random_crash_loop, daemon=True) + threads.append(crash_thread) + crash_thread.start() + else: + crash_at = [self.args.duration / 3, self.args.duration * 2 / 3] + started = time.monotonic() + for offset in crash_at: + while not self.stop.is_set() and time.monotonic() - started < offset: + time.sleep(0.1) + if not self.stop.is_set(): + try: + self.crash_and_restart() + except Exception as error: + self.fail("crash-recovery", error) + while not self.stop.is_set() and time.monotonic() < self.deadline: + time.sleep(0.1) + self.stop.set() + for thread in threads: + thread.join(timeout=35) + if not self.errors: + self.verify_durable_invariants("final") + self.stop_server() + self.report() + return 1 if self.errors else 0 + + def report(self): + elapsed = time.monotonic() - self.started_at + print(f"duration_s={elapsed:.1f} requested_s={self.args.duration}") + for name, count in sorted(self.metrics.items()): + print(f"count.{name}={count}") + for name, values in sorted(self.latencies.items()): + if not values: + continue + ordered = sorted(values) + p95 = ordered[min(len(ordered) - 1, int(len(ordered) * 0.95))] + print( + f"latency_ms.{name}.median={statistics.median(values):.3f} " + f"p95={p95:.3f} max={max(values):.3f}" + ) + if self.samples: + rss = [sample[2] for sample in self.samples] + cpu = [sample[3] for sample in self.samples] + sizes = [sample[6] for sample in self.samples] + print( + f"rss_mib.min={min(rss):.1f} median={statistics.median(rss):.1f} " + f"max={max(rss):.1f}" + ) + print(f"cpu_percent.mean={statistics.mean(cpu):.1f} max={max(cpu):.1f}") + print(f"database_mib.min={min(sizes):.1f} max={max(sizes):.1f} final={sizes[-1]:.1f}") + by_generation = collections.defaultdict(list) + for sample in self.samples: + by_generation[sample[1]].append(sample) + io_samples = [sample for sample in self.samples if sample[4] >= 0] + if io_samples: + total_reads = 0 + total_writes = 0 + for generation, samples in by_generation.items(): + available = [sample for sample in samples if sample[4] >= 0] + if available: + total_reads += max(sample[4] for sample in available) - min( + sample[4] for sample in available + ) + total_writes += max(sample[5] for sample in available) - min( + sample[5] for sample in available + ) + print(f"process_block_io.read_ops={total_reads} write_ops={total_writes}") + else: + print("process_block_io=unavailable_on_host") + for error in self.errors: + print(f"ERROR {error}") + print("verdict=PASS" if not self.errors else "verdict=FAIL") + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--server", default="target/release/elyrasql") + parser.add_argument("--data", required=True) + parser.add_argument("--log", required=True) + parser.add_argument("--port", type=int, default=33327) + parser.add_argument("--duration", type=int, default=300) + parser.add_argument("--crash-min-ms", type=int, default=0) + parser.add_argument("--crash-max-ms", type=int, default=0) + parser.add_argument("--crash-long-probability", type=float, default=0.0) + parser.add_argument("--crash-long-max-ms", type=int, default=5000) + parser.add_argument("--startup-crash-probability", type=float, default=0.0) + parser.add_argument("--max-startup-crashes", type=int, default=2) + parser.add_argument("--stop-before-kill-probability", type=float, default=0.0) + parser.add_argument("--restart-attempts", type=int, default=1) + args = parser.parse_args() + if args.crash_min_ms < 0 or args.crash_max_ms < args.crash_min_ms: + parser.error("crash interval must satisfy 0 <= min <= max") + if args.crash_long_max_ms < args.crash_max_ms: + parser.error("--crash-long-max-ms must be at least --crash-max-ms") + if args.restart_attempts <= 0: + parser.error("--restart-attempts must be positive") + for name in ( + "crash_long_probability", + "startup_crash_probability", + "stop_before_kill_probability", + ): + if not 0 <= getattr(args, name) <= 1: + parser.error(f"--{name.replace('_', '-')} must be between 0 and 1") + campaign = Campaign(args) + try: + raise SystemExit(campaign.run()) + finally: + campaign.stop.set() + campaign.stop_server() + + +if __name__ == "__main__": + main() diff --git a/crates/elyra-engine/src/catalog.rs b/crates/elyra-engine/src/catalog.rs index aa06bf8..5ddb1e4 100644 --- a/crates/elyra-engine/src/catalog.rs +++ b/crates/elyra-engine/src/catalog.rs @@ -73,6 +73,10 @@ pub struct TableDef { /// FOREIGN KEY constraints. #[serde(default)] pub foreign_keys: Vec, + /// Physical keyspace generation. It lives in a separate catalog key so the + /// serialized `TableDef` remains backward compatible. + #[serde(skip)] + pub storage_generation: u64, } /// A FOREIGN KEY: `columns` in this table reference `ref_columns` of @@ -99,6 +103,20 @@ pub enum RefAction { } impl TableDef { + pub fn storage_name(&self) -> String { + storage_name(&self.name, self.storage_generation) + } + + pub fn data_prefix(&self) -> Vec { + data_prefix_generation(&self.name, self.storage_generation) + } + + pub fn data_key(&self, encoded: &[u8]) -> Vec { + let mut key = self.data_prefix(); + key.extend_from_slice(encoded); + key + } + pub fn has_pk(&self) -> bool { !self.pk_cols.is_empty() } @@ -172,12 +190,20 @@ pub fn index_table_prefix(table: &str) -> Vec { format!("index::{table}::").into_bytes() } +pub fn index_table_prefix_generation(table: &str, generation: u64) -> Vec { + format!("index::{}::", storage_name(table, generation)).into_bytes() +} + /// Prefix under which all NULL-keyed index entries of a table live (see /// [`IndexDef::indexes_nulls`]). pub fn indexnull_table_prefix(table: &str) -> Vec { format!("indexnull::{table}::").into_bytes() } +pub fn indexnull_table_prefix_generation(table: &str, generation: u64) -> Vec { + format!("indexnull::{}::", storage_name(table, generation)).into_bytes() +} + pub fn catalog_key(table: &str) -> Vec { format!("catalog::{table}").into_bytes() } @@ -517,6 +543,24 @@ pub fn wcount_key(table: &str) -> Vec { format!("meta::wcount::{table}").into_bytes() } +pub fn generation_key(table: &str) -> Vec { + format!("meta::generation::{table}").into_bytes() +} + +pub fn storage_name(table: &str, generation: u64) -> String { + if generation == 0 { + table.to_string() + } else { + // A leading NUL cannot occur in a SQL identifier, making generated + // physical names disjoint from every legacy generation-0 table name. + format!("\0generation_{generation:016x}::{table}") + } +} + +pub fn data_prefix_generation(table: &str, generation: u64) -> Vec { + format!("data::{}::", storage_name(table, generation)).into_bytes() +} + /// Prefix under which all rows of a table live. pub fn data_prefix(table: &str) -> Vec { format!("data::{table}::").into_bytes() @@ -636,10 +680,20 @@ pub async fn load(db: &Session, table: &str) -> Result { } } } - let mut def = match db.get(catalog_key(table)).await? { + let mut values = db + .multi_get(vec![catalog_key(table), generation_key(table)]) + .await? + .into_iter(); + let mut def = match values.next().flatten() { Some(bytes) => TableDef::decode(&bytes)?, None => return Err(Error::Catalog(format!("no such table: {table}"))), }; + def.storage_generation = values + .next() + .flatten() + .and_then(|bytes| bytes.as_slice().try_into().ok()) + .map(u64::from_le_bytes) + .unwrap_or(0); // Declared widths live in their own key, so they are merged in here rather // than in `TableDef::decode`. The result is cached with the definition, so // this costs one extra read per cache miss, not per query. diff --git a/crates/elyra-engine/src/collmig.rs b/crates/elyra-engine/src/collmig.rs index 073b68a..ce2a5c7 100644 --- a/crates/elyra-engine/src/collmig.rs +++ b/crates/elyra-engine/src/collmig.rs @@ -88,8 +88,8 @@ pub async fn migrate(db: &Session) -> Result<()> { // from the previous folding behind. Deleted in batches for the same reason // the rows are: a large table must not be held in memory. for p in [ - catalog::index_table_prefix(table), - catalog::indexnull_table_prefix(table), + catalog::index_table_prefix_generation(table, def.storage_generation), + catalog::indexnull_table_prefix_generation(table, def.storage_generation), ] { let mut cur: Option> = None; loop { @@ -109,7 +109,7 @@ pub async fn migrate(db: &Session) -> Result<()> { } } - let prefix = catalog::data_prefix(table); + let prefix = def.data_prefix(); let mut puts: Vec<(Vec, Vec)> = Vec::new(); let mut dels: Vec> = Vec::new(); let mut cursor: Option> = None; @@ -144,7 +144,7 @@ pub async fn migrate(db: &Session) -> Result<()> { .map(|&c| row.get(c).cloned().unwrap_or(Value::Null)) .collect(); let encoded = crate::keyenc::encode_key_coll(&vals, &def.pk_collations())?; - let newk = catalog::data_key(table, &encoded); + let newk = def.data_key(&encoded); if newk != k { // Either another re-keyed row already claimed these bytes, or a // row that needs no re-keying is already sitting on them (`ae` diff --git a/crates/elyra-engine/src/exec.rs b/crates/elyra-engine/src/exec.rs index bc8c77f..34a10da 100644 --- a/crates/elyra-engine/src/exec.rs +++ b/crates/elyra-engine/src/exec.rs @@ -780,6 +780,7 @@ pub async fn create_table( col_meta, checks, foreign_keys, + storage_generation: 0, }; let widths = catalog::ColumnWidths { bits: ct @@ -2802,6 +2803,7 @@ async fn create_table_as( col_meta: Vec::new(), checks: Vec::new(), foreign_keys: Vec::new(), + storage_generation: 0, }; let mut puts = vec![(catalog_key(name), def.encode()?)]; if let Some(declarations) = declarations.as_ref() { @@ -2846,14 +2848,13 @@ async fn create_table_as( /// TRUNCATE TABLE: remove all rows and index entries, reset counters. pub async fn truncate(db: &Session, name: &str) -> Result { - if !catalog::exists(db, name).await? { - return Err(Error::Catalog(format!("no such table: {name}"))); - } + let def = catalog::load(db, name).await?; + let storage_name = def.storage_name(); let mut deletes = vec![rowid_key(name), autoinc_key(name)]; for prefix in [ - data_prefix(name), - index_table_prefix(name), - indexnull_table_prefix(name), + def.data_prefix(), + index_table_prefix(&storage_name), + indexnull_table_prefix(&storage_name), ] { let mut cursor: Option> = None; loop { @@ -2879,6 +2880,15 @@ pub async fn alter_table( name: &ObjectName, ops: &[AlterTableOperation], ) -> Result { + if !db.in_txn() { + if let [AlterTableOperation::AddConstraint(sqlparser::ast::TableConstraint::PrimaryKey { + columns, + .. + })] = ops + { + return alter_add_primary_key_shadow(db, name, columns).await; + } + } // ALTER helpers persist catalog, row, and index changes independently. Keep // the whole statement behind a private checkpoint so a later operation // cannot expose changes made by an earlier one. @@ -2922,6 +2932,214 @@ pub async fn alter_table( } } +fn rewrite_batch_rows() -> usize { + std::env::var("ELYRASQL_REWRITE_BATCH_ROWS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(100_000) + .clamp(1, 100_000) +} + +fn generation_gc_key(table: &str, generation: u64) -> Vec { + format!("meta::generation-gc::{table}::{generation:016x}").into_bytes() +} + +const GENERATION_GC_PREFIX: &[u8] = b"meta::generation-gc::"; + +fn generation_gc_value(table: &str, generation: u64) -> Result> { + bincode::serialize(&(table, generation)).map_err(|error| Error::Storage(error.to_string())) +} + +async fn cleanup_generation(db: &elyra_storage::Db, table: &str, generation: u64) -> Result<()> { + let prefixes = [ + catalog::data_prefix_generation(table, generation), + catalog::index_table_prefix_generation(table, generation), + catalog::indexnull_table_prefix_generation(table, generation), + ]; + for prefix in prefixes { + loop { + let batch = db.scan_batch(prefix.clone(), None, 4096).await?; + if batch.is_empty() { + break; + } + db.commit(Vec::new(), batch.into_iter().map(|(key, _)| key).collect()) + .await?; + } + } + db.commit(Vec::new(), vec![generation_gc_key(table, generation)]) + .await +} + +pub(crate) async fn resume_generation_cleanup(db: &elyra_storage::Db) -> Result<()> { + let mut cursor = None; + loop { + let markers = db + .scan_batch(GENERATION_GC_PREFIX.to_vec(), cursor.clone(), 256) + .await?; + if markers.is_empty() { + break; + } + let last = markers.len() < 256; + cursor = markers.last().map(|(key, _)| key.clone()); + for (_, value) in markers { + let (table, generation): (String, u64) = + bincode::deserialize(&value).map_err(|error| Error::Storage(error.to_string()))?; + cleanup_generation(db, &table, generation).await?; + } + if last { + break; + } + } + Ok(()) +} + +async fn alter_add_primary_key_shadow( + db: &Session, + name: &ObjectName, + columns: &[Ident], +) -> Result { + let table = stored_table_ident(db, name)?; + db.begin()?; + let result = async { + let mut definition = catalog::load(db, &table).await?; + let old_generation = definition.storage_generation; + let new_generation = old_generation + .checked_add(1) + .ok_or_else(|| Error::Storage("table generation exhausted".into()))?; + let old_definition = definition.clone(); + + if definition.has_pk() { + return Err(Error::Query("multiple primary keys are not allowed".into())); + } + if columns.is_empty() { + return Err(Error::Query( + "ALTER TABLE ADD PRIMARY KEY requires at least one column".into(), + )); + } + let mut primary_columns = Vec::with_capacity(columns.len()); + for column in columns { + let index = definition + .schema + .columns + .iter() + .position(|candidate| predicate::identifier_eq(&candidate.name, &column.value)) + .ok_or_else(|| Error::Catalog(format!("unknown column: {column}")))?; + if primary_columns.contains(&index) { + return Err(Error::Query(format!( + "duplicate column '{}' in primary key", + column.value + ))); + } + primary_columns.push(index); + } + definition.pk_cols = primary_columns; + for &column in &definition.pk_cols { + definition.schema.columns[column].nullable = false; + } + definition.storage_generation = new_generation; + + // The table write sequence and catalog value form a compact validation + // token for the snapshot used to build the shadow generation. + db.lock_keys(&[wcount_key(&table), catalog_key(&table)]); + + // Remove an unreachable generation left by an interrupted earlier + // attempt before reusing its generation number. + cleanup_generation(&db.raw_db(), &table, new_generation).await?; + + let source_prefix = old_definition.data_prefix(); + let target_prefix = definition.data_prefix(); + let primary_collations = definition.pk_collations(); + let batch_rows = rewrite_batch_rows(); + let mut cursor = None; + let mut cancel = db.cancel_check(); + let mut pacer = Pacer::new(); + loop { + let batch = db + .scan_batch(source_prefix.clone(), cursor.clone(), batch_rows) + .await?; + if batch.is_empty() { + break; + } + let last = batch.len() < batch_rows; + cursor = batch.last().map(|(key, _)| key.clone()); + let mut new_entries = Vec::new(); + let mut auxiliary_entries = Vec::new(); + for (_, encoded_row) in batch { + cancel.tick()?; + pacer.tick().await; + let row = rowdec::decode_row(&encoded_row)?; + if definition + .pk_cols + .iter() + .any(|&column| row[column].is_null()) + { + return Err(Error::Query( + "primary key columns cannot contain NULL".into(), + )); + } + let clustered = + keyenc::encode_columns_coll(&row, &definition.pk_cols, &primary_collations)?; + let mut data_key = target_prefix.clone(); + data_key.extend_from_slice(&clustered); + let (non_unique, unique) = + index::partition_entries_for_row(&definition, &row, &data_key)?; + new_entries.push((data_key, encoded_row)); + new_entries.extend(unique); + auxiliary_entries.extend(non_unique); + } + db.raw_db() + .commit_insert(new_entries, auxiliary_entries, Vec::new()) + .await?; + if last { + break; + } + } + + let marker = generation_gc_key(&table, old_generation); + db.commit_write( + vec![ + (catalog_key(&table), definition.encode()?), + ( + catalog::generation_key(&table), + new_generation.to_le_bytes().to_vec(), + ), + (marker, generation_gc_value(&table, old_generation)?), + bump_wcount(db, &table).await?, + ], + vec![rowid_key(&table)], + ) + .await?; + db.commit().await?; + Ok((old_generation, new_generation)) + } + .await; + + match result { + Ok((old_generation, _new_generation)) => { + let cleanup_db = db.raw_db().clone(); + let cleanup_table = table.clone(); + tokio::spawn(async move { + if let Err(error) = + cleanup_generation(&cleanup_db, &cleanup_table, old_generation).await + { + tracing::warn!(%error, table = %cleanup_table, "generation cleanup failed"); + } + }); + Ok(QueryResult::Affected(0)) + } + Err(error) => { + db.rollback(); + // A failed build never made the target generation reachable. + if let Ok(definition) = catalog::load(db, &table).await { + if let Some(new_generation) = definition.storage_generation.checked_add(1) { + let _ = cleanup_generation(&db.raw_db(), &table, new_generation).await; + } + } + Err(error) + } + } +} + async fn alter_table_inner( db: &Session, name: &ObjectName, @@ -3175,13 +3393,13 @@ async fn alter_add_primary_key(db: &Session, def: &mut TableDef, columns: &[Iden let mut deletes = Vec::new(); let mut clustered_keys = std::collections::HashSet::new(); let pk_collations = def.pk_collations(); - let clustered_prefix = data_prefix(&def.name); + let clustered_prefix = def.data_prefix(); let rewrite_budget = db.transaction_write_budget_remaining(); let mut rewrite_bytes = puts .iter() .map(|(key, value)| key.len() + value.len()) .sum(); - let prefix = data_prefix(&old_def.name); + let prefix = old_def.data_prefix(); let mut cursor = None; loop { let batch = db.scan_batch(prefix.clone(), cursor.clone(), 4096).await?; @@ -3708,7 +3926,7 @@ async fn alter_add_column( .map(|&position| row[position].clone()) .collect::>(); data_key( - &def.name, + &def.storage_name(), &keyenc::encode_key_coll(&primary_values, &def.pk_collations())?, ) } else { @@ -3806,7 +4024,7 @@ async fn alter_drop_column(db: &Session, def: &mut TableDef, name: &str) -> Resu } // Rewrite rows without the dropped position. - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let mut cursor: Option> = None; let mut puts = Vec::new(); loop { @@ -3863,7 +4081,17 @@ async fn alter_rename_table(db: &Session, def: &mut TableDef, new: &str) -> Resu return Err(Error::Catalog(format!("table already exists: {new}"))); } let old = def.name.clone(); + let old_generation = def.storage_generation; + let old_prefix = catalog::data_prefix_generation(&old, old_generation); + let target_generation = db + .get(catalog::generation_key(new)) + .await? + .and_then(|bytes| bytes.as_slice().try_into().ok()) + .map(u64::from_le_bytes) + .unwrap_or(0) + .max(old_generation); def.name = new.to_string(); + def.storage_generation = target_generation; for foreign_key in &mut def.foreign_keys { if foreign_key.ref_table.eq_ignore_ascii_case(&old) { foreign_key.ref_table = new.to_string(); @@ -3874,7 +4102,6 @@ async fn alter_rename_table(db: &Session, def: &mut TableDef, new: &str) -> Resu let mut deletes: Vec> = Vec::new(); // Re-key all data rows and rebuild their index entries under the new name. - let old_prefix = data_prefix(&old); let mut cursor: Option> = None; loop { let chunk = db @@ -3887,7 +4114,8 @@ async fn alter_rename_table(db: &Session, def: &mut TableDef, new: &str) -> Resu cursor = chunk.last().map(|(k, _)| k.clone()); for (old_key, v) in chunk { let clustered = &old_key[old_prefix.len()..]; - let new_key = data_key(new, clustered); + let mut new_key = def.data_prefix(); + new_key.extend_from_slice(clustered); let row: Vec = rowdec::decode_row(&v)?; deletes.push(old_key); puts.push((new_key.clone(), v)); @@ -3902,8 +4130,8 @@ async fn alter_rename_table(db: &Session, def: &mut TableDef, new: &str) -> Resu // the NULL-keyed entries under `indexnull::` (rebuilt under the new name by // `entries_for_row` above). for old_index_prefix in [ - format!("index::{old}::").into_bytes(), - format!("indexnull::{old}::").into_bytes(), + catalog::index_table_prefix_generation(&old, old_generation), + catalog::indexnull_table_prefix_generation(&old, old_generation), ] { let mut cursor: Option> = None; loop { @@ -3926,7 +4154,22 @@ async fn alter_rename_table(db: &Session, def: &mut TableDef, new: &str) -> Resu // Move catalog + table-scoped metadata. deletes.push(catalog_key(&old)); + // Retain the old name's watermark. A deferred cleanup may still be + // deleting an earlier generation, so a later CREATE with this name must + // not reuse that physical keyspace. + if old_generation != 0 { + puts.push(( + catalog::generation_key(&old), + old_generation.to_le_bytes().to_vec(), + )); + } puts.push((catalog_key(new), def.encode()?)); + if target_generation != 0 { + puts.push(( + catalog::generation_key(new), + target_generation.to_le_bytes().to_vec(), + )); + } // MySQL carries referencing foreign keys across a table rename. Update // every child catalog in the same write so later DML never probes the old // table name. @@ -4012,7 +4255,7 @@ pub async fn create_fulltext_index( // Persist the catalog and backfill index entries for existing rows. let mut puts: Vec<(Vec, Vec)> = vec![(catalog_key(table), def.encode()?)]; - let prefix = data_prefix(table); + let prefix = def.data_prefix(); let mut cursor: Option> = None; loop { let chunk = db.scan_batch(prefix.clone(), cursor.clone(), 4096).await?; @@ -4101,7 +4344,7 @@ pub async fn create_index(db: &Session, ci: CreateIndex) -> Result // Persist the new catalog and backfill index entries for existing rows. let mut puts: Vec<(Vec, Vec)> = vec![(catalog_key(&table), def.encode()?)]; - let prefix = data_prefix(&table); + let prefix = def.data_prefix(); let mut cursor: Option> = None; loop { let chunk = db.scan_batch(prefix.clone(), cursor.clone(), 4096).await?; @@ -4182,11 +4425,12 @@ async fn collect_index_entry_keys( table: &str, index_names: &[&str], ) -> Result>> { + let storage_table = catalog::load(db, table).await?.storage_name(); let mut keys = Vec::new(); for index_name in index_names { for prefix in [ - index::index_scan_prefix(table, index_name), - index::indexnull_scan_prefix(table, index_name), + index::index_scan_prefix(&storage_table, index_name), + index::indexnull_scan_prefix(&storage_table, index_name), ] { let mut cursor = None; loop { @@ -4320,7 +4564,7 @@ pub async fn insert(db: &Session, vindex: &VectorRegistry, ins: Insert) -> Resul let on_dup = !dup_sets.is_empty(); let has_pk = def.has_pk(); let pk_colls = def.pk_collations(); - let clustered_prefix = data_prefix(&name); + let clustered_prefix = def.data_prefix(); // Load rowid counter once for tables without a PK. let mut next_rowid = if has_pk { @@ -5166,13 +5410,18 @@ fn fk_probe_key(parent: &TableDef, ref_cols: &[String], vals: &[Value]) -> Resul }; if !parent.pk_cols.is_empty() && name_match(&parent.pk_cols) { return Ok(data_key( - &parent.name, + &parent.storage_name(), &keyenc::encode_key_coll(vals, &parent.pk_collations())?, )); } for idx in &parent.indexes { if idx.unique && !idx.vector && name_match(&idx.cols) { - return index::unique_probe_key(&parent.name, &idx.name, vals, &idx.col_collations); + return index::unique_probe_key( + &parent.storage_name(), + &idx.name, + vals, + &idx.col_collations, + ); } } Err(Error::Query(format!( @@ -6102,7 +6351,7 @@ async fn select_inner( if order_is_pk_asc_prefix(&def, &resolved) && !selective_filter(&def, filter.as_ref())? { let need = offset.saturating_add(lim); - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let mut rows: Vec> = Vec::with_capacity(need.min(4096)); if !db.in_txn() { // Autocommit: iterate clustered order in one read transaction, @@ -6174,7 +6423,7 @@ async fn select_inner( && !selective_filter(&def, filter.as_ref())? { let need = offset.saturating_add(lim); - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let sch = def.schema.clone(); let f = filter.clone(); // With no residual filter, skip the first `offset` rows without @@ -6232,7 +6481,7 @@ async fn select_inner( if !db.in_txn() && !selective_filter(&def, filter.as_ref())? { if let Some(plan) = secondary_order_plan(&def, &resolved) { let need = offset.saturating_add(lim); - let iprefix = index::index_scan_prefix(&def.name, &plan.index); + let iprefix = index::index_scan_prefix(&def.storage_name(), &plan.index); let has_filter = filter.is_some(); let walk_budget = if has_filter { ordered_scan_budget(need) @@ -6271,7 +6520,7 @@ async fn select_inner( // the NULL prefix comes first (NULLs sort first); for DESC the // value prefix comes first (NULLs last). Both give the exact // MySQL ordering including a PK tiebreaker. - let nprefix = index::indexnull_scan_prefix(&def.name, &plan.index); + let nprefix = index::indexnull_scan_prefix(&def.storage_name(), &plan.index); let run_two = |skip: usize, want: usize| { let sch = def.schema.clone(); let f = filter.clone(); @@ -6421,7 +6670,7 @@ async fn select_inner( // which merges the MVCC snapshot with the transaction's own overlay, // so this is correct in autocommit AND inside a transaction (the old // code fell back to a full in-memory sort while in a transaction). - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let mut cursor: Option> = None; let asc: Vec = resolved.iter().map(|(_, a)| *a).collect(); let colls: Vec = resolved @@ -6771,7 +7020,7 @@ async fn streaming_nlj_select( } }; - let prefix = data_prefix(&ddef.name); + let prefix = ddef.data_prefix(); let mut cursor: Option> = None; let mut out: Vec> = Vec::new(); 'outer: loop { @@ -7269,7 +7518,7 @@ async fn build_join_chain( .collect() }) .filter(|cols| cols.len() * SELECTIVE_COPY_WEIGHT < plen); - let prefix = data_prefix(&p.def.name); + let prefix = p.def.data_prefix(); let mut cursor: Option> = None; let mut decoded: Vec = Vec::with_capacity(plen); @@ -7707,7 +7956,7 @@ async fn streaming_join_aggregate( // the spilling aggregator -- so a large join + GROUP BY is bounded by the // group state (which spills), not the join output size. let mut sa = SpillAgg::new(&plan); - let prefix = data_prefix(&ddef.name); + let prefix = ddef.data_prefix(); let mut cursor: Option> = None; let selective = chain_is_selective(&steps); let mut bufs: Vec = (0..steps.len()).map(|_| ChainBuf::default()).collect(); @@ -7877,7 +8126,7 @@ async fn streaming_join_order( limit, crate::sort::sort_max_rows(), ); - let prefix = data_prefix(&ddef.name); + let prefix = ddef.data_prefix(); let mut cursor: Option> = None; let mut keybuf: Vec = Vec::with_capacity(resolved.len()); let selective = chain_is_selective(&steps); @@ -8964,7 +9213,7 @@ async fn lookup_rows_by_eq( }; if def.pk_cols == [col] { let key = data_key( - &def.name, + &def.storage_name(), &keyenc::encode_coll(value, def.collation_of(col))?, ); return Ok(match db.get(key).await? { @@ -8973,7 +9222,8 @@ async fn lookup_rows_by_eq( }); } if let Some(idx) = index::index_on(def, col) { - let dks = index::lookup_eq(db, &def.name, idx, std::slice::from_ref(value)).await?; + let dks = + index::lookup_eq(db, &def.storage_name(), idx, std::slice::from_ref(value)).await?; let blobs = db.multi_get(dks).await?; let mut out = Vec::new(); for b in blobs.into_iter().flatten() { @@ -8999,7 +9249,7 @@ async fn lookup_keys_by_eq( ) -> Result>> { if def.pk_cols == [col] { let key = data_key( - &def.name, + &def.storage_name(), &keyenc::encode_coll(value, def.collation_of(col))?, ); // The primary key identifies at most one row, but it still has to exist. @@ -9012,7 +9262,7 @@ async fn lookup_keys_by_eq( let Some(idx) = index::index_on(def, col) else { return Ok(Vec::new()); }; - index::lookup_eq(db, &def.name, idx, std::slice::from_ref(value)).await + index::lookup_eq(db, &def.storage_name(), idx, std::slice::from_ref(value)).await } /// If `on` is `A = B` with one operand referencing only the driving side and @@ -9421,7 +9671,7 @@ async fn table_rows(db: &Session, def: &TableDef) -> Result { return Ok(rows); } } - let rows = db.raw_db().count_prefix(data_prefix(&def.name)).await?; + let rows = db.raw_db().count_prefix(def.data_prefix()).await?; cache .lock() .unwrap_or_else(|e| e.into_inner()) @@ -9449,11 +9699,11 @@ async fn clustered_range( def: &TableDef, rq: &RangeQuery, ) -> Result, Vec)>> { - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let coll = def.pk_collations().first().copied().unwrap_or_default(); let mut start = match &rq.lo { Some((v, incl)) => { - let mut b = data_key(&def.name, &keyenc::encode_coll(v, coll)?); + let mut b = data_key(&def.storage_name(), &keyenc::encode_coll(v, coll)?); if !*incl { b.push(0x00); // strictly after the row with pk == v } @@ -9463,7 +9713,7 @@ async fn clustered_range( }; let end = match &rq.hi { Some((v, incl)) => { - let mut b = data_key(&def.name, &keyenc::encode_coll(v, coll)?); + let mut b = data_key(&def.storage_name(), &keyenc::encode_coll(v, coll)?); if *incl { b.push(0x00); // include the row with pk == v } @@ -9516,7 +9766,7 @@ async fn index_range( ) -> Result, Vec)>>> { let lo = rq.lo.as_ref().map(|(v, i)| (v, *i)); let hi = rq.hi.as_ref().map(|(v, i)| (v, *i)); - let data_keys = index::lookup_range(db, &def.name, idx, lo, hi).await?; + let data_keys = index::lookup_range(db, &def.storage_name(), idx, lo, hi).await?; if let Some(b) = budget { if data_keys.len() > b { return Ok(None); @@ -9550,7 +9800,8 @@ async fn composite_index_range( .as_ref() .map(|(value, inclusive)| (value, *inclusive)); let data_keys = - index::lookup_prefix_range(db, &def.name, query.index, &query.prefix, lo, hi).await?; + index::lookup_prefix_range(db, &def.storage_name(), query.index, &query.prefix, lo, hi) + .await?; if budget.is_some_and(|budget| data_keys.len() > budget) { return Ok(None); } @@ -12058,7 +12309,7 @@ async fn collect_null_rows( want: usize, budget: usize, ) -> Result<(Vec>, bool)> { - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let sch = def.schema.clone(); let f = filter.clone(); let (rows, _examined, budget_hit) = db @@ -13039,7 +13290,9 @@ async fn collect_matches_inner( continue; } let stem = crate::ft::stem(&cleaned); - for dk in index::fulltext_lookup(db, &def.name, &idx.name, &stem).await? { + for dk in index::fulltext_lookup(db, &def.storage_name(), &idx.name, &stem) + .await? + { if seen.insert(dk.clone()) { cand.push(dk); } @@ -13069,7 +13322,7 @@ async fn collect_matches_inner( if def.has_pk() { if let Some(vals) = key_eq_values(def, filter, &def.pk_cols)? { let key = data_key( - &def.name, + &def.storage_name(), &keyenc::encode_key_coll(&vals, &def.pk_collations())?, ); if let Some(bytes) = db.get(key.clone()).await? { @@ -13088,7 +13341,7 @@ async fn collect_matches_inner( continue; } if let Some(vals) = key_eq_values(def, filter, &idx.cols)? { - let data_keys = index::lookup_eq(db, &def.name, idx, &vals).await?; + let data_keys = index::lookup_eq(db, &def.storage_name(), idx, &vals).await?; let blobs = db.multi_get(data_keys.clone()).await?; for (data_key, blob) in data_keys.into_iter().zip(blobs) { if let Some(bytes) = blob { @@ -13212,7 +13465,7 @@ async fn collect_matches_inner( } } - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let mut cursor: Option> = None; loop { let chunk = db.scan_batch(prefix.clone(), cursor.clone(), 4096).await?; @@ -13471,7 +13724,7 @@ pub async fn update( let new_key = if def.has_pk() { let pk_vals: Vec = def.pk_cols.iter().map(|&i| new_row[i].clone()).collect(); data_key( - &name, + &def.storage_name(), &keyenc::encode_key_coll(&pk_vals, &def.pk_collations())?, ) } else { @@ -13909,7 +14162,7 @@ async fn lookup_child_rows( .iter() .find(|ix| !ix.vector && ix.cols == cols) { - let data_keys = index::lookup_eq(db, &child.name, idx, vals).await?; + let data_keys = index::lookup_eq(db, &child.storage_name(), idx, vals).await?; let blobs = db.multi_get(data_keys.clone()).await?; let mut out = Vec::new(); for (k, b) in data_keys.into_iter().zip(blobs) { @@ -14136,7 +14389,7 @@ async fn multi_update( let base = extract_base_row(&joined, &info.col_idx); let pk_vals: Vec = info.def.pk_cols.iter().map(|&i| base[i].clone()).collect(); let pk_key = data_key( - &info.name, + &info.def.storage_name(), &keyenc::encode_key_coll(&pk_vals, &info.def.pk_collations())?, ); let entry = updated.entry(qual.clone()).or_default(); @@ -14181,7 +14434,7 @@ async fn multi_update( .map(|&i| new_base[i].clone()) .collect(); let new_key = data_key( - &info.name, + &info.def.storage_name(), &keyenc::encode_key_coll(&new_pk, &info.def.pk_collations())?, ); deletes.extend(index::entry_keys_for_row(&info.def, old_base, pk_key)?); @@ -14266,7 +14519,7 @@ async fn multi_delete( let base = extract_base_row(&joined, &info.col_idx); let pk_vals: Vec = info.def.pk_cols.iter().map(|&i| base[i].clone()).collect(); let pk_key = data_key( - &info.name, + &info.def.storage_name(), &keyenc::encode_key_coll(&pk_vals, &info.def.pk_collations())?, ); per_table.entry(q.clone()).or_default().insert(pk_key, base); @@ -18834,6 +19087,7 @@ async fn create_temp_table(db: &Session, base: &str, schema: &Schema) -> Result< col_meta: Vec::new(), checks: Vec::new(), foreign_keys: Vec::new(), + storage_generation: 0, } .encode()?; let owner_number = TEMP_OWNER_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed); @@ -20462,13 +20716,13 @@ async fn hybrid_select( .find(|i| i.fulltext && i.single_col() == Some(text_ci)); if let Some(idx) = ft_idx { for term in &terms { - for dk in index::fulltext_lookup(db, &def.name, &idx.name, term).await? { + for dk in index::fulltext_lookup(db, &def.storage_name(), &idx.name, term).await? { *ft_score.entry(dk).or_default() += 1; } } } else { // No full-text index: scan and score by distinct query-term presence. - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let mut cursor: Option> = None; loop { let batch = db.scan_batch(prefix.clone(), cursor.clone(), 4096).await?; @@ -20667,7 +20921,7 @@ async fn olap_aggregate( // keyspace without decoding any rows, in parallel over clustered ranges, // and seed the result directly instead of feeding N rows. if filter.is_none() && plan.is_count_star_only() && !db.in_txn() { - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let raw = db.raw_db(); let n = match pk_split_ranges(&raw, def, &prefix, agg_workers()).await? { Some(ranges) => { @@ -20921,7 +21175,7 @@ async fn scan_columnar_scalar( specs: &[(elyra_olap::AggFunc, Option, bool)], ) -> Result> { let ncols = def.schema.columns.len(); - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let raw = db.raw_db(); let workers = agg_workers(); if workers > 1 { @@ -21265,7 +21519,7 @@ async fn scan_columnar_group( explicit_ranges: Option, Vec)>>, ) -> Result, Vec)>>> { let ncols = def.schema.columns.len(); - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let raw = db.raw_db(); let workers = agg_workers(); // Work units: explicit (zone-map surviving) ranges if given, otherwise the @@ -21333,7 +21587,7 @@ async fn get_or_build_zonemap( return Ok(Some(zm)); } let raw = db.raw_db(); - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let upper = prefix_successor(&prefix); let mut check = db.cancel_check(); let b = raw @@ -21406,7 +21660,7 @@ async fn build_cached_table( e0: u64, ) -> Result> { let budget = colcache::budget_bytes(); - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let raw = db.raw_db(); struct Acc { blobs: Vec>, @@ -21525,7 +21779,7 @@ async fn scan_aggregate_fast( filter: Option, plan: &AggPlan, ) -> Result { - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let schema = def.schema.clone(); let needed = agg_needed_mask(&schema, filter.as_ref(), plan); let ncols = schema.columns.len(); @@ -21870,7 +22124,7 @@ async fn partitioned_aggregate( let extend = !plan.arg_exprs().is_empty(); let mut sa = SpillAgg::new(plan); let mut fedbuf: Vec = Vec::new(); - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); // In autocommit, pin one snapshot so this multi-batch scan reads a single // consistent view (concurrent commits are all-or-nothing across the whole // aggregate). In a transaction the session snapshot+overlay is already @@ -21974,7 +22228,7 @@ async fn index_count_eq(db: &Session, def: &TableDef, filter: &Expr) -> Result Result Result { let mut seen: Vec = vec![0; ncols]; let mut rng: u64 = 0x9E37_79B9_7F4A_7C15; - let prefix = data_prefix(name); + let prefix = def.data_prefix(); let mut cursor: Option> = None; let mut rows = 0u64; loop { @@ -22686,7 +22940,12 @@ pub async fn drop_table(db: &Session, name: &str, if_exists: bool) -> Result Result>> { + let definition = catalog::load(db, name).await?; + let storage_name = definition.storage_name(); // Collect the table's data and index keys in batches. + // The generation watermark deliberately survives DROP. Deferred cleanup + // for this logical name may still be running, and a recreated table must + // use a keyspace newer than anything cleanup can target. let mut deletes = vec![ catalog_key(name), rowid_key(name), @@ -22696,9 +22955,9 @@ async fn table_delete_keys(db: &Session, name: &str) -> Result>> { catalog::coldecl_key(name), ]; for prefix in [ - data_prefix(name), - index_table_prefix(name), - indexnull_table_prefix(name), + definition.data_prefix(), + index_table_prefix(&storage_name), + indexnull_table_prefix(&storage_name), ] { let mut cursor: Option> = None; loop { @@ -22980,6 +23239,47 @@ fn parse_vector(s: &str, dim: u32) -> Result> { Ok(vals) } +#[cfg(test)] +mod generation_cleanup_tests { + use super::{generation_gc_key, generation_gc_value, resume_generation_cleanup}; + use crate::catalog; + use elyra_storage::Db; + + #[tokio::test] + async fn startup_cleanup_reclaims_every_stale_generation_keyspace() { + let db = Db::in_memory().unwrap(); + let table = "table::with::separators"; + let generation = 7; + let stale_keys = [ + catalog::data_prefix_generation(table, generation), + catalog::index_table_prefix_generation(table, generation), + catalog::indexnull_table_prefix_generation(table, generation), + ] + .map(|mut prefix| { + prefix.extend_from_slice(b"entry"); + prefix + }); + let marker = generation_gc_key(table, generation); + let mut puts = stale_keys + .iter() + .cloned() + .map(|key| (key, b"value".to_vec())) + .collect::>(); + puts.push(( + marker.clone(), + generation_gc_value(table, generation).unwrap(), + )); + db.commit(puts, Vec::new()).await.unwrap(); + + resume_generation_cleanup(&db).await.unwrap(); + + for key in stale_keys { + assert_eq!(db.get(key).await.unwrap(), None); + } + assert_eq!(db.get(marker).await.unwrap(), None); + } +} + #[cfg(test)] mod cte_rewrite_tests { use crate::{Engine, QueryResult, Session}; @@ -24213,6 +24513,7 @@ mod plan_tests { col_meta: Vec::new(), checks: Vec::new(), foreign_keys: Vec::new(), + storage_generation: 0, } } diff --git a/crates/elyra-engine/src/index.rs b/crates/elyra-engine/src/index.rs index 64b750f..8c4cff0 100644 --- a/crates/elyra-engine/src/index.rs +++ b/crates/elyra-engine/src/index.rs @@ -12,7 +12,7 @@ use crate::session::Session; use elyra_core::{Collation, Result, Value}; -use crate::catalog::{data_prefix, IndexDef, TableDef}; +use crate::catalog::{IndexDef, TableDef}; use crate::ft; use crate::keyenc; @@ -31,11 +31,12 @@ fn fulltext_entry_keys( text.push_str(&s); } } - let clustered = &data_key[data_prefix(&def.name).len()..]; + let storage_table = def.storage_name(); + let clustered = &data_key[def.data_prefix().len()..]; ft::unique_terms(&text) .into_iter() .map(|term| { - let mut k = format!("index::{}::{}::", def.name, idx.name).into_bytes(); + let mut k = format!("index::{storage_table}::{}::", idx.name).into_bytes(); k.extend_from_slice(term.as_bytes()); k.push(0); k.extend_from_slice(clustered); @@ -51,7 +52,7 @@ pub async fn fulltext_lookup( index: &str, term: &str, ) -> Result>> { - let mut prefix = format!("index::{table}::{index}::").into_bytes(); + let mut prefix = index_prefix(table, index); prefix.extend_from_slice(term.as_bytes()); prefix.push(0); let mut cursor: Option> = None; @@ -77,7 +78,11 @@ pub async fn fulltext_lookup( type Entry = (Vec, Vec); fn index_prefix(table: &str, index: &str) -> Vec { - format!("index::{table}::{index}::").into_bytes() + index_prefix_storage(table, index) +} + +fn index_prefix_storage(storage_table: &str, index: &str) -> Vec { + format!("index::{storage_table}::{index}::").into_bytes() } /// Public key prefix covering every entry of a secondary index, for an ordered @@ -92,16 +97,17 @@ pub fn index_scan_prefix(table: &str, index: &str) -> Vec { /// with the row's data key as value, so walking this prefix yields the NULL rows /// ordered by primary key (the stable-pagination tiebreaker for the NULL block). pub fn indexnull_scan_prefix(table: &str, index: &str) -> Vec { - format!("indexnull::{table}::{index}::").into_bytes() + indexnull_prefix_storage(table, index) } -/// Entry key for a NULL-keyed row in a single-column index: the NULL prefix -/// followed by the row's clustered primary key (so NULLs never collide and are -/// ordered by PK). -fn null_entry_key(table: &str, index: &str, data_key: &[u8]) -> Vec { - let mut k = indexnull_scan_prefix(table, index); - k.extend_from_slice(&data_key[data_prefix(table).len()..]); - k +fn indexnull_prefix_storage(storage_table: &str, index: &str) -> Vec { + format!("indexnull::{storage_table}::{index}::").into_bytes() +} + +fn null_entry_key_storage(storage_table: &str, index: &str, clustered_key: &[u8]) -> Vec { + let mut key = indexnull_prefix_storage(storage_table, index); + key.extend_from_slice(clustered_key); + key } /// Prefix for all entries with a given tuple of column values (equality), @@ -118,21 +124,16 @@ fn value_prefix( Ok(k) } -fn value_prefix_encoded(table: &str, index: &str, encoded: &[u8]) -> Vec { - let mut key = index_prefix(table, index); - key.extend_from_slice(encoded); - key.push(0); - key -} - -fn entry_key_encoded( - table: &str, +fn entry_key_encoded_storage( + storage_table: &str, index: &str, encoded: &[u8], clustered_key: &[u8], unique: bool, ) -> Vec { - let mut key = value_prefix_encoded(table, index, encoded); + let mut key = index_prefix_storage(storage_table, index); + key.extend_from_slice(encoded); + key.push(0); if !unique { key.extend_from_slice(clustered_key); } @@ -160,7 +161,8 @@ pub fn partition_entries_for_row( ) -> Result<(Vec, Vec)> { let mut nonuniq = Vec::new(); let mut uniq = Vec::new(); - let clustered_key = &data_key[b"data::".len() + def.name.len() + b"::".len()..]; + let storage_table = def.storage_name(); + let clustered_key = &data_key[def.data_prefix().len()..]; for idx in &def.indexes { if idx.vector { continue; @@ -180,14 +182,20 @@ pub fn partition_entries_for_row( // `indexnull::` keyspace (never unique -- NULLs don't collide). if idx.indexes_nulls && idx.cols.len() == 1 && row[idx.cols[0]].is_null() { nonuniq.push(( - null_entry_key(&def.name, &idx.name, data_key), + null_entry_key_storage(&storage_table, &idx.name, clustered_key), data_key.to_vec(), )); } continue; }; let entry = ( - entry_key_encoded(&def.name, &idx.name, &encoded, clustered_key, idx.unique), + entry_key_encoded_storage( + &storage_table, + &idx.name, + &encoded, + clustered_key, + idx.unique, + ), data_key.to_vec(), ); if idx.unique { @@ -203,6 +211,7 @@ pub fn partition_entries_for_row( /// existence checks on paths that cannot use writer-side collision detection. pub fn unique_probe_keys(def: &TableDef, row: &[Value]) -> Result>> { let mut out = Vec::new(); + let storage_table = def.storage_name(); for idx in &def.indexes { if idx.vector || !idx.unique { continue; @@ -214,7 +223,10 @@ pub fn unique_probe_keys(def: &TableDef, row: &[Value]) -> Result>> let Some(encoded) = encoded else { continue; }; - out.push(value_prefix_encoded(&def.name, &idx.name, &encoded)); + let mut key = index_prefix_storage(&storage_table, &idx.name); + key.extend_from_slice(&encoded); + key.push(0); + out.push(key); } Ok(out) } @@ -232,7 +244,8 @@ pub fn entries_for_row( data_key: &[u8], ) -> Result, Vec)>> { let mut out = Vec::new(); - let clustered_key = &data_key[b"data::".len() + def.name.len() + b"::".len()..]; + let storage_table = def.storage_name(); + let clustered_key = &data_key[def.data_prefix().len()..]; for idx in &def.indexes { if idx.vector { continue; // vector indexes are maintained separately @@ -251,14 +264,20 @@ pub fn entries_for_row( // Single-column NULL-indexing (see `partition_entries_for_row`). if idx.indexes_nulls && idx.cols.len() == 1 && row[idx.cols[0]].is_null() { out.push(( - null_entry_key(&def.name, &idx.name, data_key), + null_entry_key_storage(&storage_table, &idx.name, clustered_key), data_key.to_vec(), )); } continue; }; out.push(( - entry_key_encoded(&def.name, &idx.name, &encoded, clustered_key, idx.unique), + entry_key_encoded_storage( + &storage_table, + &idx.name, + &encoded, + clustered_key, + idx.unique, + ), data_key.to_vec(), )); } diff --git a/crates/elyra-engine/src/lib.rs b/crates/elyra-engine/src/lib.rs index 122e574..b97ab7a 100644 --- a/crates/elyra-engine/src/lib.rs +++ b/crates/elyra-engine/src/lib.rs @@ -114,7 +114,8 @@ impl Engine { /// Must run before the server accepts connections: no query may observe a /// half-migrated keyspace. pub async fn migrate_collation(&self) -> elyra_core::Result<()> { - crate::collmig::migrate(&self.session()).await + crate::collmig::migrate(&self.session()).await?; + crate::exec::resume_generation_cleanup(&self.db).await } pub fn session(&self) -> Session { diff --git a/crates/elyra-engine/src/session.rs b/crates/elyra-engine/src/session.rs index b1036b4..232fe1e 100644 --- a/crates/elyra-engine/src/session.rs +++ b/crates/elyra-engine/src/session.rs @@ -735,6 +735,12 @@ impl Session { mem: _, undo_mem: _, } = tx; + let catalog_changed = puts + .keys() + .any(|key| key.starts_with(b"catalog::") || key.starts_with(b"sys::trigger::")) + || deletes + .iter() + .any(|key| key.starts_with(b"catalog::") || key.starts_with(b"sys::trigger::")); let ranges = if serializable { coalesce_ranges(ranges) @@ -764,7 +770,9 @@ impl Session { .filter(|key| !is_meta(key) && !range_covers(key)) .cloned(), ); - keyset.extend(locked.iter().filter(|k| !is_meta(k)).cloned()); + // Explicitly locked keys include table write-sequence metadata used by + // shadow rewrites to detect any concurrent table mutation. + keyset.extend(locked.iter().cloned()); if serializable { keyset.extend(reads.iter().filter(|k| !is_meta(k)).cloned()); } @@ -807,7 +815,8 @@ impl Session { let put_vec: Vec<(Vec, Vec)> = puts.into_iter().collect(); let del_vec: Vec> = deletes.into_iter().collect(); // On conflict the transaction is already cleared above -> aborted. - self.db + let result = self + .db .commit_validated( Validation { keys: expected, @@ -816,7 +825,11 @@ impl Session { put_vec, del_vec, ) - .await + .await; + if result.is_ok() && catalog_changed { + crate::catalog::bump_epoch(); + } + result } pub fn rollback(&self) { @@ -968,16 +981,16 @@ impl Session { deletes: Vec>, ) -> Result<()> { // Any write to a `catalog::` key changes a table definition; bump the - // catalog epoch so cached TableDefs are refreshed. Bumping eagerly (even - // for a buffered transactional write that may roll back) is safe -- it - // only forces a re-read, never serves stale schema. - if puts + // catalog epoch so cached TableDefs are refreshed. Bump eagerly for + // transactional visibility and again after a successful storage commit, + // closing the window where another session could cache the old schema. + let catalog_changed = puts .iter() .any(|(k, _)| k.starts_with(b"catalog::") || k.starts_with(b"sys::trigger::")) || deletes .iter() - .any(|k| k.starts_with(b"catalog::") || k.starts_with(b"sys::trigger::")) - { + .any(|k| k.starts_with(b"catalog::") || k.starts_with(b"sys::trigger::")); + if catalog_changed { crate::catalog::bump_epoch(); } crate::catalog::note_feature_writes(&puts, &deletes); @@ -1043,7 +1056,11 @@ impl Session { return Ok(()); } } - self.db.commit(puts, deletes).await + let result = self.db.commit(puts, deletes).await; + if result.is_ok() && catalog_changed { + crate::catalog::bump_epoch(); + } + result } } diff --git a/crates/elyra-engine/src/stream.rs b/crates/elyra-engine/src/stream.rs index 768988e..818d4f5 100644 --- a/crates/elyra-engine/src/stream.rs +++ b/crates/elyra-engine/src/stream.rs @@ -13,7 +13,7 @@ use elyra_core::{ColumnType, Error, Result, Schema, Value}; use elyra_storage::Db; use sqlparser::ast::Expr; -use crate::catalog::{data_prefix, TableDef}; +use crate::catalog::TableDef; use crate::predicate; /// How many storage rows to pull per underlying scan step. @@ -130,7 +130,7 @@ impl RowStream { schema: spec.out_schema, src: Source::Scan(Scan { db, - prefix: data_prefix(&table.name), + prefix: table.data_prefix(), cursor: None, full_schema: table.schema.clone(), projection: spec.projection, diff --git a/crates/elyra-engine/src/vindex.rs b/crates/elyra-engine/src/vindex.rs index 6f9f004..faf7617 100644 --- a/crates/elyra-engine/src/vindex.rs +++ b/crates/elyra-engine/src/vindex.rs @@ -22,7 +22,7 @@ use crate::session::Session; use elyra_core::{Error, Result, Value}; use elyra_vector::{Hnsw, HnswParts, Metric}; -use crate::catalog::{data_prefix, wcount_key, TableDef}; +use crate::catalog::{wcount_key, TableDef}; /// On-disk vector-index cache (see ESQL-27). The graph is a regenerable cache, /// so it lives in a sibling directory `.vidx/` (like `.raftstate`), @@ -358,7 +358,7 @@ async fn scan_current( def: &TableDef, col: usize, ) -> Result<(usize, Vec<(Vec, Vec)>)> { - let prefix = data_prefix(&def.name); + let prefix = def.data_prefix(); let mut cursor: Option> = None; let mut out: Vec<(Vec, Vec)> = Vec::new(); let mut dim = 0usize; @@ -506,7 +506,7 @@ fn reconcile( #[cfg(test)] mod tests { use super::*; - use crate::catalog::TableDef; + use crate::catalog::{data_prefix, TableDef}; use crate::lockmgr::LockManager; use elyra_core::{ColumnDef, ColumnType, Schema}; use elyra_storage::Db; @@ -541,6 +541,7 @@ mod tests { col_meta: Vec::new(), checks: Vec::new(), foreign_keys: Vec::new(), + storage_generation: 0, } } diff --git a/crates/elyra-server/tests/wire.rs b/crates/elyra-server/tests/wire.rs index d12bbf2..25a46bf 100644 --- a/crates/elyra-server/tests/wire.rs +++ b/crates/elyra-server/tests/wire.rs @@ -1017,6 +1017,149 @@ async fn alter_table_add_primary_key_reclusters_existing_rows_atomically() { } } +#[tokio::test] +async fn shadow_primary_key_generation_supports_subsequent_table_operations() { + let srv = TestServer::start().await; + let mut connection = srv.conn().await; + + connection + .query_drop("CREATE TABLE gen_ops (id INT, label TEXT, INDEX label_idx(label))") + .await + .unwrap(); + connection + .query_drop("INSERT INTO gen_ops VALUES (1, 'one'), (2, 'two'), (3, 'three')") + .await + .unwrap(); + connection + .query_drop("ALTER TABLE gen_ops ADD PRIMARY KEY (id)") + .await + .unwrap(); + connection + .query_drop("UPDATE gen_ops SET label = 'second' WHERE id = 2") + .await + .unwrap(); + let indexed: Option = connection + .query_first("SELECT id FROM gen_ops WHERE label = 'second'") + .await + .unwrap(); + assert_eq!(indexed, Some(2)); + + connection + .query_drop("DELETE FROM gen_ops WHERE id = 1") + .await + .unwrap(); + connection + .query_drop("INSERT INTO gen_ops VALUES (4, 'four')") + .await + .unwrap(); + connection + .query_drop("RENAME TABLE gen_ops TO gen_ops_new") + .await + .unwrap(); + let rows: Vec<(i64, String)> = connection + .query("SELECT id, label FROM gen_ops_new ORDER BY id") + .await + .unwrap(); + assert_eq!( + rows, + vec![ + (2, "second".into()), + (3, "three".into()), + (4, "four".into()), + ] + ); + + connection + .query_drop("ANALYZE TABLE gen_ops_new") + .await + .unwrap(); + connection + .query_drop("TRUNCATE TABLE gen_ops_new") + .await + .unwrap(); + let count: Option = connection + .query_first("SELECT COUNT(*) FROM gen_ops_new") + .await + .unwrap(); + assert_eq!(count, Some(0)); + connection + .query_drop("INSERT INTO gen_ops_new VALUES (5, 'after truncate')") + .await + .unwrap(); +} + +#[tokio::test] +async fn recreated_table_does_not_reuse_a_generation_being_cleaned() { + let srv = TestServer::start().await; + let mut connection = srv.conn().await; + + connection + .query_drop("CREATE TABLE generation_reuse (id INT, payload TEXT)") + .await + .unwrap(); + for start in (0..20_000).step_by(1_000) { + let values = (start..start + 1_000) + .map(|id| format!("({id},'old-{id}')")) + .collect::>() + .join(","); + connection + .query_drop(format!("INSERT INTO generation_reuse VALUES {values}")) + .await + .unwrap(); + } + connection + .query_drop("ALTER TABLE generation_reuse ADD PRIMARY KEY (id)") + .await + .unwrap(); + + // The ALTER returns while its old physical generation is reclaimed. DROP + // and CREATE must retain a generation watermark so this new table cannot + // be placed back into the keyspace that cleanup is still deleting. + connection + .query_drop("DROP TABLE generation_reuse") + .await + .unwrap(); + connection + .query_drop("CREATE TABLE generation_reuse (id INT PRIMARY KEY, payload TEXT)") + .await + .unwrap(); + connection + .query_drop("INSERT INTO generation_reuse VALUES (1,'new')") + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let rows: Vec<(i64, String)> = connection + .query("SELECT id, payload FROM generation_reuse") + .await + .unwrap(); + assert_eq!(rows, vec![(1, "new".into())]); + + // RENAME must also honor a watermark left by an earlier table that used + // the destination name. + connection + .query_drop("DROP TABLE generation_reuse") + .await + .unwrap(); + connection + .query_drop("CREATE TABLE generation_source (id INT PRIMARY KEY, payload TEXT)") + .await + .unwrap(); + connection + .query_drop("INSERT INTO generation_source VALUES (2,'renamed')") + .await + .unwrap(); + connection + .query_drop("RENAME TABLE generation_source TO generation_reuse") + .await + .unwrap(); + let renamed: Vec<(i64, String)> = connection + .query("SELECT id, payload FROM generation_reuse") + .await + .unwrap(); + assert_eq!(renamed, vec![(2, "renamed".into())]); +} + #[tokio::test] async fn alter_add_primary_key_rejects_a_concurrent_post_scan_insert() { let srv = TestServer::start().await;