diff --git a/test_runner/regress/test_wal_acceptor.py b/test_runner/regress/test_wal_acceptor.py index 98f0ad91b9..a35086fb7e 100644 --- a/test_runner/regress/test_wal_acceptor.py +++ b/test_runner/regress/test_wal_acceptor.py @@ -1642,6 +1642,108 @@ def test_safekeeper_without_pageserver_and_pg_restore( log.info(f"Restore rate: {restore_rate:.2f} bytes/second") + +def test_safekeeper_without_pageserver_and_waltest( + test_output_dir: str, + port_distributor: PortDistributor, + pg_bin: PgBin, + neon_binpath: Path, +): + # Create the environment in the test-specific output dir + repo_dir = Path(os.path.join(test_output_dir, "repo")) + + pg_conf_options: dict[str, str] = { + "shared_buffers": "8GB", + } + + env = SafekeeperEnv( + repo_dir, + port_distributor, + pg_bin, + neon_binpath, + pg_conf_options=pg_conf_options, + ) + + wal_test = """ +CREATE OR REPLACE FUNCTION public.waltest() + RETURNS text + LANGUAGE plpgsql +AS $function$ +declare + i int; + j int; + lastts timestamp; + nowts timestamp; + lag bigint; + bandwidth numeric; + lastlsn pg_lsn; + nowlsn pg_lsn; + + moving_avg numeric; + result text := ''; -- Initialize an empty string to capture messages +begin + moving_avg = 0; + for i in 1..1000 loop -- 1 GiB of WAL + loop + nowlsn = pg_current_wal_insert_lsn(); + lag = (select nowlsn - received_lsn from backpressure_lsns()); + if lag > 200_000_000 then + exit; + end if; + + -- emit 1 MB of WAL + for j in 1..100 loop + perform pg_logical_emit_message(false, '', repeat('x', 10486)); + end loop; + end loop; + nowts = clock_timestamp(); + + if lastlsn is not null then + bandwidth = (nowlsn - lastlsn) / (extract(epoch from nowts) - extract(epoch from lastts)); + + moving_avg = moving_avg + bandwidth / 10 - (moving_avg / 10); + + -- Capture the message instead of raising a notice + result := result || format('bandwidth: %s kB / s (avg %s kB / s)%s', + lpad(round(bandwidth / 1024)::text, 10), + lpad(round(moving_avg / 1024)::text, 10), + chr(10) -- Newline for formatting + ); + end if; + lastlsn = nowlsn; + lastts = nowts; + perform pg_sleep(0.1); + end loop; + return result; -- Return the concatenated string +end; +$function$ +; + """ + + with env: + env.init() + assert env.postgres is not None + env.postgres.connstr() + shared_buffers = env.postgres.safe_psql("show shared_buffers")[0][0] + log.info(f"shared_buffers: {shared_buffers}") + size_before = env.postgres.safe_psql("select pg_database_size('postgres');")[0][0] + log.info(f"Database size before test: {size_before}") + env.postgres.safe_psql("create extension neon;") + env.postgres.safe_psql(wal_test) + env.postgres.safe_psql("SET statement_timeout=0;") + start_time = time.time() + output = env.postgres.safe_psql("select waltest();")[0][0] + end_time = time.time() + duration = end_time - start_time + log.info(output) + size_after = env.postgres.safe_psql("select pg_database_size('postgres');")[0][0] + log.info(f"Database size after test: {size_after}") + # Calculate the restore rate in bytes/second + restored_size = size_after - size_before + restore_rate = restored_size / duration + log.info(f"test duration: {duration:.2f} seconds") + log.info(f"Average ingest rate: {restore_rate:.2f} bytes/second") + def test_safekeeper_without_pageserver( test_output_dir: str, port_distributor: PortDistributor,