mirror of
https://github.com/neondatabase/neon.git
synced 2026-01-15 01:12:56 +00:00
092a9b74d3
Newer version of mypy fixes buggy error when trying to update only boto3 stubs. However it brings new checks and starts to yell when we index into cusror.fetchone without checking for None first. So this introduces a wrapper to simplify quering for scalar values. I tried to use cursor_factory connection argument but without success. There can be a better way to do that, but this looks the simplest
81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
import asyncio
|
|
import random
|
|
|
|
from fixtures.neon_fixtures import NeonEnv, NeonEnvBuilder, Postgres
|
|
from fixtures.log_helper import log
|
|
from fixtures.utils import query_scalar
|
|
|
|
# Test configuration
|
|
#
|
|
# Create a table with {num_rows} rows, and perform {updates_to_perform} random
|
|
# UPDATEs on it, using {num_connections} separate connections.
|
|
num_connections = 10
|
|
num_rows = 100000
|
|
updates_to_perform = 10000
|
|
|
|
updates_performed = 0
|
|
|
|
|
|
# Run random UPDATEs on test table
|
|
async def update_table(pg: Postgres):
|
|
global updates_performed
|
|
pg_conn = await pg.connect_async()
|
|
|
|
while updates_performed < updates_to_perform:
|
|
updates_performed += 1
|
|
id = random.randrange(1, num_rows)
|
|
row = await pg_conn.fetchrow(f'UPDATE foo SET counter = counter + 1 WHERE id = {id}')
|
|
|
|
|
|
# Perform aggressive GC with 0 horizon
|
|
async def gc(env: NeonEnv, timeline: str):
|
|
psconn = await env.pageserver.connect_async()
|
|
|
|
while updates_performed < updates_to_perform:
|
|
await psconn.execute(f"do_gc {env.initial_tenant.hex} {timeline} 0")
|
|
|
|
|
|
# At the same time, run UPDATEs and GC
|
|
async def update_and_gc(env: NeonEnv, pg: Postgres, timeline: str):
|
|
workers = []
|
|
for worker_id in range(num_connections):
|
|
workers.append(asyncio.create_task(update_table(pg)))
|
|
workers.append(asyncio.create_task(gc(env, timeline)))
|
|
|
|
# await all workers
|
|
await asyncio.gather(*workers)
|
|
|
|
|
|
#
|
|
# Aggressively force GC, while running queries.
|
|
#
|
|
# (repro for https://github.com/neondatabase/neon/issues/1047)
|
|
#
|
|
def test_gc_aggressive(neon_env_builder: NeonEnvBuilder):
|
|
|
|
# Disable pitr, because here we want to test branch creation after GC
|
|
neon_env_builder.pageserver_config_override = "tenant_config={pitr_interval = '0 sec'}"
|
|
env = neon_env_builder.init_start()
|
|
env.neon_cli.create_branch("test_gc_aggressive", "main")
|
|
pg = env.postgres.create_start('test_gc_aggressive')
|
|
log.info('postgres is running on test_gc_aggressive branch')
|
|
|
|
with pg.cursor() as cur:
|
|
timeline = query_scalar(cur, "SHOW neon.timeline_id")
|
|
|
|
# Create table, and insert the first 100 rows
|
|
cur.execute('CREATE TABLE foo (id int, counter int, t text)')
|
|
cur.execute(f'''
|
|
INSERT INTO foo
|
|
SELECT g, 0, 'long string to consume some space' || g
|
|
FROM generate_series(1, {num_rows}) g
|
|
''')
|
|
cur.execute('CREATE INDEX ON foo(id)')
|
|
|
|
asyncio.run(update_and_gc(env, pg, timeline))
|
|
|
|
cur.execute('SELECT COUNT(*), SUM(counter) FROM foo')
|
|
r = cur.fetchone()
|
|
assert r is not None
|
|
assert r == (num_rows, updates_to_perform)
|