Merge pull request #17 from ololobus/tpc-ds

Add TPC-DS stress tests

Author: maksm90

.gitignore

@@ -9,3 +9,5 @@ pg_query_state--*.sql
 cscope.out
 tags
 Dockerfile
+tmp_stress
+

.travis.yml

@@ -18,11 +18,11 @@ notifications:
         on_failure: always
 
 env:
-    - PG_VERSION=12 LEVEL=hardcore
+    - PG_VERSION=12 LEVEL=hardcore USE_TPCDS=1
     - PG_VERSION=12
-    - PG_VERSION=11 LEVEL=hardcore
+    - PG_VERSION=11 LEVEL=hardcore USE_TPCDS=1
     - PG_VERSION=11
-    - PG_VERSION=10 LEVEL=hardcore
+    - PG_VERSION=10 LEVEL=hardcore USE_TPCDS=1
     - PG_VERSION=10
     - PG_VERSION=9.6 LEVEL=hardcore
     - PG_VERSION=9.6

Dockerfile.tmpl

@@ -2,12 +2,12 @@ FROM postgres:${PG_VERSION}-alpine
 
 # Install dependencies
 RUN apk add --no-cache \
-	openssl curl \
+	openssl curl git \
 	perl perl-ipc-run \
 	make musl-dev gcc bison flex coreutils \
 	zlib-dev libedit-dev \
 	clang clang-analyzer linux-headers \
-	python2 python2-dev py2-virtualenv;
+	python3 python3-dev py3-virtualenv;
 
 
 # Install fresh valgrind
@@ -35,4 +35,4 @@ ADD . /pg/testdir
 WORKDIR /pg/testdir
 
 USER postgres
-ENTRYPOINT LEVEL=${LEVEL} /run.sh
+ENTRYPOINT LEVEL=${LEVEL} USE_TPCDS=${USE_TPCDS} /run.sh

Makefile

@@ -9,7 +9,7 @@ DATA_built = $(EXTENSION)--$(EXTVERSION).sql
 PGFILEDESC = "pg_query_state - facility to track progress of plan execution"
 
 EXTRA_CLEAN = ./isolation_output $(EXTENSION)--$(EXTVERSION).sql \
-	Dockerfile ./tests/*.pyc
+	Dockerfile ./tests/*.pyc ./tmp_stress
 
 ifdef USE_PGXS
 PG_CONFIG ?= pg_config

README.md

@@ -27,26 +27,41 @@ Add module name to the `shared_preload_libraries` parameter in `postgresql.conf`
 shared_preload_libraries = 'pg_query_state'
 ```
 It is essential to restart the PostgreSQL instance. After that, execute the following query in psql:
-```
+```sql
 CREATE EXTENSION pg_query_state;
 ```
 Done!
 
 ## Tests
-Tests using parallel sessions using python 2.7 script:
-   ```
-   python tests/pg_qs_test_runner.py [OPTION]...
-   ```
+Test using parallel sessions with Python 3+ compatible script:
+```shell
+python tests/pg_qs_test_runner.py [OPTION]...
+```
 *prerequisite packages*:
 * `psycopg2` version 2.6 or later
 * `PyYAML` version 3.11 or later
-   
+* `progressbar2` for stress test progress reporting
+
 *options*:
 * *- -host* --- postgres server host, default value is *localhost*
 * *- -port* --- postgres server port, default value is *5432*
 * *- -database* --- database name, default value is *postgres*
 * *- -user* --- user name, default value is *postgres*
 * *- -password* --- user's password, default value is empty
+* *- -tpc-ds* --- runs only stress tests on TPC-DS benchmark
+
+Or run all tests in `Docker` using:
+
+```shell
+export LEVEL=hardcore
+export USE_TPCDS=1
+export PG_VERSION=12
+
+docker-compose build
+docker-compose run tests
+```
+
+There are different test levels: `hardcore`, `nightmare` (runs tests under `valgrind`) and `stress` (runs tests under `TPC-DS` load). 
 
 ## Function pg\_query\_state
 ```plpgsql
@@ -93,11 +108,11 @@ This parameters is set on called side before running any queries whose states ar
 
 ## Examples
 Set maximum number of parallel workers on `gather` node equals `2`:
-```
+```sql
 postgres=# set max_parallel_workers_per_gather = 2;
 ```
 Assume one backend with pid = 49265 performs a simple query:
-```
+```sql
 postgres=# select pg_backend_pid();
  pg_backend_pid
  ----------------
@@ -106,7 +121,7 @@ postgres=# select pg_backend_pid();
 postgres=# select count(*) from foo join bar on foo.c1=bar.c1;
 ```
 Other backend can extract intermediate state of execution that query:
-```
+```sql
 postgres=# \x
 postgres=# select * from pg_query_state(49265);
 -[ RECORD 1 ]+-------------------------------------------------------------------------------------------------------------------------
@@ -151,11 +166,11 @@ In example above working backend spawns two parallel workers with pids `49324` a
 `Seq Scan` node has statistics on passed loops (average number of rows delivered to `Nested Loop` and number of passed loops are shown) and statistics on current loop. Other nodes has statistics only for current loop as this loop is first (`loop number` = 1).
 
 Assume first backend executes some function:
-```
+```sql
 postgres=# select n_join_foo_bar();
 ```
 Other backend can get the follow output:
-```
+```sql
 postgres=# select * from pg_query_state(49265);
 -[ RECORD 1 ]+------------------------------------------------------------------------------------------------------------------
 pid          | 49265
@@ -181,7 +196,7 @@ leader_pid   | (null)
 First row corresponds to function call, second - to query which is in the body of that function.
 
 We can get result plans in different format (e.g. `json`):
-```
+```sql
 postgres=# select * from pg_query_state(pid := 49265, format := 'json');
 -[ RECORD 1 ]+------------------------------------------------------------
 pid          | 49265

mk_dockerfile.sh

@@ -1,3 +1,5 @@
+#!/usr/bin/env sh
+
 if [ -z ${PG_VERSION+x} ]; then
 	echo PG_VERSION is not set!
 	exit 1
@@ -7,10 +9,16 @@ if [ -z ${LEVEL+x} ]; then
 	LEVEL=scan-build
 fi
 
+if [ -z ${USE_TPCDS+x} ]; then
+	USE_TPCDS=0
+fi
+
 echo PG_VERSION=${PG_VERSION}
 echo LEVEL=${LEVEL}
+echo USE_TPCDS=${USE_TPCDS}
 
 sed \
 	-e 's/${PG_VERSION}/'${PG_VERSION}/g \
 	-e 's/${LEVEL}/'${LEVEL}/g \
+	-e 's/${USE_TPCDS}/'${USE_TPCDS}/g \
 	Dockerfile.tmpl > Dockerfile

run_tests.sh

@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 
 #
-# Copyright (c) 2018, Postgres Professional
+# Copyright (c) 2019, Postgres Professional
 #
 # supported levels:
 #		* standard
@@ -55,7 +55,7 @@ fi
 
 # build and install PostgreSQL
 if [ "$LEVEL" = "hardcore" ] || \
-  [ "$LEVEL" = "nightmare" ]; then
+   [ "$LEVEL" = "nightmare" ]; then
 	# enable Valgrind support
 	sed -i.bak "s/\/* #define USE_VALGRIND *\//#define USE_VALGRIND/g" src/include/pg_config_manual.h
 
@@ -143,10 +143,14 @@ if [ -f regression.diffs ]; then cat regression.diffs; fi
 
 # run python tests
 set +x -e
-virtualenv /tmp/env && source /tmp/env/bin/activate &&
-pip install PyYAML && pip install psycopg2
+python3 -m venv /tmp/env && source /tmp/env/bin/activate &&
+pip install -r tests/requirements.txt
 set -e #exit virtualenv with error code
 python tests/pg_qs_test_runner.py --port $PGPORT
+if [[ "$USE_TPCDS" == "1" ]]; then
+	python tests/pg_qs_test_runner.py --port $PGPORT --tpc-ds-setup
+	python tests/pg_qs_test_runner.py --port $PGPORT --tpc-ds-run
+fi
 deactivate
 set -x
 
@@ -170,4 +174,4 @@ gcov *.c *.h
 set +ux
 
 # send coverage stats to Codecov
-bash <(curl -s https://codecov.io/bash)
+bash <(curl -s https://codecov.io/bash)

tests/common.py

@@ -0,0 +1,103 @@
+'''
+common.py
+Copyright (c) 2016-2020, Postgres Professional
+'''
+
+import psycopg2
+import psycopg2.extensions
+import select
+import time
+
+BACKEND_IS_IDLE_INFO = 'INFO:  state of backend is idle\n'
+BACKEND_IS_ACTIVE_INFO = 'INFO:  state of backend is active\n'
+
+def wait(conn):
+	"""wait for some event on connection to postgres"""
+	while 1:
+		state = conn.poll()
+		if state == psycopg2.extensions.POLL_OK:
+			break
+		elif state == psycopg2.extensions.POLL_WRITE:
+			select.select([], [conn.fileno()], [])
+		elif state == psycopg2.extensions.POLL_READ:
+			select.select([conn.fileno()], [], [])
+		else:
+			raise psycopg2.OperationalError("poll() returned %s" % state)
+
+def n_async_connect(config, n=1):
+	"""establish n asynchronious connections to the postgres with specified config"""
+
+	aconfig = config.copy()
+	aconfig['async'] = True
+
+	result = []
+	for _ in range(n):
+		conn = psycopg2.connect(**aconfig)
+		wait(conn)
+		result.append(conn)
+	return result
+
+def n_close(conns):
+	"""close connections to postgres"""
+
+	for conn in conns:
+		conn.close()
+
+def pg_query_state(config, pid, verbose=False, costs=False, timing=False, \
+								buffers=False, triggers=False, format='text'):
+	"""
+	Get query state from backend with specified pid and optional parameters.
+	Save any warning, info, notice and log data in global variable 'notices'
+	"""
+
+	conn = psycopg2.connect(**config)
+	curs = conn.cursor()
+
+	curs.callproc('pg_query_state', (pid, verbose, costs, timing, buffers, triggers, format))
+	result = curs.fetchall()
+	notices = conn.notices[:]
+	conn.close()
+
+	return result, notices
+
+def onetime_query_state(config, async_conn, query, args={}, num_workers=0):
+	"""
+	Get intermediate state of 'query' on connection 'async_conn' after number of 'steps'
+	of node executions from start of query
+	"""
+
+	acurs = async_conn.cursor()
+
+	set_guc(async_conn, 'enable_mergejoin', 'off')
+	set_guc(async_conn, 'max_parallel_workers_per_gather', num_workers)
+	acurs.execute(query)
+
+	# extract current state of query progress
+	MAX_PG_QS_RETRIES = 10
+	DELAY_BETWEEN_RETRIES = 0.1
+	pg_qs_args = {
+			'config': config,
+			'pid': async_conn.get_backend_pid()
+			}
+	for k, v in args.items():
+		pg_qs_args[k] = v
+	n_retries = 0
+	while True:
+		result, notices = pg_query_state(**pg_qs_args)
+		n_retries += 1
+		if len(result) > 0:
+			break
+		if n_retries >= MAX_PG_QS_RETRIES:
+			# pg_query_state callings don't return any result, more likely run
+			# query has completed
+			break
+		time.sleep(DELAY_BETWEEN_RETRIES)
+	wait(async_conn)
+
+	set_guc(async_conn, 'enable_mergejoin', 'on')
+	return result, notices
+
+def set_guc(async_conn, param, value):
+	acurs = async_conn.cursor()
+	acurs.execute('set %s to %s' % (param, value))
+	wait(async_conn)