extract the l0stack generator into a reusable python module

test_runner/deep_layers_with_delta.py

@@ -15,33 +15,28 @@ Measure impact of that.
 
 Some commands from shell history used in combination with this script:
 
- NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=futures-unordered ./target/release/neon_local pageserver restart
- NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=serial ./target/release/neon_local pageserver restart
- NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=parallel ./target/release/neon_local pageserver restart
-
-cat >> .neon/pageserver_1/pageserver.toml <<"EOF"
-# customizations
-virtual_file_io_mode = "direct"
-page_service_pipelining={mode="pipelined", max_batch_size=32, execution="concurrent-futures"}
+```
+ ./target/release/neon_local stop
+ rm -rf .neon
+ ./target/release/neon_local init
+ cat >> .neon/pageserver_1/pageserver.toml <<"EOF"
+ # customizations
+ virtual_file_io_mode = "direct"
+ page_service_pipelining={mode="pipelined", max_batch_size=32, execution="concurrent-futures"}
 EOF
+ ./target/release/neon_local start
 
-starting from scrach:
-    psql 'postgresql://localhost:1235/storage_controller' -c 'DELETE FROM tenant_shards'
-    NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=... ./target/release/neon_local pageserver restart
-    ./target/debug/neon_local endpoint stop foo
-    rm -rf .neon/endpoints/foo
-    ./target/debug/neon_local endpoint create foo
-    echo 'full_page_writes=off' >> .neon/endpoints/foo/postgresql.conf
-    ./target/debug/neon_local endpoint start foo
-    non-package-mode-py3.10christian@neon-hetzner-dev-christian:[~/src/neon/test_runner]: poetry run python3 deep_layers_with_delta.py
-    Need 6400 pages, 38400 rows
-    0
-    modified rows 6400
-    1
-    modified rows 6400
-    ...
-    19
-    modified rows 6400
+ psql 'postgresql://localhost:1235/storage_controller' -c 'DELETE FROM tenant_shards'
+ NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=futures-unordered ./target/release/neon_local pageserver restart
+ sleep 2
+ ./target/release/neon_local tenant create --set-default
+ ./target/debug/neon_local endpoint stop foo
+ rm -rf .neon/endpoints/foo
+ ./target/debug/neon_local endpoint create foo
+ echo 'full_page_writes=off' >> .neon/endpoints/foo/postgresql.conf
+ ./target/debug/neon_local endpoint start foo
+ pushd test_runner; poetry run python3 deep_layers_with_delta.py; popd
+```
 
 Layer files created:
 
@@ -77,82 +72,25 @@ Now use pagebench to measure perf, e.g.
 
 """
 
-
-from pathlib import Path
-import time
-from fixtures.common_types import Lsn, TenantId, TenantShardId, TimelineId
+import psycopg2
+from fixtures.common_types import TenantShardId, TimelineId
 from fixtures.pageserver.http import PageserverHttpClient
+from fixtures.pageserver.makelayers.l0stack import L0StackShape, make_l0_stack_standalone
 
 ps_http = PageserverHttpClient(port=9898, is_testing_enabled_or_skip=lambda: None)
 vps_http = PageserverHttpClient(port=1234, is_testing_enabled_or_skip=lambda: None)
 
 tenants = ps_http.tenant_list()
 assert len(tenants) == 1
-tenant_id = tenants[0]["id"]
+tenant_shard_id = TenantShardId.parse(tenants[0]["id"])
 
-timlines = ps_http.timeline_list(tenant_id)
+timlines = ps_http.timeline_list(tenant_shard_id)
 assert len(timlines) == 1
-timeline_id = timlines[0]["timeline_id"]
-
-config = {
-    "gc_period": "0s",  # disable periodic gc
-    "checkpoint_timeout": "10 years",
-    "compaction_period": "1h",  # doesn't matter, but 0 value will kill walredo every 10s
-    "compaction_threshold": 100000, # we just want L0s
-    "compaction_target_size": 134217728,
-    "checkpoint_distance": 268435456,
-    "image_creation_threshold": 100000, # we just want L0s
-}
-
-vps_http.set_tenant_config(tenant_id, config)
+timeline_id = TimelineId(timlines[0]["timeline_id"])
 
 connstr = "postgresql://cloud_admin@localhost:55432/postgres"
-
-import psycopg2
-
-def last_record_lsn(
-    pageserver_http_client: PageserverHttpClient,
-    tenant: TenantId | TenantShardId,
-    timeline: TimelineId,
-) -> Lsn:
-    detail = pageserver_http_client.timeline_detail(tenant, timeline)
-
-    lsn_str = detail["last_record_lsn"]
-    assert isinstance(lsn_str, str)
-    return Lsn(lsn_str)
-
 conn = psycopg2.connect(connstr)
-conn.autocommit = True
-cur = conn.cursor()
 
-# each tuple is 23 (header) + 100 bytes = 123 bytes
-# page header si 24 bytes
-# 8k page size
-# (8k-24bytes) / 123 bytes = 63 tuples per page
-# set fillfactor to 10 to have 6 tuples per page
-cur.execute("DROP TABLE IF EXISTS data")
-cur.execute("CREATE TABLE data(id bigint, row char(92)) with (fillfactor=10)")
-desired_size = 50 * 1024 * 1024  # 50MiB
-need_pages = desired_size // 8192
-need_rows = need_pages * 6
-print(f"Need {need_pages} pages, {need_rows} rows")
-cur.execute(f"INSERT INTO data SELECT i,'row'||i FROM generate_series(1, {need_rows}) as i")
-
-# every iteration updates one tuple in each page
-delta_stack_height = 20
-for i in range( 0, delta_stack_height):
-    print(i)
-    cur.execute(f"UPDATE data set row = row||',u' where id % 6 = {i%6}")
-    print("modified rows", cur.rowcount)
-    cur.execute("SELECT pg_current_wal_flush_lsn()")
-    flush_lsn = Lsn(cur.fetchall()[0][0])
-
-    while True:
-        last_record = last_record_lsn(ps_http, tenant_id, timeline_id)
-        if last_record >= flush_lsn:
-            break
-        time.sleep(0.1)
-
-    ps_http.timeline_checkpoint(tenant_id, timeline_id)
-    ps_http.timeline_compact(tenant_id, timeline_id)
+shape = L0StackShape(logical_table_size_mib=50, delta_stack_height=20)
 
+make_l0_stack_standalone(vps_http, ps_http, tenant_shard_id, timeline_id, conn, shape)

test_runner/fixtures/common_types.py

@@ -208,6 +208,10 @@ class ShardIndex:
             shard_count=int(input[2:4], 16),
         )
 
+    @property
+    def is_sharded(self) -> bool:
+        return self.shard_count != 0
+
 
 class TenantShardId:
     def __init__(self, tenant_id: TenantId, shard_number: int, shard_count: int):

test_runner/fixtures/neon_fixtures.py

@@ -313,6 +313,10 @@ class PgProtocol:
         """
         return self.safe_psql(query, log_query=log_query)[0][0]
 
+    def show_timeline_id(self) -> TimelineId:
+        """SHOW neon.timeline_id"""
+        return TimelineId(cast("str", self.safe_psql("show neon.timeline_id")[0][0]))
+
 
 class PageserverWalReceiverProtocol(StrEnum):
     VANILLA = "vanilla"

test_runner/fixtures/pageserver/makelayers/l0stack.py

@@ -0,0 +1,104 @@
+from dataclasses import dataclass
+
+from psycopg2.extensions import connection as PgConnection
+
+from fixtures.common_types import Lsn, ShardIndex, TenantShardId, TimelineId
+from fixtures.log_helper import log
+from fixtures.neon_fixtures import Endpoint
+from fixtures.pageserver.http import PageserverHttpClient
+from fixtures.pageserver.utils import wait_for_last_record_lsn
+
+
+@dataclass
+class L0StackShape:
+    logical_table_size_mib: int = 50
+    delta_stack_height: int = 20
+
+
+def make_l0_stack(endpoint: Endpoint, shape: L0StackShape):
+    env = endpoint.env
+
+    # TDOO: wait for storcon to finish any reonciles before jumping to action here?
+    shards = env.storage_controller.tenant_describe(endpoint.tenant_id)
+
+    shard_idx = ShardIndex.parse(shards[0]["shard_id"])
+    assert shard_idx.shard_count == 1, "see above"
+    tenant_shard_id = TenantShardId(
+        endpoint.tenant_id, shard_idx.shard_number, shard_idx.shard_count
+    )
+
+    ps = env.get_pageserver(shards[0]["node_id"])
+
+    timeline_id = endpoint.show_timeline_id()
+
+    vps_http = env.storage_controller.pageserver_api()
+    ps_http = ps.http_client()
+    endpoint_conn = endpoint.connect()
+    make_l0_stack_standalone(vps_http, ps_http, tenant_shard_id, timeline_id, endpoint_conn, shape)
+
+
+def make_l0_stack_standalone(
+    vps_http: PageserverHttpClient,
+    ps_http: PageserverHttpClient,
+    tenant_shard_id: TenantShardId,
+    timeline_id: TimelineId,
+    endpoint_conn: PgConnection,
+    shape: L0StackShape,
+):
+    """
+    Creates stack of L0 deltas each of which should have 1 Value::Delta per page in `data`.
+
+    Usable from scripts (ad-hoc manual testing in neon_local) and pytest code alike.
+    """
+
+    assert (
+        not tenant_shard_id.shard_index.is_sharded
+    ), "the current implementation only supports unsharded tenants"
+
+    tenant_id = tenant_shard_id.tenant_id
+    conn = endpoint_conn
+    desired_size = shape.logical_table_size_mib * 1024 * 1024
+
+    config = {
+        "gc_period": "0s",  # disable periodic gc
+        "checkpoint_timeout": "10 years",
+        "compaction_period": "1h",  # doesn't matter, but 0 value will kill walredo every 10s
+        "compaction_threshold": 100000,  # we just want L0s
+        "compaction_target_size": 134217728,
+        "checkpoint_distance": 268435456,
+        "image_creation_threshold": 100000,  # we just want L0s
+    }
+
+    vps_http.set_tenant_config(tenant_id, config)
+
+    conn.autocommit = True
+    cur = conn.cursor()
+
+    # each tuple is 23 (header) + 100 bytes = 123 bytes
+    # page header si 24 bytes
+    # 8k page size
+    # (8k-24bytes) / 123 bytes = 63 tuples per page
+    # set fillfactor to 10 to have 6 tuples per page
+    cur.execute("DROP TABLE IF EXISTS data")
+    cur.execute("CREATE TABLE data(id bigint, row char(92)) with (fillfactor=10)")
+    need_pages = desired_size // 8192
+    need_rows = need_pages * 6
+    log.info(f"Need {need_pages} pages, {need_rows} rows")
+    cur.execute(f"INSERT INTO data SELECT i,'row'||i FROM generate_series(1, {need_rows}) as i")
+
+    def settle_and_flush():
+        cur.execute("SELECT pg_current_wal_flush_lsn()")
+        flush_lsn = Lsn(cur.fetchall()[0][0])
+        wait_for_last_record_lsn(ps_http, tenant_shard_id, timeline_id, flush_lsn)
+        ps_http.timeline_checkpoint(tenant_id, timeline_id)
+        ps_http.timeline_compact(tenant_id, timeline_id)
+
+    # create an L0 for the initial data we just inserted
+    settle_and_flush()
+    # every iteration updates one tuple in each page
+    delta_stack_height = shape.delta_stack_height
+    for i in range(0, delta_stack_height):
+        log.info(i)
+        cur.execute(f"UPDATE data set row = row||',u' where id % 6 = {i%6}")
+        log.info("modified rows", cur.rowcount)
+        settle_and_flush()

test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py

@@ -1,11 +1,11 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
 import dataclasses
 import json
-from pathlib import Path
 import sys
 import time
+from dataclasses import dataclass
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 import pytest
@@ -238,7 +238,6 @@ def setup_tenant_template(env: NeonEnv, pg_bin: PgBin, scale: int):
     return (template_tenant, template_timeline, config)
 
 
-
 def run_pagebench_benchmark(
     env: NeonEnv, pg_bin: PgBin, record, duration_secs: int, n_clients: int
 ):
@@ -254,7 +253,7 @@ def run_pagebench_benchmark(
         pageserver_cpu_seconds_total: float
         pageserver_layers_visited_per_vectored_read_global_buckets: dict[int, int]
 
-        def __sub__(self, other: "Metrics") -> "Metrics":
+        def __sub__(self, other: Metrics) -> Metrics:
             return Metrics(
                 time=self.time - other.time,
                 pageserver_cpu_seconds_total=self.pageserver_cpu_seconds_total
@@ -265,7 +264,7 @@ def run_pagebench_benchmark(
                 },
             )
 
-        def normalize(self, by) -> "Metrics":
+        def normalize(self, by) -> Metrics:
             return Metrics(
                 time=self.time / by,
                 pageserver_cpu_seconds_total=self.pageserver_cpu_seconds_total / by,
@@ -277,6 +276,7 @@ def run_pagebench_benchmark(
 
     def get_metrics() -> Metrics:
         pageserver_metrics = ps_http.get_metrics()
+
         def parse_le_label(s):
             try:
                 return int(s)
@@ -285,6 +285,7 @@ def run_pagebench_benchmark(
                     return sys.maxsize
                 else:
                     raise
+
         return Metrics(
             time=time.time(),
             pageserver_cpu_seconds_total=pageserver_metrics.query_one(
@@ -328,7 +329,6 @@ def run_pagebench_benchmark(
 
     metrics = metrics_after - metrics_before
 
-
     #
     # Record the results
     #
@@ -368,4 +368,6 @@ def run_pagebench_benchmark(
                     report=MetricReport.TEST_PARAM,
                 )
         else:
-            record(f"counters.{metric}", metric_value=value, unit="", report=MetricReport.TEST_PARAM)
+            record(
+                f"counters.{metric}", metric_value=value, unit="", report=MetricReport.TEST_PARAM
+            )