diff --git a/doc/developer/design/20260716_pager_hydration_experiment.md b/doc/developer/design/20260716_pager_hydration_experiment.md new file mode 100644 index 0000000000000..a4594a33148fa --- /dev/null +++ b/doc/developer/design/20260716_pager_hydration_experiment.md @@ -0,0 +1,320 @@ +# Pager hydration experiment + +## Goal + +Measure the impact of the column-paged batcher's spill-to-disk mechanism (the +"pager") on hydration time, as a function of workload size. +The experiment holds the replica size fixed and sweeps the TPCH scale factor, so +the only thing that changes within a size is how much arrangement data the +workload produces, and therefore how much the pager pushes to disk. + +The question we answer: for a given workload, how much slower (or faster) does a +replica hydrate on the paged batcher with spill versus the legacy batcher it +replaces? + +## Background + +The pager is the spill path of the column-paged merge batcher, active at arrange +sites when building arrangements. +Three replica-scoped dynamic configuration parameters govern it. + +* `enable_column_paged_batcher`: use the paged batcher at all. + When `false`, arrange sites use the legacy columnation path. +* `enable_column_paged_batcher_spill`: allow the pager to evict chunks to the + disk backend under memory pressure. + This is the on/off switch for the pager. + With it `false`, the pager keeps every chunk resident regardless of budget, so + no disk is used. +* `column_paged_batcher_lz4`: compress spilled chunks. + +Two related parameters are global, not replica-scoped, so they cannot differ +between two replicas of one cluster. +`column_paged_batcher_budget_fraction` sets the resident-byte budget before +spill (code default 5% of the replica memory limit, floored at 128 MiB; set to +0.01 environment-wide on the target region). +`column_paged_batcher_swap_pageout` eagerly evicts compressed swap-backend +chunks (code default off; set on environment-wide on the target region). +They only affect the `pager_on` replica, because `pager_off` runs the legacy +batcher and does not use the pager at all. + +Hydration is measured by wall-clock, using `mz_internal.mz_hydration_statuses`, +which exposes `object_id`, `replica_id`, and `hydrated`. +The script records a start timestamp, then polls this view for the five named +materialized views on both replicas, recording the timestamp at which each +`(object_id, replica_id)` pair first flips to `hydrated=true`. +The hydration time for a pair is that timestamp minus the start. + +The internally-measured `mz_internal.mz_compute_hydration_times.time_ns` is not +usable here. Verified against the target region, that collection populates +`time_ns` for index (arrangement) exports but leaves it NULL for materialized +view exports, which never appear in the underlying per-worker hydration-time log. +Since the workload is materialized views, wall-clock is the only reliable metric. +This gap is tracked as CLU-175 +(https://linear.app/materializeinc/issue/CLU-175); it does not block the +experiment. + +Wall-clock resolution equals the poll interval (about 2 seconds). Hydration at +the scale factors of interest runs for many seconds to minutes, so the interval +is negligible against the measured durations. + +## Flag mapping + +The two replicas differ only by name. +The name-to-flag mapping is applied out of band, outside this script, through +per-replica scoped system-parameter overrides. +The script only creates replicas with the fixed names `pager_on` and +`pager_off`. + +The mechanism, verified against the staging region via +`mz_internal.mz_replica_system_parameters`, is minimal. +The environment-wide `ALTER SYSTEM` settings already have the pager fully on: +`enable_column_paged_batcher=on`, `enable_column_paged_batcher_spill=on`, +`column_paged_batcher_lz4=on`, `column_paged_batcher_swap_pageout=on`, and +`column_paged_batcher_budget_fraction=0.01`. +`pager_on` carries no scoped override, so it inherits that fully-on state. +`pager_off` carries exactly one scoped override, `enable_column_paged_batcher=false`, +which routes it to the legacy columnation batcher and makes the spill and lz4 +settings moot. + +| Effective state | `pager_on` | `pager_off` | +| --- | --- | --- | +| `enable_column_paged_batcher` | on (inherited) | `false` (override) | +| `enable_column_paged_batcher_spill` | on (inherited) | moot | +| `column_paged_batcher_lz4` | on (inherited) | moot | +| `column_paged_batcher_swap_pageout` | on (inherited) | moot | +| `column_paged_batcher_budget_fraction` | 0.01 (inherited) | moot | + +`pager_on` therefore runs the paged batcher with spill, lz4 compression, and +eager swap pageout, at a 1% resident budget. +`pager_off` runs the legacy columnation batcher (`Col2ValBatcher` / +`RowRowBuilder`), the path that shipped before the paged batcher. +This makes `pager_off` the real-world production alternative rather than a +never-spilling paged batcher. + +The measured delta therefore covers the whole package: adopting the paged +batcher and enabling spill. +It does not separate the batcher-adoption cost from the disk-I/O cost. +That separation is possible by adding a third replica that runs the paged +batcher with spill disabled, but it is out of scope here. + +## Topology + +The experiment runs against a single Materialize region in staging, driven over +pgwire. + +```mermaid +graph LR + subgraph src["cluster: ldgen_sf<N> (source, small, single ingest worker)"] + S["LOAD GENERATOR TPCH
SCALE FACTOR N"] + end + subgraph test["cluster: test_sf<N> (unmanaged, 3200cc)"] + R1["replica: pager_on"] + R2["replica: pager_off"] + end + S -->|persist| MV["5 materialized views:
Q3 Q5 Q9 Q18 Q21"] + MV --> R1 + MV --> R2 +``` + +The load generator runs on its own cluster, one per scale factor, separate from +the cluster under test. +Ingestion CPU never contaminates hydration timing. + +The test cluster is unmanaged with two explicitly-named replicas. +Unmanaged is required because managed replicas are identical, but the experiment +needs two replicas that differ by name so the out-of-band mapping applies +different pager flags. +Both replicas are the same size, `3200cc` (62 vCPU, 470 GiB memory, 705 GiB +disk, single process), so the only difference between them is the flag. +The `r8gd_cpu-62` size the design first targeted exists in the size map but is +not among this region's `allowed_cluster_replica_sizes`; `3200cc` matches its +shape (62 vCPU, ~470 GiB memory, ~705 GiB disk) and is permitted. +`M.1-8xlarge` (equivalently `xlarge`) is a whole-machine size with the same +62 vCPU and 470 GiB memory but 2820 GiB disk, four times the disk of `3200cc`. +It is the preferred fixed size because the larger disk removes any spill +capacity confound at high scale factors, with no compute difference. +This size is the `--size` default; the flag exists to retarget the experiment, +and whatever value it holds applies to both replicas identically. + +Both replicas sit in the same cluster, so they hydrate the exact same +materialized-view dataflows, are directly comparable, and each reports its own +`time_ns`. + +## Load generator lifecycle + +The TPCH load generator is single-threaded, so it is slow and must not be +restarted needlessly. + +* Each scale factor gets its own dedicated `ldgen_sf` cluster and TPCH source, + created and ingested exactly once. +* The source cluster is small; extra workers do not speed up single-threaded + ingest. +* On start, if `ldgen_sf` exists and its snapshot frontier has advanced + (ingest complete), the script skips straight to the compute phase. + It never restarts an already-ingested source. +* Teardown never touches sources by default. + Only the test cluster's replicas churn between trials. + A `--purge` flag is required to remove sources. + +## Workload + +Five join- and aggregation-heavy TPCH queries are materialized: Q3, Q5, Q9, Q18, +Q21. +These build the largest internal arrangements, which is where the pager is +exercised. +Lighter TPCH queries are excluded because they dilute the pager signal. + +The source is created with +`CREATE SOURCE ... FROM LOAD GENERATOR TPCH (SCALE FACTOR N) FOR ALL TABLES`, +which exposes the eight TPCH tables (`lineitem`, `orders`, `customer`, +`supplier`, `nation`, `region`, `part`, `partsupp`) with the standard TPCH +column names (`l_*`, `o_*`, and so on). + +The materialized-view bodies are the `SELECT` statements for Q03, Q05, Q09, Q18, +and Q21 in `test/sqllogictest/tpch_create_materialized_view.slt`. +That file models the same table and column names the load generator exposes, so +the query bodies are reused verbatim, wrapped in +`CREATE MATERIALIZED VIEW q IN CLUSTER test_sf AS `. +The script reads the bodies from that file rather than transcribing them, so the +queries stay in sync with the canonical definitions. + +## Measured scenarios + +Two hydration paths are measured per scale factor, because materialized views +behave differently on first build versus replica restart. + +**Initial hydration.** +With both replicas up and idle, create the five materialized views. +Each replica builds arrangements from the source snapshot, and one replica wins +the persist write. +This measures compute-from-scratch plus writing output. + +**Re-hydration.** +The materialized views are already populated. +For each flavor, drop the cluster replica then re-create it, which is the +replication-factor 0 to 1 transition. +The fresh replica rebuilds all in-memory arrangements from persisted inputs. +This measures arrangement rebuild without recomputing output. + +Only replicas churn during re-hydration trials. +The materialized views are never dropped, so their dataflows stay live and a +re-added replica genuinely receives the dataflow and rehydrates it, producing a +real `time_ns`. + +## Sweep and matrix + +* Scale factors: 1, 10, 30, 100. +* Objects: five materialized views (Q3, Q5, Q9, Q18, Q21). +* Flavors: `pager_on`, `pager_off`. +* Scenarios: initial and re-hydration, three trials each. + An initial trial drops and re-creates all five views (with both replicas up), + then measures. + A re-hydration trial drops and re-creates both replicas (views left in place), + then measures. + +A full measurement cell is identified by +`scale_factor x scenario x flavor x query x trial`. +Per scale factor this is 2 scenarios x 3 trials x 2 flavors x 5 queries = 60 +`time_ns` values, and 240 across the four scale factors. +Both flavors are measured within the same trial, since the two replicas coexist. + +With the environment-wide budget at 1% of memory (~4.7 GiB on the 470 GiB +`3200cc` size), `pager_on` should start spilling at a lower scale factor than +the 5% default would require. +The pager metrics (see Live monitoring) confirm per trial whether spill actually +occurred; if a scale factor shows no spill, the budget fraction can be lowered +further, which touches only `pager_on`. + +## Measurement procedure + +For one trial: + +1. Record the start timestamp `t0` from `SELECT now()`. +2. Bring both replicas into the target state for the scenario (create views, or + drop and re-create replicas). +3. Poll `mz_internal.mz_hydration_statuses`, filtered to the five named views on + both replicas, about every 2 seconds. On each poll, for any + `(object_id, replica_id)` pair newly `hydrated=true` and not yet recorded, + capture `now()` as that pair's hydration completion time. Stop when all ten + pairs are recorded or a timeout is hit. +4. Append one row per pair to the results CSV: + `sf, scenario, flavor, query, trial, hydration_seconds`, where + `hydration_seconds = completion - t0` and `query` is the stable view name. + +The poll must filter to the five named views. `mz_hydration_statuses` lists every +dataflow-backed object on a replica, including system introspection arrangements, +so an unfiltered count reaches the target long before the views hydrate. +Results are keyed by the stable view name because `object_id` changes every time +a view is dropped and re-created, which happens on every initial-hydration +trial. Correlating by `object_id` across trials would break. + +Both replicas coexist in the cluster and hydrate the same dataflows, so a single +trial measures both flavors at once. +All completion times are captured while the replicas are alive, before any +teardown. + +## Live monitoring + +The wall-clock hydration time answers how long, but not why. +While the experiment runs, resource metrics are read from Grafana (Prometheus) +to characterize each trial: memory utilization, user and system CPU time +(rtime and stime), CPU utilization, and disk throughput. + +To correlate Grafana series with trials, the script records a window manifest in +addition to the results CSV. +Each trial emits one row: `sf, scenario, flavor, replica_id, started_at, +ended_at`, where the timestamps bracket the hydration window (replica or view +creation until all objects report hydrated). +`replica_id` is the catalog replica id, read from `mz_cluster_replicas`, used to +select the replica's series in Prometheus. + +Metrics read per window: + +* Memory: replica RSS and memory-limit utilization. +* CPU: user and system CPU seconds (rtime, stime) and CPU utilization. +* Disk: read and write throughput to the spill backend. +* Pager activity, from the process-wide pager metrics, which directly show + whether and how much the `pager_on` replica spilled: `mz_column_pager_pageouts_total`, + `mz_column_pager_paged_bytes_out_total`, `mz_column_pager_pageins_total`, + `mz_column_pager_pagein_bytes_total`, `mz_column_pager_budget_remaining_bytes`, + `mz_column_pager_budget_configured_bytes`. + +The pager metrics are only nonzero on `pager_on`, since `pager_off` runs the +legacy batcher. +Their delta across a window confirms the spill volume that the hydration-time +delta is attributed to. + +Monitoring is read-only and does not gate the experiment. +The script proceeds on the wall-clock signal; Grafana queries run alongside, +keyed by the window manifest, so metric collection can also happen after the run +over the recorded windows. + +## Script + +A single Python program drives the experiment over pgwire. + +* Idempotent and resumable per scale factor. + Existing ingested sources are detected and reused. +* Flags: `--scale-factors`, `--trials`, `--size`, `--keep` (skip teardown), + `--purge` (also remove sources). +* Output: a CSV of raw `hydration_seconds` rows, a window-manifest CSV + (`sf, scenario, flavor, replica_id, started_at, ended_at`) for Grafana + correlation, and a printed summary table of median `hydration_seconds` per + `scale_factor x scenario x flavor`. +* Designed to run overnight, since SF 100 initial hydration recomputes the views + three times. + +## Out of scope + +* Sweeping replica disk_limit across size families. + That confounds disk with cpu, memory, and worker count, since those scale + together. +* Sweeping `column_paged_batcher_budget_fraction` as an independent axis. + It is global and cannot differ between the two replicas in one cluster. +* Separating batcher-adoption cost from disk-I/O cost via a third + spill-disabled paged-batcher replica. + The two-replica design compares the paged batcher with spill against the + legacy batcher as a single package. +* Eager swap pageout (`column_paged_batcher_swap_pageout`). + Left at its default; a follow-on variant once the base spill effect is + characterized. lz4 compression is in scope and enabled on `pager_on`. diff --git a/misc/experiments/pager-hydration/README.md b/misc/experiments/pager-hydration/README.md new file mode 100644 index 0000000000000..3a0f76891c850 --- /dev/null +++ b/misc/experiments/pager-hydration/README.md @@ -0,0 +1,117 @@ +# Pager hydration experiment + +Measures the impact of the column-paged batcher's spill-to-disk mechanism (the +"pager") on hydration time, comparing a paged-batcher replica against the legacy +columnation batcher across TPCH scale factors. + +Full design and rationale: `doc/developer/design/20260716_pager_hydration_experiment.md`. + +## What it does + +Per scale factor it builds, over a TPCH load-generator source, a workload of +18 base-table pk/fk indexes, 5 indexed views, and 5 materialized views for the +join-heavy queries Q03, Q05, Q09, Q18, Q21. +It then measures how long two configurations take to hydrate that workload. + +* `pager_on`: the column-paged batcher with spill and lz4. +* `pager_off`: the legacy columnation batcher. + +Hydration time comes from `mz_internal.mz_compute_hydration_times.time_ns` for +index and join or aggregation MV exports, with wall-clock via +`mz_internal.mz_hydration_statuses` as the cross-check and the fallback. + +The source uses a `TICK INTERVAL` so its write frontier never goes final. +This is required. With a bounded source the frontier is final and +`mz_hydration_statuses` reports a freshly added replica as hydrated before it +has loaded anything, which makes re-hydration timing meaningless. +The TPCH tick inserts and retracts, so total data size stays roughly constant. + +## Two runners + +### `run.py`: staging or cloud region + +Drives a region over pgwire via `bin/mz`. +The `pager_on` and `pager_off` axis is two replicas in one cluster that differ +only by name. +The name-to-flag mapping is a per-replica scoped system parameter override, +applied out of band by the delivery-service config sync, not by this script +(see `doc/developer/design/20260609_scoped_feature_flags.md`). +The script only creates replicas named `pager_on` and `pager_off`, and verifies +the override resolved via `mz_internal.mz_replica_system_parameters`. + +Prerequisites. + +* A writable `mz` config path. `~/.config/materialize/mz.toml` is read-only in + some sandboxes, so copy it somewhere writable and pass `--config`. +* The environment-wide pager flags set to the intended `pager_on` state + (`enable_column_paged_batcher`, `enable_column_paged_batcher_spill`, + `column_paged_batcher_lz4`, and the tuning knobs + `column_paged_batcher_budget_fraction`, `column_paged_batcher_swap_pageout`). + `pager_on` inherits these; `pager_off` carries a scoped + `enable_column_paged_batcher=false` override. +* Replica sizes that are in the region's `allowed_cluster_replica_sizes`. + `M.1-8xlarge` (whole machine, large disk) is the default. + +Example. + +``` +python3 run.py \ + --config /path/to/writable/mz.toml \ + --region aws/us-east-1 --profile staging \ + --scale-factors 1,10,30,100 --trials 3 --size M.1-8xlarge --tick 1s \ + --outdir results +``` + +### `run_local.py`: local `bin/environmentd` + +For fast iteration without a region. +Scoped per-replica parameters are not reproducible locally, so the pager axis is +a global `ALTER SYSTEM SET` applied sequentially: set the flag, add one replica, +hydrate, measure, drop, then flip the flag. +Application DDL and queries go to the external SQL port; `ALTER SYSTEM` goes to +the internal SQL port as `mz_system`. + +Prerequisites. + +* `bin/environmentd` running (with CockroachDB). +* Sizes are local `scale=1,workers=N` strings, small by default. + +Example. + +``` +python3 run_local.py --scale-factors 1,3 --trials 3 +``` + +## Outputs + +* `results.csv`: one row per object per condition per trial: + `sf, flavor, object, kind, trial, hydration_seconds, time_ns`. +* `windows.csv` (staging runner): per-trial replica id and wall-clock window, + for correlating Grafana resource metrics. + +Both runners are idempotent and resumable: a completed cell is skipped on +relaunch, and an already-ingested source is reused rather than re-created. + +## Grafana + +The staging runner does not pull metrics; it records the window manifest so an +operator or an MCP-enabled agent can pull per-window resource metrics from the +region's Prometheus. +Useful series, filtered by +`cluster_environmentd_materialize_cloud_replica_name`: + +* `mz_memory_limiter_memory_usage_bytes`: peak memory (RAM plus swap). +* `mz_column_pager_paged_bytes_out_total`: pager spill volume (pager path only, + not OS swap). +* `container_cpu_usage_seconds_total`, `container_memory_usage_bytes`: per pod. + +## Notes + +* `time_ns` is NULL for bare single-reduce materialized views (CLU-175) but + populated for indexes and for join or aggregation MVs. +* Both conditions can reach disk. `pager_on` via the managed lz4 path, + `pager_off` via kernel swap on a swap-enabled size. The pager metrics count + only the managed path. +* The column pager bounds transient merge-batcher memory, not the final + arrangement, so it does not lower peak RSS and does not by itself prevent an + out-of-memory kill when the final arrangements exceed the replica's ceiling. diff --git a/misc/experiments/pager-hydration/run.py b/misc/experiments/pager-hydration/run.py new file mode 100644 index 0000000000000..03c4b562eaadb --- /dev/null +++ b/misc/experiments/pager-hydration/run.py @@ -0,0 +1,485 @@ +#!/usr/bin/env python3 +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +"""Pager hydration experiment. + +Measures the impact of the column-paged batcher's spill-to-disk mechanism (the +"pager") on hydration time and memory, as a function of TPCH scale factor, +against a staging Materialize region. + +Design: doc/developer/design/20260716_pager_hydration_experiment.md + +Two replicas per test cluster differ only by name; the name-to-flag mapping is +applied out of band via per-replica scoped system parameters: + + pager_on : inherits environment-wide (paged batcher + spill + lz4 on) + pager_off: scoped override enable_column_paged_batcher=false (legacy batcher) + +Per scale factor the workload is, for five heavy TPCH queries (Q03/05/09/18/21), +both an indexed view and a materialized view, plus the base-table pk/fk indexes +the joins use. Indexes report a precise internal hydration time +(mz_compute_hydration_times.time_ns); materialized views do not (CLU-175), so +they are timed by wall-clock via mz_hydration_statuses. + +The source uses a TICK INTERVAL so its write frontier never goes final. Without +that, a bounded source's frontier is final and mz_hydration_statuses reports a +freshly added replica as hydrated before it has loaded anything, which makes +re-hydration timing meaningless. The tick inserts and retracts, so total data +size stays roughly constant. +""" + +import argparse +import csv +import os +import re +import subprocess +import sys +import time +from pathlib import Path + +# --- Configuration -------------------------------------------------------- + +# M.1-8xlarge: 62 vCPU, 470 GiB memory, 2820 GiB disk, single process +# (whole machine). Same compute as 3200cc but 4x the disk, so pager_on has spill +# headroom at high scale factors and disk capacity is not a confound. +DEFAULT_SIZE = "M.1-8xlarge" +DEFAULT_SOURCE_SIZE = "200cc" # 4 vCPU: headroom for the single ingest thread. +DEFAULT_SCALE_FACTORS = [1, 10, 30, 100] +DEFAULT_TRIALS = 3 +DEFAULT_TICK = "1s" +QUERIES = ["Q03", "Q05", "Q09", "Q18", "Q21"] +REPLICAS = ["pager_on", "pager_off"] +SCENARIOS = ["initial", "rehydrate"] +SLT = "test/sqllogictest/tpch_create_materialized_view.slt" + +# Base-table pk/fk indexes (name, table, key columns). Mirrors the canonical +# TPCH index set for the tables the five queries join. +BASE_INDEXES = [ + ("pk_nation", "nation", "n_nationkey"), + ("fk_nation_region", "nation", "n_regionkey"), + ("pk_region", "region", "r_regionkey"), + ("pk_part", "part", "p_partkey"), + ("pk_supplier", "supplier", "s_suppkey"), + ("fk_supplier_nation", "supplier", "s_nationkey"), + ("pk_partsupp", "partsupp", "ps_partkey, ps_suppkey"), + ("fk_partsupp_part", "partsupp", "ps_partkey"), + ("fk_partsupp_supp", "partsupp", "ps_suppkey"), + ("pk_customer", "customer", "c_custkey"), + ("fk_customer_nation", "customer", "c_nationkey"), + ("pk_orders", "orders", "o_orderkey"), + ("fk_orders_cust", "orders", "o_custkey"), + ("pk_lineitem", "lineitem", "l_orderkey, l_linenumber"), + ("fk_lineitem_order", "lineitem", "l_orderkey"), + ("fk_lineitem_part", "lineitem", "l_partkey"), + ("fk_lineitem_supp", "lineitem", "l_suppkey"), + ("fk_lineitem_partsupp", "lineitem", "l_partkey, l_suppkey"), +] + +POLL_INTERVAL = 3.0 +INGEST_POLL_INTERVAL = 15.0 + + +class Mz: + """Wrapper around `mz ... sql -- -tAq -c` for one region/profile.""" + + def __init__(self, mz_bin, config, region, profile): + self.base = [ + mz_bin, "--config", config, "--region", region, + "--profile", profile, "sql", "--", "-tAq", "-c", + ] + + def __call__(self, stmt, timeout=900, search_path=None): + env = os.environ.copy() + if search_path is not None: + # Set search_path at connection startup so a single DDL statement + # can use unqualified table names. A `SET ...; CREATE ...` prelude + # would form a multi-statement transaction, which DDL cannot enter. + env["PGOPTIONS"] = f"-c search_path={search_path}" + # Retry transient CLI/coordinator blips, which show up under heavy load + # at high scale factors as a non-zero exit with little or no stderr. A + # persistent error still surfaces after the final attempt. + last = None + for attempt in range(3): + try: + r = subprocess.run( + self.base + [stmt], capture_output=True, text=True, + timeout=timeout, env=env, + ) + except subprocess.TimeoutExpired as e: + last = f"timeout after {timeout}s" + time.sleep(5) + continue + if r.returncode == 0: + return [ + line for line in r.stdout.splitlines() + if line.strip() and not line.startswith("Time:") + ] + last = r.stderr or r.stdout + time.sleep(5) + raise RuntimeError(f"SQL failed after retries: {stmt}\n{last}") + + def one(self, stmt, timeout=900, search_path=None): + rows = self(stmt, timeout=timeout, search_path=search_path) + return rows[0] if rows else None + + +# --- Names ---------------------------------------------------------------- + +def schema(sf): + return f"sf{sf}" + + +def src_cluster(sf): + return f"ldgen_sf{sf}" + + +def test_cluster(sf): + return f"test_sf{sf}" + + +def extract_bodies(): + """Return {Qxx: SELECT body} for the target queries from the canonical slt.""" + text = Path(SLT).read_text() + bodies = {} + for q in QUERIES: + m = re.search(rf"CREATE MATERIALIZED VIEW {q} AS\n(.*?);\n", text, re.S) + if not m: + raise RuntimeError(f"missing {q} in {SLT}") + bodies[q] = m.group(1).strip() + return bodies + + +# --- Source --------------------------------------------------------------- + +def source_status(mz, sf): + return mz.one( + f"SELECT status FROM mz_internal.mz_source_statuses s " + f"JOIN mz_sources so ON s.id = so.id " + f"JOIN mz_schemas sc ON so.schema_id = sc.id " + f"WHERE sc.name = '{schema(sf)}' AND so.name = 'ldgen';" + ) + + +def ensure_source(mz, sf, source_size, tick): + """Create the ticking TPCH source for `sf`, or reuse a running one. + + The single-threaded generator must never be restarted, so a running source + is left untouched. Each scale factor lives in its own schema so per-table + subsources do not collide across factors. + """ + sch, cl = schema(sf), src_cluster(sf) + st = source_status(mz, sf) + if st == "running": + print(f" [sf{sf}] source running, reusing") + return + if st is None: + print(f" [sf{sf}] creating source cluster + TPCH SF{sf} (tick {tick})") + if not mz.one(f"SELECT 1 FROM mz_schemas WHERE name = '{sch}';"): + mz(f"CREATE SCHEMA {sch};") + if not mz.one(f"SELECT 1 FROM mz_clusters WHERE name = '{cl}';"): + mz(f"CREATE CLUSTER {cl} SIZE '{source_size}';") + mz( + f"CREATE SOURCE {sch}.ldgen IN CLUSTER {cl} " + f"FROM LOAD GENERATOR TPCH (SCALE FACTOR {sf}, TICK INTERVAL '{tick}') " + f"FOR ALL TABLES;" + ) + else: + print(f" [sf{sf}] source present but status={st}; waiting") + # A load-generator source reports `running` only after its snapshot is + # committed. Do not probe with count(*): that scan exceeds the SQL timeout + # at high scale factors. + while source_status(mz, sf) != "running": + print(f" [sf{sf}] ingest status={source_status(mz, sf)}") + time.sleep(INGEST_POLL_INTERVAL) + + +# --- Test cluster + workload --------------------------------------------- + +def ensure_test_cluster(mz, sf): + cl = test_cluster(sf) + if not mz.one(f"SELECT 1 FROM mz_clusters WHERE name = '{cl}';"): + mz(f"CREATE CLUSTER {cl} REPLICAS ();") + print(f" [sf{sf}] created unmanaged cluster {cl}") + + +def add_replicas(mz, sf, size): + for name in REPLICAS: + mz(f"CREATE CLUSTER REPLICA {test_cluster(sf)}.{name} SIZE '{size}';") + + +def drop_replicas(mz, sf): + for name in REPLICAS: + mz(f"DROP CLUSTER REPLICA IF EXISTS {test_cluster(sf)}.{name};") + + +def replica_ids(mz, sf): + rows = mz( + f"SELECT r.name, r.id FROM mz_cluster_replicas r " + f"JOIN mz_clusters c ON r.cluster_id = c.id " + f"WHERE c.name = '{test_cluster(sf)}';" + ) + return {row.split("|")[0]: row.split("|")[1] for row in rows} + + +def workload_exists(mz, sf): + return bool(mz.one( + f"SELECT 1 FROM mz_materialized_views mv " + f"JOIN mz_schemas sc ON mv.schema_id = sc.id " + f"WHERE sc.name = '{schema(sf)}' AND mv.name = 'mq03';" + )) + + +def create_workload(mz, sf, bodies): + """Base-table indexes, indexed views, and MVs for the five queries.""" + sch, cl = schema(sf), test_cluster(sf) + for name, table, cols in BASE_INDEXES: + mz(f"CREATE INDEX {name} IN CLUSTER {cl} ON {table} ({cols});", search_path=sch) + for q in QUERIES: + v = f"v{q.lower()}" + mz(f"CREATE VIEW {sch}.{v} AS {bodies[q]};", search_path=sch) + mz(f"CREATE DEFAULT INDEX IN CLUSTER {cl} ON {sch}.{v};", search_path=sch) + mz( + f"CREATE MATERIALIZED VIEW {sch}.m{q.lower()} IN CLUSTER {cl} " + f"AS {bodies[q]};", + search_path=sch, + ) + + +def drop_workload(mz, sf): + sch = schema(sf) + for q in QUERIES: + mz(f"DROP MATERIALIZED VIEW IF EXISTS {sch}.m{q.lower()};") + mz(f"DROP VIEW IF EXISTS {sch}.v{q.lower()} CASCADE;") + for name, _, _ in BASE_INDEXES: + mz(f"DROP INDEX IF EXISTS {sch}.{name};") + + +def object_map(mz, sf): + """Return {object_id: (label, kind)} for all indexes and MVs on the cluster. + + kind is 'index' (time_ns available) or 'mv' (wall-clock only). + """ + cl, sch = test_cluster(sf), schema(sf) + out = {} + for row in mz( + f"SELECT i.id, i.name FROM mz_indexes i " + f"JOIN mz_clusters c ON i.cluster_id = c.id " + f"JOIN mz_relations r ON i.on_id = r.id " + f"JOIN mz_schemas s ON r.schema_id = s.id " + f"WHERE c.name = '{cl}' AND s.name = '{sch}';" + ): + oid, name = row.split("|") + out[oid] = (name, "index") + for row in mz( + f"SELECT mv.id, mv.name FROM mz_materialized_views mv " + f"JOIN mz_schemas s ON mv.schema_id = s.id WHERE s.name = '{sch}';" + ): + oid, name = row.split("|") + out[oid] = (name, "mv") + return out + + +def check_flags(mz, sf): + return mz( + f"SELECT r.name, p.name, p.value " + f"FROM mz_internal.mz_replica_system_parameters p " + f"JOIN mz_cluster_replicas r ON p.replica_id = r.id " + f"JOIN mz_clusters c ON r.cluster_id = c.id " + f"WHERE c.name = '{test_cluster(sf)}' AND p.name LIKE '%column_paged_batcher%' " + f"ORDER BY r.name, p.name;" + ) + + +# --- Measurement ---------------------------------------------------------- + +def measure(mz, sf, objs, target_ids, timeout): + """Time hydration of `objs` on the given replica ids. + + Returns (rows, windows). rows is a list of + (object_label, kind, flavor, wallclock_s, time_ns_or_empty). windows maps + flavor -> (started_at_db, ended_at_db). + + Wall-clock completion is when mz_hydration_statuses first reports each + (object, replica) hydrated. This is honest here because the ticking source + keeps the write frontier live. time_ns is read afterwards for index objects. + """ + ids_in = ",".join(f"'{i}'" for i in target_ids) + objs_in = ",".join(f"'{o}'" for o in objs) + n_expected = len(objs) * len(target_ids) + + t0_db = mz.one("SELECT now();") + t0 = time.monotonic() + done = {} # (object_id, replica_id) -> wallclock seconds + deadline = t0 + timeout + while time.monotonic() < deadline: + rows = mz( + f"SELECT object_id, replica_id FROM mz_internal.mz_hydration_statuses " + f"WHERE object_id IN ({objs_in}) AND replica_id IN ({ids_in}) AND hydrated;" + ) + now = time.monotonic() + for row in rows: + oid, rid = row.split("|") + if (oid, rid) not in done: + done[(oid, rid)] = round(now - t0, 2) + print(f" hydrated {len(done)}/{n_expected}", flush=True) + if len(done) >= n_expected: + break + time.sleep(POLL_INTERVAL) + else: + print(f" WARNING: timed out, {n_expected - len(done)} pairs missing") + + t1_db = mz.one("SELECT now();") + + # Precise internal time for index exports (NULL for MVs, CLU-175). + time_ns = {} + for row in mz( + f"SELECT object_id, replica_id, time_ns " + f"FROM mz_internal.mz_compute_hydration_times " + f"WHERE object_id IN ({objs_in}) AND replica_id IN ({ids_in}) " + f"AND time_ns IS NOT NULL;" + ): + oid, rid, tns = row.split("|") + time_ns[(oid, rid)] = tns + + rows = [] + for (oid, rid), secs in done.items(): + label, kind = objs[oid] + rows.append((label, kind, target_ids[rid], secs, time_ns.get((oid, rid), ""))) + windows = {flavor: (t0_db, t1_db) for flavor in target_ids.values()} + return rows, windows + + +def run_scenario(mz, sf, scenario, trial, size, bodies, timeout, verify_flags=False): + print(f" [sf{sf}] {scenario} trial {trial}") + if scenario == "initial": + drop_workload(mz, sf) + drop_replicas(mz, sf) + add_replicas(mz, sf, size) + if verify_flags: + flags = check_flags(mz, sf) + print(f" pager overrides: {flags if flags else '(none; inherit env-wide)'}") + ids = {rid: name for name, rid in replica_ids(mz, sf).items()} + create_workload(mz, sf, bodies) + else: # rehydrate: workload stays live, replicas churn. + if not workload_exists(mz, sf): + add_replicas(mz, sf, size) + create_workload(mz, sf, bodies) + measure(mz, sf, object_map(mz, sf), + {rid: n for n, rid in replica_ids(mz, sf).items()}, timeout) + drop_replicas(mz, sf) + add_replicas(mz, sf, size) + ids = {rid: name for name, rid in replica_ids(mz, sf).items()} + + objs = object_map(mz, sf) + rows, windows = measure(mz, sf, objs, ids, timeout) + result_rows, window_rows = [], [] + for (label, kind, flavor, secs, tns) in sorted(rows): + result_rows.append([sf, scenario, flavor, label, kind, trial, secs, tns]) + for flavor, (start, end) in windows.items(): + rid = next(r for r, f in ids.items() if f == flavor) + window_rows.append([sf, scenario, flavor, trial, rid, start, end]) + print(f" recorded {len(result_rows)} rows " + f"({sum(1 for r in result_rows if r[4]=='index')} index, " + f"{sum(1 for r in result_rows if r[4]=='mv')} mv)") + return result_rows, window_rows + + +# --- Main ----------------------------------------------------------------- + +def main(): + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--mz-bin", default="bin/mz") + ap.add_argument("--config", default=os.environ.get("MZ_CONFIG", "")) + ap.add_argument("--region", default="aws/us-east-1") + ap.add_argument("--profile", default="staging") + ap.add_argument("--scale-factors", default=",".join(map(str, DEFAULT_SCALE_FACTORS))) + ap.add_argument("--trials", type=int, default=DEFAULT_TRIALS) + ap.add_argument("--size", default=DEFAULT_SIZE) + ap.add_argument("--source-size", default=DEFAULT_SOURCE_SIZE) + ap.add_argument("--tick", default=DEFAULT_TICK) + ap.add_argument("--outdir", default="pager-hydration-results") + ap.add_argument("--hydrate-timeout", type=int, default=5400) + ap.add_argument("--keep", action="store_true", help="skip test-cluster teardown") + ap.add_argument("--purge", action="store_true", help="also drop sources") + args = ap.parse_args() + + if not args.config: + print("ERROR: pass --config or set MZ_CONFIG (writable mz.toml path)") + sys.exit(1) + + mz = Mz(args.mz_bin, args.config, args.region, args.profile) + scale_factors = [int(x) for x in args.scale_factors.split(",")] + bodies = extract_bodies() + outdir = Path(args.outdir) + outdir.mkdir(parents=True, exist_ok=True) + + env_id = mz.one("SELECT mz_environment_id();") or "unknown" + namespace = "environment-" + env_id.removeprefix(args.region.replace("/", "-") + "-") + print(f"environment {env_id} (namespace {namespace})") + + results_path = outdir / "results.csv" + windows_path = outdir / "windows.csv" + done_cells = set() + if results_path.exists(): + with results_path.open() as f: + for row in csv.DictReader(f): + done_cells.add((int(row["sf"]), row["scenario"], int(row["trial"]))) + if done_cells: + print(f"resuming: {len(done_cells)} cells already done") + + rf = results_path.open("a", newline="") + wf = windows_path.open("a", newline="") + rw, ww = csv.writer(rf), csv.writer(wf) + if rf.tell() == 0: + rw.writerow(["sf", "scenario", "flavor", "object", "kind", "trial", + "hydration_seconds", "time_ns"]) + if wf.tell() == 0: + ww.writerow(["sf", "scenario", "flavor", "trial", "replica_id", + "started_at", "ended_at"]) + rf.flush() + wf.flush() + + for sf in scale_factors: + print(f"=== scale factor {sf} ===") + ensure_source(mz, sf, args.source_size, args.tick) + ensure_test_cluster(mz, sf) + first = True + for scenario in SCENARIOS: + for trial in range(1, args.trials + 1): + if (sf, scenario, trial) in done_cells: + print(f" [sf{sf}] {scenario} trial {trial} done, skipping") + continue + result_rows, window_rows = run_scenario( + mz, sf, scenario, trial, args.size, bodies, + args.hydrate_timeout, verify_flags=first, + ) + rw.writerows(result_rows) + ww.writerows(window_rows) + rf.flush() + wf.flush() + first = False + if not args.keep: + drop_replicas(mz, sf) + drop_workload(mz, sf) + mz(f"DROP CLUSTER IF EXISTS {test_cluster(sf)} CASCADE;") + print(f" [sf{sf}] test cluster torn down (source kept)") + if args.purge: + mz(f"DROP CLUSTER IF EXISTS {src_cluster(sf)} CASCADE;") + mz(f"DROP SCHEMA IF EXISTS {schema(sf)} CASCADE;") + print(f" [sf{sf}] source purged") + rf.close() + wf.close() + print(f"\nresults -> {results_path}\nwindows -> {windows_path}") + print(f"grafana namespace: {namespace}") + print("DONE") + + +if __name__ == "__main__": + main() diff --git a/misc/experiments/pager-hydration/run_local.py b/misc/experiments/pager-hydration/run_local.py new file mode 100644 index 0000000000000..38deb7d0bd1eb --- /dev/null +++ b/misc/experiments/pager-hydration/run_local.py @@ -0,0 +1,462 @@ +#!/usr/bin/env python3 +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +"""Local pager-hydration repro. + +Adapted from run.py to run against a local `bin/environmentd`, for fast +iteration on the slow-hydration / serial-fetch investigation without looping +through CI or a staging region. + +Differences from run.py, forced by running locally: + +* No LaunchDarkly. Scoped per-replica system parameters are written only by the + config sync loop from the delivery service (see + doc/developer/design/20260609_scoped_feature_flags.md), so the cloud approach of two + replicas differing only by a scoped `enable_column_paged_batcher` override is + not reproducible locally. Instead the pager on/off axis is a GLOBAL + `ALTER SYSTEM SET`, applied sequentially: set the flag, add one replica, let + it hydrate, measure, drop. Only one flag value is ever live at a time. +* Connections use psql. Application DDL/queries go to the external SQL port + (default 6875, user `materialize`). `ALTER SYSTEM` goes to the internal SQL + port (default 6877, user `mz_system`), because that is the superuser surface + locally. +* Cluster sizes are local `scale=1,workers=N` strings (see + src/catalog/src/config.rs), small by default so a laptop can run them. +* Scale factors default to 1 and 3. + +The pager on/off comparison is measured rehydrate-style: the workload (base-table +indexes, indexed views, materialized views for five heavy TPCH queries) is +created once per scale factor and left live, and only the single replica churns +under each flag value. The source uses a TICK INTERVAL so its write frontier +never goes final, otherwise mz_hydration_statuses reports a freshly added +replica hydrated before it has loaded anything. +""" + +import argparse +import csv +import os +import re +import subprocess +import sys +import time +from pathlib import Path + +# --- Configuration -------------------------------------------------------- + +DEFAULT_SIZE = "scale=1,workers=8" # small: 8 timely workers, whole process. +DEFAULT_SOURCE_SIZE = "scale=1,workers=2" +DEFAULT_SCALE_FACTORS = [1, 3] +DEFAULT_TRIALS = 3 +DEFAULT_TICK = "1s" +QUERIES = ["Q03", "Q05", "Q09", "Q18", "Q21"] +# (label, enable_column_paged_batcher). pager_on = paged batcher + spill; +# pager_off = legacy columnation batcher (no pager). +PAGER_CONDITIONS = [("pager_on", True), ("pager_off", False)] +SLT = "test/sqllogictest/tpch_create_materialized_view.slt" + +BASE_INDEXES = [ + ("pk_nation", "nation", "n_nationkey"), + ("fk_nation_region", "nation", "n_regionkey"), + ("pk_region", "region", "r_regionkey"), + ("pk_part", "part", "p_partkey"), + ("pk_supplier", "supplier", "s_suppkey"), + ("fk_supplier_nation", "supplier", "s_nationkey"), + ("pk_partsupp", "partsupp", "ps_partkey, ps_suppkey"), + ("fk_partsupp_part", "partsupp", "ps_partkey"), + ("fk_partsupp_supp", "partsupp", "ps_suppkey"), + ("pk_customer", "customer", "c_custkey"), + ("fk_customer_nation", "customer", "c_nationkey"), + ("pk_orders", "orders", "o_orderkey"), + ("fk_orders_cust", "orders", "o_custkey"), + ("pk_lineitem", "lineitem", "l_orderkey, l_linenumber"), + ("fk_lineitem_order", "lineitem", "l_orderkey"), + ("fk_lineitem_part", "lineitem", "l_partkey"), + ("fk_lineitem_supp", "lineitem", "l_suppkey"), + ("fk_lineitem_partsupp", "lineitem", "l_partkey, l_suppkey"), +] + +POLL_INTERVAL = 3.0 +INGEST_POLL_INTERVAL = 15.0 + +# Pager knobs set alongside the batcher flag so pager_on exercises the full +# spill path (matches the environment-wide staging config). +PAGER_ON_FLAGS = { + "enable_column_paged_batcher": "true", + "enable_column_paged_batcher_spill": "true", + "column_paged_batcher_lz4": "true", +} +PAGER_OFF_FLAGS = { + "enable_column_paged_batcher": "false", +} + + +class Psql: + """Runs `psql -tAq -c` against one (host, port, user).""" + + def __init__(self, host, port, user, database): + self.base = [ + "psql", "-tAqX", "-h", host, "-p", str(port), + "-U", user, "-d", database, "-c", + ] + + def __call__(self, stmt, timeout=900, search_path=None): + env = os.environ.copy() + # No password locally (trust/dev). Keep libpq from prompting. + env.setdefault("PGPASSWORD", "") + if search_path is not None: + # search_path at connection startup lets one DDL use unqualified + # names. A `SET ...; CREATE ...` prelude would form a multi-statement + # transaction, which DDL cannot enter. + env["PGOPTIONS"] = f"-c search_path={search_path}" + last = None + for _attempt in range(3): + try: + r = subprocess.run( + self.base + [stmt], capture_output=True, text=True, + timeout=timeout, env=env, + ) + except subprocess.TimeoutExpired: + last = f"timeout after {timeout}s" + time.sleep(5) + continue + if r.returncode == 0: + return [ln for ln in r.stdout.splitlines() if ln.strip()] + last = r.stderr or r.stdout + time.sleep(5) + raise RuntimeError(f"SQL failed after retries: {stmt}\n{last}") + + def one(self, stmt, timeout=900, search_path=None): + rows = self(stmt, timeout=timeout, search_path=search_path) + return rows[0] if rows else None + + +# --- Names ---------------------------------------------------------------- + +def schema(sf): + return f"sf{sf}" + + +def src_cluster(sf): + return f"ldgen_sf{sf}" + + +def test_cluster(sf): + return f"test_sf{sf}" + + +def extract_bodies(): + """Return {Qxx: SELECT body} for the target queries from the canonical slt.""" + text = Path(SLT).read_text() + bodies = {} + for q in QUERIES: + m = re.search(rf"CREATE MATERIALIZED VIEW {q} AS\n(.*?);\n", text, re.S) + if not m: + raise RuntimeError(f"missing {q} in {SLT}") + bodies[q] = m.group(1).strip() + return bodies + + +# --- Pager flag (global, sequential) -------------------------------------- + +def set_pager(system, on): + """Set the global pager flags for the next replica's hydration. + + Replica-local dyncfgs are pushed to running replicas live, so this must be + set before adding the replica whose hydration we intend to measure, and only + one value can be live across the environment at a time. + """ + flags = PAGER_ON_FLAGS if on else PAGER_OFF_FLAGS + for name, value in flags.items(): + system(f"ALTER SYSTEM SET {name} = {value};") + # Reset the flags not present in the OFF set back to default so a prior ON + # run does not leak spill/lz4 into an OFF run. + if not on: + for name in ("enable_column_paged_batcher_spill", "column_paged_batcher_lz4"): + system(f"ALTER SYSTEM RESET {name};") + + +# --- Source --------------------------------------------------------------- + +def source_status(mz, sf): + return mz.one( + f"SELECT status FROM mz_internal.mz_source_statuses s " + f"JOIN mz_sources so ON s.id = so.id " + f"JOIN mz_schemas sc ON so.schema_id = sc.id " + f"WHERE sc.name = '{schema(sf)}' AND so.name = 'ldgen';" + ) + + +def ensure_source(mz, sf, source_size, tick): + """Create the ticking TPCH source for `sf`, or reuse a running one.""" + sch, cl = schema(sf), src_cluster(sf) + st = source_status(mz, sf) + if st == "running": + print(f" [sf{sf}] source running, reusing") + return + if st is None: + print(f" [sf{sf}] creating source cluster + TPCH SF{sf} (tick {tick})") + if not mz.one(f"SELECT 1 FROM mz_schemas WHERE name = '{sch}';"): + mz(f"CREATE SCHEMA {sch};") + if not mz.one(f"SELECT 1 FROM mz_clusters WHERE name = '{cl}';"): + mz(f"CREATE CLUSTER {cl} SIZE '{source_size}';") + mz( + f"CREATE SOURCE {sch}.ldgen IN CLUSTER {cl} " + f"FROM LOAD GENERATOR TPCH (SCALE FACTOR {sf}, TICK INTERVAL '{tick}') " + f"FOR ALL TABLES;" + ) + else: + print(f" [sf{sf}] source present but status={st}; waiting") + # A load-generator source reports `running` only after its snapshot is + # committed. Do not probe with count(*): that scan can exceed the timeout. + while source_status(mz, sf) != "running": + print(f" [sf{sf}] ingest status={source_status(mz, sf)}") + time.sleep(INGEST_POLL_INTERVAL) + + +# --- Test cluster + workload ---------------------------------------------- + +def ensure_test_cluster(mz, sf): + cl = test_cluster(sf) + if not mz.one(f"SELECT 1 FROM mz_clusters WHERE name = '{cl}';"): + mz(f"CREATE CLUSTER {cl} REPLICAS ();") + print(f" [sf{sf}] created unmanaged cluster {cl}") + + +def add_replica(mz, sf, name, size): + mz(f"CREATE CLUSTER REPLICA {test_cluster(sf)}.{name} SIZE '{size}';") + + +def drop_replica(mz, sf, name): + mz(f"DROP CLUSTER REPLICA IF EXISTS {test_cluster(sf)}.{name};") + + +def replica_id(mz, sf, name): + return mz.one( + f"SELECT r.id FROM mz_cluster_replicas r " + f"JOIN mz_clusters c ON r.cluster_id = c.id " + f"WHERE c.name = '{test_cluster(sf)}' AND r.name = '{name}';" + ) + + +def workload_exists(mz, sf): + return bool(mz.one( + f"SELECT 1 FROM mz_materialized_views mv " + f"JOIN mz_schemas sc ON mv.schema_id = sc.id " + f"WHERE sc.name = '{schema(sf)}' AND mv.name = 'mq03';" + )) + + +def create_workload(mz, sf, bodies): + """Base-table indexes, indexed views, and MVs for the five queries.""" + sch, cl = schema(sf), test_cluster(sf) + for name, table, cols in BASE_INDEXES: + mz(f"CREATE INDEX {name} IN CLUSTER {cl} ON {table} ({cols});", search_path=sch) + for q in QUERIES: + v = f"v{q.lower()}" + mz(f"CREATE VIEW {sch}.{v} AS {bodies[q]};", search_path=sch) + mz(f"CREATE DEFAULT INDEX IN CLUSTER {cl} ON {sch}.{v};", search_path=sch) + mz( + f"CREATE MATERIALIZED VIEW {sch}.m{q.lower()} IN CLUSTER {cl} " + f"AS {bodies[q]};", + search_path=sch, + ) + + +def object_map(mz, sf): + """Return {object_id: (label, kind)} for all indexes and MVs on the cluster.""" + cl, sch = test_cluster(sf), schema(sf) + out = {} + for row in mz( + f"SELECT i.id, i.name FROM mz_indexes i " + f"JOIN mz_clusters c ON i.cluster_id = c.id " + f"JOIN mz_relations r ON i.on_id = r.id " + f"JOIN mz_schemas s ON r.schema_id = s.id " + f"WHERE c.name = '{cl}' AND s.name = '{sch}';" + ): + oid, name = row.split("|") + out[oid] = (name, "index") + for row in mz( + f"SELECT mv.id, mv.name FROM mz_materialized_views mv " + f"JOIN mz_schemas s ON mv.schema_id = s.id WHERE s.name = '{sch}';" + ): + oid, name = row.split("|") + out[oid] = (name, "mv") + return out + + +# --- Measurement ---------------------------------------------------------- + +def measure(mz, objs, rid, flavor, timeout): + """Time hydration of `objs` on a single replica id. + + Returns rows of (object_label, kind, flavor, wallclock_s, time_ns_or_empty). + Wall-clock completion is when mz_hydration_statuses first reports each object + hydrated on the replica. time_ns is read afterwards for index objects + (mz_compute_hydration_times; NULL for MVs, CLU-175). + """ + objs_in = ",".join(f"'{o}'" for o in objs) + n_expected = len(objs) + + t0 = time.monotonic() + done = {} + deadline = t0 + timeout + while time.monotonic() < deadline: + rows = mz( + f"SELECT object_id FROM mz_internal.mz_hydration_statuses " + f"WHERE object_id IN ({objs_in}) AND replica_id = '{rid}' AND hydrated;" + ) + now = time.monotonic() + for oid in rows: + if oid not in done: + done[oid] = round(now - t0, 2) + print(f" [{flavor}] hydrated {len(done)}/{n_expected}", flush=True) + if len(done) >= n_expected: + break + time.sleep(POLL_INTERVAL) + else: + print(f" [{flavor}] WARNING: timed out, {n_expected - len(done)} missing") + + time_ns = {} + for row in mz( + f"SELECT object_id, time_ns " + f"FROM mz_internal.mz_compute_hydration_times " + f"WHERE object_id IN ({objs_in}) AND replica_id = '{rid}' " + f"AND time_ns IS NOT NULL;" + ): + oid, tns = row.split("|") + time_ns[oid] = tns + + rows = [] + for oid, secs in done.items(): + label, kind = objs[oid] + rows.append((label, kind, flavor, secs, time_ns.get(oid, ""))) + return rows + + +def run_cell(app, system, sf, flavor, on, trial, size, timeout): + """Set the pager flag, add a fresh replica, time its hydration, drop it.""" + print(f" [sf{sf}] {flavor} trial {trial}") + name = f"{flavor}_t{trial}" + drop_replica(app, sf, name) + set_pager(system, on) + add_replica(app, sf, name, size) + rid = replica_id(app, sf, name) + objs = object_map(app, sf) + rows = measure(app, objs, rid, flavor, timeout) + drop_replica(app, sf, name) + out = [[sf, flavor, label, kind, trial, secs, tns] + for (label, kind, _flavor, secs, tns) in sorted(rows)] + print(f" recorded {len(out)} rows " + f"({sum(1 for r in out if r[3]=='index')} index, " + f"{sum(1 for r in out if r[3]=='mv')} mv)") + return out + + +# --- Main ----------------------------------------------------------------- + +def main(): + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--host", default="localhost") + ap.add_argument("--app-port", type=int, default=6875) + ap.add_argument("--app-user", default="materialize") + ap.add_argument("--system-port", type=int, default=6877, + help="internal SQL port for ALTER SYSTEM (mz_system)") + ap.add_argument("--system-user", default="mz_system") + ap.add_argument("--database", default="materialize") + ap.add_argument("--scale-factors", default=",".join(map(str, DEFAULT_SCALE_FACTORS))) + ap.add_argument("--trials", type=int, default=DEFAULT_TRIALS) + ap.add_argument("--size", default=DEFAULT_SIZE) + ap.add_argument("--source-size", default=DEFAULT_SOURCE_SIZE) + ap.add_argument("--tick", default=DEFAULT_TICK) + ap.add_argument("--outdir", default="pager-hydration-results-local") + ap.add_argument("--hydrate-timeout", type=int, default=1800) + ap.add_argument("--keep", action="store_true", help="skip test-cluster teardown") + ap.add_argument("--purge", action="store_true", help="also drop sources") + args = ap.parse_args() + + app = Psql(args.host, args.app_port, args.app_user, args.database) + system = Psql(args.host, args.system_port, args.system_user, args.database) + + # Fail fast with a clear message if environmentd is not up. + try: + app.one("SELECT 1;", timeout=15) + system.one("SELECT 1;", timeout=15) + except Exception as e: + print(f"ERROR: cannot reach local environmentd " + f"(app {args.host}:{args.app_port}, system {args.host}:{args.system_port}).\n" + f"Start it with `bin/environmentd` (CockroachDB must be running).\n{e}") + sys.exit(1) + + scale_factors = [int(x) for x in args.scale_factors.split(",")] + bodies = extract_bodies() + outdir = Path(args.outdir) + outdir.mkdir(parents=True, exist_ok=True) + + results_path = outdir / "results.csv" + done_cells = set() + if results_path.exists(): + with results_path.open() as f: + for row in csv.DictReader(f): + done_cells.add((int(row["sf"]), row["flavor"], int(row["trial"]))) + if done_cells: + print(f"resuming: {len(done_cells)} cells already done") + + rf = results_path.open("a", newline="") + rw = csv.writer(rf) + if rf.tell() == 0: + rw.writerow(["sf", "flavor", "object", "kind", "trial", + "hydration_seconds", "time_ns"]) + rf.flush() + + for sf in scale_factors: + print(f"=== scale factor {sf} ===") + ensure_source(app, sf, args.source_size, args.tick) + ensure_test_cluster(app, sf) + if not workload_exists(app, sf): + create_workload(app, sf, bodies) + print(f" [sf{sf}] workload created") + for flavor, on in PAGER_CONDITIONS: + for trial in range(1, args.trials + 1): + if (sf, flavor, trial) in done_cells: + print(f" [sf{sf}] {flavor} trial {trial} done, skipping") + continue + rows = run_cell(app, system, sf, flavor, on, trial, + args.size, args.hydrate_timeout) + rw.writerows(rows) + rf.flush() + if not args.keep: + drop_workload_and_cluster(app, sf) + print(f" [sf{sf}] test cluster torn down (source kept)") + if args.purge: + app(f"DROP CLUSTER IF EXISTS {src_cluster(sf)} CASCADE;") + app(f"DROP SCHEMA IF EXISTS {schema(sf)} CASCADE;") + print(f" [sf{sf}] source purged") + + # Leave the environment with the pager flag at its default. + system("ALTER SYSTEM RESET enable_column_paged_batcher;") + system("ALTER SYSTEM RESET enable_column_paged_batcher_spill;") + system("ALTER SYSTEM RESET column_paged_batcher_lz4;") + rf.close() + print(f"\nresults -> {results_path}") + print("DONE") + + +def drop_workload_and_cluster(mz, sf): + sch = schema(sf) + for q in QUERIES: + mz(f"DROP MATERIALIZED VIEW IF EXISTS {sch}.m{q.lower()};") + mz(f"DROP VIEW IF EXISTS {sch}.v{q.lower()} CASCADE;") + for name, _, _ in BASE_INDEXES: + mz(f"DROP INDEX IF EXISTS {sch}.{name};") + mz(f"DROP CLUSTER IF EXISTS {test_cluster(sf)} CASCADE;") + + +if __name__ == "__main__": + main()