Merge remote-tracking branch 'origin/main' into layer_compression-heikki2

test_runner/batch_others/test_ancestor_branch.py

@@ -1,11 +1,9 @@
-import subprocess
-import asyncio
 from contextlib import closing
 
 import psycopg2.extras
 import pytest
 from fixtures.log_helper import log
-from fixtures.zenith_fixtures import ZenithEnvBuilder
+from fixtures.zenith_fixtures import ZenithEnv, ZenithEnvBuilder, ZenithPageserverApiException
 
 
 #
@@ -33,6 +31,10 @@ def test_ancestor_branch(zenith_env_builder: ZenithEnvBuilder):
             'compaction_target_size': '4194304',
         })
 
+    with closing(env.pageserver.connect()) as psconn:
+        with psconn.cursor(cursor_factory=psycopg2.extras.DictCursor) as pscur:
+            pscur.execute("failpoints flush-frozen=sleep(10000)")
+
     env.zenith_cli.create_timeline(f'main', tenant_id=tenant)
     pg_branch0 = env.postgres.create_start('main', tenant_id=tenant)
     branch0_cur = pg_branch0.connect().cursor()
@@ -116,3 +118,17 @@ def test_ancestor_branch(zenith_env_builder: ZenithEnvBuilder):
 
     branch2_cur.execute('SELECT count(*) FROM foo')
     assert branch2_cur.fetchone() == (300000, )
+
+
+def test_ancestor_branch_detach(zenith_simple_env: ZenithEnv):
+    env = zenith_simple_env
+
+    parent_timeline_id = env.zenith_cli.create_branch("test_ancestor_branch_detach_parent", "empty")
+
+    env.zenith_cli.create_branch("test_ancestor_branch_detach_branch1",
+                                 "test_ancestor_branch_detach_parent")
+
+    ps_http = env.pageserver.http_client()
+    with pytest.raises(ZenithPageserverApiException,
+                       match="Failed to detach inmem tenant timeline"):
+        ps_http.timeline_detach(env.initial_tenant, parent_timeline_id)

test_runner/batch_others/test_createdropdb.py

@@ -32,7 +32,16 @@ def test_createdb(zenith_simple_env: ZenithEnv):
 
     # Test that you can connect to the new database on both branches
     for db in (pg, pg2):
-        db.connect(dbname='foodb').close()
+        with closing(db.connect(dbname='foodb')) as conn:
+            with conn.cursor() as cur:
+                # Check database size in both branches
+                cur.execute(
+                    'select pg_size_pretty(pg_database_size(%s)), pg_size_pretty(sum(pg_relation_size(oid))) from pg_class where relisshared is false;',
+                    ('foodb', ))
+                res = cur.fetchone()
+                # check that dbsize equals sum of all relation sizes, excluding shared ones
+                # This is how we define dbsize in zenith for now
+                assert res[0] == res[1]
 
 
 #

test_runner/batch_others/test_lsn_mapping.py

@@ -0,0 +1,84 @@
+from contextlib import closing
+from datetime import timedelta, timezone, tzinfo
+import math
+from uuid import UUID
+import psycopg2.extras
+import psycopg2.errors
+from fixtures.zenith_fixtures import ZenithEnv, ZenithEnvBuilder, Postgres
+from fixtures.log_helper import log
+import time
+
+
+#
+# Test pageserver get_lsn_by_timestamp API
+#
+def test_lsn_mapping(zenith_env_builder: ZenithEnvBuilder):
+    zenith_env_builder.num_safekeepers = 1
+    env = zenith_env_builder.init_start()
+
+    new_timeline_id = env.zenith_cli.create_branch('test_lsn_mapping')
+    pgmain = env.postgres.create_start("test_lsn_mapping")
+    log.info("postgres is running on 'test_lsn_mapping' branch")
+
+    ps_conn = env.pageserver.connect()
+    ps_cur = ps_conn.cursor()
+    conn = pgmain.connect()
+    cur = conn.cursor()
+
+    # Create table, and insert rows, each in a separate transaction
+    # Disable synchronous_commit to make this initialization go faster.
+    #
+    # Each row contains current insert LSN and the current timestamp, when
+    # the row was inserted.
+    cur.execute("SET synchronous_commit=off")
+    cur.execute("CREATE TABLE foo (x integer)")
+    tbl = []
+    for i in range(1000):
+        cur.execute(f"INSERT INTO foo VALUES({i})")
+        cur.execute(f'SELECT clock_timestamp()')
+        # Get the timestamp at UTC
+        after_timestamp = cur.fetchone()[0].replace(tzinfo=None)
+        tbl.append([i, after_timestamp])
+
+    # Execute one more transaction with synchronous_commit enabled, to flush
+    # all the previous transactions
+    cur.execute("SET synchronous_commit=on")
+    cur.execute("INSERT INTO foo VALUES (-1)")
+
+    # Check edge cases: timestamp in the future
+    probe_timestamp = tbl[-1][1] + timedelta(hours=1)
+    ps_cur.execute(
+        f"get_lsn_by_timestamp {env.initial_tenant.hex} {new_timeline_id.hex} '{probe_timestamp.isoformat()}Z'"
+    )
+    result = ps_cur.fetchone()[0]
+    assert result == 'future'
+
+    # timestamp too the far history
+    probe_timestamp = tbl[0][1] - timedelta(hours=10)
+    ps_cur.execute(
+        f"get_lsn_by_timestamp {env.initial_tenant.hex} {new_timeline_id.hex} '{probe_timestamp.isoformat()}Z'"
+    )
+    result = ps_cur.fetchone()[0]
+    assert result == 'past'
+
+    # Probe a bunch of timestamps in the valid range
+    for i in range(1, len(tbl), 100):
+        probe_timestamp = tbl[i][1]
+
+        # Call get_lsn_by_timestamp to get the LSN
+        ps_cur.execute(
+            f"get_lsn_by_timestamp {env.initial_tenant.hex} {new_timeline_id.hex} '{probe_timestamp.isoformat()}Z'"
+        )
+        lsn = ps_cur.fetchone()[0]
+
+        # Launch a new read-only node at that LSN, and check that only the rows
+        # that were supposed to be committed at that point in time are visible.
+        pg_here = env.postgres.create_start(branch_name='test_lsn_mapping',
+                                            node_name='test_lsn_mapping_read',
+                                            lsn=lsn)
+        with closing(pg_here.connect()) as conn_here:
+            with conn_here.cursor() as cur_here:
+                cur_here.execute("SELECT max(x) FROM foo")
+                assert cur_here.fetchone()[0] == i
+
+        pg_here.stop_and_destroy()

test_runner/batch_others/test_recovery.py

@@ -0,0 +1,64 @@
+import os
+import time
+import psycopg2.extras
+import json
+from ast import Assert
+from contextlib import closing
+from fixtures.zenith_fixtures import ZenithEnvBuilder
+from fixtures.log_helper import log
+
+
+#
+# Test pageserver recovery after crash
+#
+def test_pageserver_recovery(zenith_env_builder: ZenithEnvBuilder):
+    zenith_env_builder.num_safekeepers = 1
+    # Override default checkpointer settings to run it more often
+    zenith_env_builder.pageserver_config_override = "tenant_config={checkpoint_distance = 1048576}"
+
+    env = zenith_env_builder.init()
+
+    # Check if failpoints enables. Otherwise the test doesn't make sense
+    f = env.zenith_cli.pageserver_enabled_features()
+
+    assert "failpoints" in f["features"], "Build pageserver with --features=failpoints option to run this test"
+    zenith_env_builder.start()
+
+    # Create a branch for us
+    env.zenith_cli.create_branch("test_pageserver_recovery", "main")
+
+    pg = env.postgres.create_start('test_pageserver_recovery')
+    log.info("postgres is running on 'test_pageserver_recovery' branch")
+
+    connstr = pg.connstr()
+
+    with closing(pg.connect()) as conn:
+        with conn.cursor() as cur:
+            with closing(env.pageserver.connect()) as psconn:
+                with psconn.cursor(cursor_factory=psycopg2.extras.DictCursor) as pscur:
+                    # Create and initialize test table
+                    cur.execute("CREATE TABLE foo(x bigint)")
+                    cur.execute("INSERT INTO foo VALUES (generate_series(1,100000))")
+
+                    # Sleep for some time to let checkpoint create image layers
+                    time.sleep(2)
+
+                    # Configure failpoints
+                    pscur.execute(
+                        "failpoints checkpoint-before-sync=sleep(2000);checkpoint-after-sync=panic")
+
+                    # Do some updates until pageserver is crashed
+                    try:
+                        while True:
+                            cur.execute("update foo set x=x+1")
+                    except Exception as err:
+                        log.info(f"Excepted server crash {err}")
+
+    log.info("Wait before server restart")
+    env.pageserver.stop()
+    env.pageserver.start()
+
+    with closing(pg.connect()) as conn:
+        with conn.cursor() as cur:
+            cur.execute("select count(*) from foo")
+            assert cur.fetchone() == (100000, )

test_runner/batch_others/test_remote_storage.py

@@ -117,7 +117,7 @@ def test_remote_storage_backup_and_restore(zenith_env_builder: ZenithEnvBuilder,
     detail = client.timeline_detail(UUID(tenant_id), UUID(timeline_id))
     assert detail['local'] is not None
     log.info("Timeline detail after attach completed: %s", detail)
-    assert lsn_from_hex(detail['local']['last_record_lsn']) == current_lsn
+    assert lsn_from_hex(detail['local']['last_record_lsn']) >= current_lsn, 'current db Lsn should shoud not be less than the one stored on remote storage'
     assert not detail['remote']['awaits_download']
 
     pg = env.postgres.create_start('main')

test_runner/batch_others/test_tenant_relocation.py

@@ -109,10 +109,11 @@ def test_tenant_relocation(zenith_env_builder: ZenithEnvBuilder,
 
     tenant = env.zenith_cli.create_tenant(UUID("74ee8b079a0e437eb0afea7d26a07209"))
     log.info("tenant to relocate %s", tenant)
-    env.zenith_cli.create_root_branch('main', tenant_id=tenant)
-    env.zenith_cli.create_branch('test_tenant_relocation', tenant_id=tenant)
 
-    tenant_pg = env.postgres.create_start(branch_name='main',
+    # attach does not download ancestor branches (should it?), just use root branch for now
+    env.zenith_cli.create_root_branch('test_tenant_relocation', tenant_id=tenant)
+
+    tenant_pg = env.postgres.create_start(branch_name='test_tenant_relocation',
                                           node_name='test_tenant_relocation',
                                           tenant_id=tenant)
 

test_runner/batch_others/test_wal_acceptor.py

@@ -573,7 +573,9 @@ def test_timeline_status(zenith_env_builder: ZenithEnvBuilder):
     timeline_id = pg.safe_psql("show zenith.zenith_timeline")[0][0]
 
     # fetch something sensible from status
-    epoch = wa_http_cli.timeline_status(tenant_id, timeline_id).acceptor_epoch
+    tli_status = wa_http_cli.timeline_status(tenant_id, timeline_id)
+    epoch = tli_status.acceptor_epoch
+    timeline_start_lsn = tli_status.timeline_start_lsn
 
     pg.safe_psql("create table t(i int)")
 
@@ -581,9 +583,13 @@ def test_timeline_status(zenith_env_builder: ZenithEnvBuilder):
     pg.stop().start()
     pg.safe_psql("insert into t values(10)")
 
-    epoch_after_reboot = wa_http_cli.timeline_status(tenant_id, timeline_id).acceptor_epoch
+    tli_status = wa_http_cli.timeline_status(tenant_id, timeline_id)
+    epoch_after_reboot = tli_status.acceptor_epoch
     assert epoch_after_reboot > epoch
 
+    # and timeline_start_lsn stays the same
+    assert tli_status.timeline_start_lsn == timeline_start_lsn
+
 
 class SafekeeperEnv:
     def __init__(self,

test_runner/fixtures/compare_fixtures.py

@@ -130,7 +130,10 @@ class VanillaCompare(PgCompare):
     def __init__(self, zenbenchmark, vanilla_pg: VanillaPostgres):
         self._pg = vanilla_pg
         self._zenbenchmark = zenbenchmark
-        vanilla_pg.configure(['shared_buffers=1MB'])
+        vanilla_pg.configure([
+            'shared_buffers=1MB',
+            'synchronous_commit=off',
+        ])
         vanilla_pg.start()
 
         # Long-lived cursor, useful for flushing

test_runner/fixtures/zenith_fixtures.py

@@ -472,20 +472,16 @@ class ZenithEnvBuilder:
 
         mock_endpoint = self.s3_mock_server.endpoint()
         mock_region = self.s3_mock_server.region()
-        mock_access_key = self.s3_mock_server.access_key()
-        mock_secret_key = self.s3_mock_server.secret_key()
         boto3.client(
             's3',
             endpoint_url=mock_endpoint,
             region_name=mock_region,
-            aws_access_key_id=mock_access_key,
-            aws_secret_access_key=mock_secret_key,
+            aws_access_key_id=self.s3_mock_server.access_key(),
+            aws_secret_access_key=self.s3_mock_server.secret_key(),
         ).create_bucket(Bucket=bucket_name)
         self.pageserver_remote_storage = S3Storage(bucket=bucket_name,
                                                    endpoint=mock_endpoint,
-                                                   region=mock_region,
-                                                   access_key=mock_access_key,
-                                                   secret_key=mock_secret_key)
+                                                   region=mock_region)
 
     def __enter__(self):
         return self
@@ -811,8 +807,6 @@ class LocalFsStorage:
 class S3Storage:
     bucket: str
     region: str
-    access_key: Optional[str]
-    secret_key: Optional[str]
     endpoint: Optional[str]
 
 
@@ -980,12 +974,32 @@ class ZenithCli:
             res.check_returncode()
             return res
 
+    def pageserver_enabled_features(self) -> Any:
+        bin_pageserver = os.path.join(str(zenith_binpath), 'pageserver')
+        args = [bin_pageserver, '--enabled-features']
+        log.info('Running command "{}"'.format(' '.join(args)))
+
+        res = subprocess.run(args,
+                             check=True,
+                             universal_newlines=True,
+                             stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE)
+        log.info(f"pageserver_enabled_features success: {res.stdout}")
+        return json.loads(res.stdout)
+
     def pageserver_start(self, overrides=()) -> 'subprocess.CompletedProcess[str]':
         start_args = ['pageserver', 'start', *overrides]
         append_pageserver_param_overrides(start_args,
                                           self.env.pageserver.remote_storage,
                                           self.env.pageserver.config_override)
-        return self.raw_cli(start_args)
+
+        s3_env_vars = None
+        if self.env.s3_mock_server:
+            s3_env_vars = {
+                'AWS_ACCESS_KEY_ID': self.env.s3_mock_server.access_key(),
+                'AWS_SECRET_ACCESS_KEY': self.env.s3_mock_server.secret_key(),
+            }
+        return self.raw_cli(start_args, extra_env_vars=s3_env_vars)
 
     def pageserver_stop(self, immediate=False) -> 'subprocess.CompletedProcess[str]':
         cmd = ['pageserver', 'stop']
@@ -1080,6 +1094,7 @@ class ZenithCli:
 
     def raw_cli(self,
                 arguments: List[str],
+                extra_env_vars: Optional[Dict[str, str]] = None,
                 check_return_code=True) -> 'subprocess.CompletedProcess[str]':
         """
         Run "zenith" with the specified arguments.
@@ -1095,7 +1110,7 @@ class ZenithCli:
 
         assert type(arguments) == list
 
-        bin_zenith = os.path.join(str(zenith_binpath), 'zenith')
+        bin_zenith = os.path.join(str(zenith_binpath), 'neon_local')
 
         args = [bin_zenith] + arguments
         log.info('Running command "{}"'.format(' '.join(args)))
@@ -1104,9 +1119,10 @@ class ZenithCli:
         env_vars = os.environ.copy()
         env_vars['ZENITH_REPO_DIR'] = str(self.env.repo_dir)
         env_vars['POSTGRES_DISTRIB_DIR'] = str(pg_distrib_dir)
-
         if self.env.rust_log_override is not None:
             env_vars['RUST_LOG'] = self.env.rust_log_override
+        for (extra_env_key, extra_env_value) in (extra_env_vars or {}).items():
+            env_vars[extra_env_key] = extra_env_value
 
         # Pass coverage settings
         var = 'LLVM_PROFILE_FILE'
@@ -1204,10 +1220,6 @@ def append_pageserver_param_overrides(
             pageserver_storage_override = f"bucket_name='{pageserver_remote_storage.bucket}',\
                 bucket_region='{pageserver_remote_storage.region}'"
 
-            if pageserver_remote_storage.access_key is not None:
-                pageserver_storage_override += f",access_key_id='{pageserver_remote_storage.access_key}'"
-            if pageserver_remote_storage.secret_key is not None:
-                pageserver_storage_override += f",secret_access_key='{pageserver_remote_storage.secret_key}'"
             if pageserver_remote_storage.endpoint is not None:
                 pageserver_storage_override += f",endpoint='{pageserver_remote_storage.endpoint}'"
 
@@ -1302,7 +1314,7 @@ class VanillaPostgres(PgProtocol):
         """Append lines into postgresql.conf file."""
         assert not self.running
         with open(os.path.join(self.pgdatadir, 'postgresql.conf'), 'a') as conf_file:
-            conf_file.writelines(options)
+            conf_file.write("\n".join(options))
 
     def start(self, log_path: Optional[str] = None):
         assert not self.running
@@ -1382,8 +1394,8 @@ def remote_pg(test_output_dir: str) -> Iterator[RemotePostgres]:
 class ZenithProxy(PgProtocol):
     def __init__(self, port: int):
         super().__init__(host="127.0.0.1",
-                         user="pytest",
-                         password="pytest",
+                         user="proxy_user",
+                         password="pytest2",
                          port=port,
                          dbname='postgres')
         self.http_port = 7001
@@ -1399,8 +1411,8 @@ class ZenithProxy(PgProtocol):
         args = [bin_proxy]
         args.extend(["--http", f"{self.host}:{self.http_port}"])
         args.extend(["--proxy", f"{self.host}:{self.port}"])
-        args.extend(["--auth-method", "password"])
-        args.extend(["--static-router", addr])
+        args.extend(["--auth-backend", "postgres"])
+        args.extend(["--auth-endpoint", "postgres://proxy_auth:pytest1@localhost:5432/postgres"])
         self._popen = subprocess.Popen(args)
         self._wait_until_ready()
 
@@ -1422,7 +1434,8 @@ class ZenithProxy(PgProtocol):
 def static_proxy(vanilla_pg) -> Iterator[ZenithProxy]:
     """Zenith proxy that routes directly to vanilla postgres."""
     vanilla_pg.start()
-    vanilla_pg.safe_psql("create user pytest with password 'pytest';")
+    vanilla_pg.safe_psql("create user proxy_auth with password 'pytest1' superuser")
+    vanilla_pg.safe_psql("create user proxy_user with password 'pytest2'")
 
     with ZenithProxy(4432) as proxy:
         proxy.start_static()
@@ -1571,6 +1584,7 @@ class Postgres(PgProtocol):
         assert self.node_name is not None
         self.env.zenith_cli.pg_stop(self.node_name, self.tenant_id, True)
         self.node_name = None
+        self.running = False
 
         return self
 
@@ -1747,6 +1761,7 @@ class SafekeeperTimelineStatus:
     acceptor_epoch: int
     flush_lsn: str
     remote_consistent_lsn: str
+    timeline_start_lsn: str
 
 
 @dataclass
@@ -1771,7 +1786,8 @@ class SafekeeperHttpClient(requests.Session):
         resj = res.json()
         return SafekeeperTimelineStatus(acceptor_epoch=resj['acceptor_state']['epoch'],
                                         flush_lsn=resj['flush_lsn'],
-                                        remote_consistent_lsn=resj['remote_consistent_lsn'])
+                                        remote_consistent_lsn=resj['remote_consistent_lsn'],
+                                        timeline_start_lsn=resj['timeline_start_lsn'])
 
     def record_safekeeper_info(self, tenant_id: str, timeline_id: str, body):
         res = self.post(

test_runner/performance/README.md

@@ -0,0 +1,23 @@
+# What performance tests do we have and how we run them
+
+Performance tests are built using the same infrastructure as our usual python integration tests. There are some extra fixtures that help to collect performance metrics, and to run tests against both vanilla PostgreSQL and Neon for comparison.
+
+## Tests that are run against local installation
+
+Most of the performance tests run against a local installation. This is not very representative of a production environment. Firstly, Postgres, safekeeper(s) and the pageserver have to share CPU and I/O resources, which can add noise to the results. Secondly, network overhead is eliminated.
+
+In the CI, the performance tests are run in the same environment as the other integration tests. We don't have control over the host that the CI runs on, so the environment may vary widely from one run to another, which makes the results across different runs noisy to compare.
+
+## Remote tests
+
+There are a few tests that marked with `pytest.mark.remote_cluster`. These tests do not set up a local environment, and instead require a libpq connection string to connect to. So they can be run on any Postgres compatible database. Currently, the CI runs these tests our staging environment daily. Staging is not an isolated environment, so there can be noise in the results due to activity of other clusters.
+
+## Noise
+
+All tests run only once. Usually to obtain more consistent performance numbers, a test should be repeated multiple times and the results be aggregated, for example by taking min, max, avg, or median.
+
+## Results collection
+
+Local test results for main branch, and results of daily performance tests, are stored in a neon project deployed in production environment. There is a Grafana dashboard that visualizes the results. Here is the [dashboard](https://observer.zenith.tech/d/DGKBm9Jnz/perf-test-results?orgId=1). The main problem with it is the unavailability to point at particular commit, though the data for that is available in the database. Needs some tweaking from someone who knows Grafana tricks.
+
+There is also an inconsistency in test naming. Test name should be the same across platforms, and results can be differentiated by the platform field. But currently, platform is sometimes included in test name because of the way how parametrization works in pytest. I.e. there is a platform switch in the dashboard with zenith-local-ci and zenith-staging variants. I.e. some tests under zenith-local-ci value for a platform switch are displayed as `Test test_runner/performance/test_bulk_insert.py::test_bulk_insert[vanilla]` and `Test test_runner/performance/test_bulk_insert.py::test_bulk_insert[zenith]` which is highly confusing.

test_runner/performance/test_hot_page.py

@@ -0,0 +1,36 @@
+import pytest
+from contextlib import closing
+from fixtures.compare_fixtures import PgCompare
+from pytest_lazyfixture import lazy_fixture  # type: ignore
+
+
+@pytest.mark.parametrize(
+    "env",
+    [
+        # The test is too slow to run in CI, but fast enough to run with remote tests
+        pytest.param(lazy_fixture("zenith_compare"), id="zenith", marks=pytest.mark.slow),
+        pytest.param(lazy_fixture("vanilla_compare"), id="vanilla", marks=pytest.mark.slow),
+        pytest.param(lazy_fixture("remote_compare"), id="remote", marks=pytest.mark.remote_cluster),
+    ])
+def test_hot_page(env: PgCompare):
+    # Update the same page many times, then measure read performance
+    num_writes = 1000000
+
+    with closing(env.pg.connect()) as conn:
+        with conn.cursor() as cur:
+
+            # Write many updates to the same row
+            with env.record_duration('write'):
+                cur.execute('create table t (i integer);')
+                cur.execute('insert into t values (0);')
+                for i in range(num_writes):
+                    cur.execute(f'update t set i = {i};')
+
+            # Write 3-4 MB to evict t from compute cache
+            cur.execute('create table f (i integer);')
+            cur.execute(f'insert into f values (generate_series(1,100000));')
+
+            # Read
+            with env.record_duration('read'):
+                cur.execute('select * from t;')
+                cur.fetchall()

test_runner/performance/test_hot_table.py

@@ -0,0 +1,35 @@
+import pytest
+from contextlib import closing
+from fixtures.compare_fixtures import PgCompare
+from pytest_lazyfixture import lazy_fixture  # type: ignore
+
+
+@pytest.mark.parametrize(
+    "env",
+    [
+        # The test is too slow to run in CI, but fast enough to run with remote tests
+        pytest.param(lazy_fixture("zenith_compare"), id="zenith", marks=pytest.mark.slow),
+        pytest.param(lazy_fixture("vanilla_compare"), id="vanilla", marks=pytest.mark.slow),
+        pytest.param(lazy_fixture("remote_compare"), id="remote", marks=pytest.mark.remote_cluster),
+    ])
+def test_hot_table(env: PgCompare):
+    # Update a small table many times, then measure read performance
+    num_rows = 100000  # Slightly larger than shared buffers size  TODO validate
+    num_writes = 1000000
+    num_reads = 10
+
+    with closing(env.pg.connect()) as conn:
+        with conn.cursor() as cur:
+
+            # Write many updates to a small table
+            with env.record_duration('write'):
+                cur.execute('create table t (i integer primary key);')
+                cur.execute(f'insert into t values (generate_series(1,{num_rows}));')
+                for i in range(num_writes):
+                    cur.execute(f'update t set i = {i + num_rows} WHERE i = {i};')
+
+            # Read the table
+            with env.record_duration('read'):
+                for i in range(num_reads):
+                    cur.execute('select * from t;')
+                    cur.fetchall()