Jan 10, 2020 · Oct 25, 2019 · Oct 25, 2019 · Oct 25, 2019 · Oct 25, 2019 · Oct 26, 2019
diff --git a/.gitignore b/.gitignore
 cscope.out
 tags
 Dockerfile
 tmp_stress

diff --git a/.travis.yml b/.travis.yml
        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
diff --git a/Dockerfile.tmpl b/Dockerfile.tmpl

 # Install dependencies
 RUN apk add --no-cache \
 openssl curl \
 openssl curlgit\
 perl perl-ipc-run \
 make musl-dev gcc bison flex coreutils \
 zlib-dev libedit-dev \
 clang clang-analyzer linux-headers \
 python2 python2-devpy2-virtualenv;
 python3 python3-devpy3-virtualenv;


 # Install fresh valgrind
 WORKDIR /pg/testdir

 USER postgres
 ENTRYPOINT LEVEL=${LEVEL} /run.sh
 ENTRYPOINT LEVEL=${LEVEL}USE_TPCDS=${USE_TPCDS}/run.sh
diff --git a/Makefile b/Makefile
 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
diff --git a/README.md b/README.md
 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 sessionsusing python 2.7 script:
 ```
 python tests/pg_qs_test_runner.py [OPTION]...
 ```
 Test using parallel sessionswith 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

 ## 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
 ----------------
 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 ]+-------------------------------------------------------------------------------------------------------------------------
 `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
 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
diff --git a/mk_dockerfile.sh b/mk_dockerfile.sh
 #!/usr/bin/env sh

 if [ -z ${PG_VERSION+x} ]; then
 echo PG_VERSION is not set!
 exit 1
 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
diff --git a/run_tests.sh b/run_tests.sh
 #!/usr/bin/env bash

 #
 # Copyright (c)2018, Postgres Professional
 # Copyright (c)2019, Postgres Professional
 #
 # supported levels:
 #* standard

 # 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


 # run python tests
 set +x -e
 virtualenv /tmp/env && source /tmp/env/bin/activate &&
 pip installPyYAML && 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

 set +ux

 # send coverage stats to Codecov
 bash <(curl -s https://codecov.io/bash)
 bash <(curl -s https://codecov.io/bash)
diff --git a/tests/common.py b/tests/common.py
 '''
 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)
Original file line number	Diff line number	Diff line change
Expand Up		@@ -9,3 +9,5 @@ pg_query_state--*.sql
		cscope.out
		tags
		Dockerfile
		tmp_stress
Original file line number	Diff line number	Diff line change
Expand Up		@@ -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
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -2,12 +2,12 @@ FROM postgres:${PG_VERSION}-alpine

		# Install dependencies
		RUN apk add --no-cache \
		openssl curl \
		openssl curlgit\
		perl perl-ipc-run \
		make musl-dev gcc bison flex coreutils \
		zlib-dev libedit-dev \
		clang clang-analyzer linux-headers \
		python2 python2-devpy2-virtualenv;
		python3 python3-devpy3-virtualenv;


		# Install fresh valgrind
Expand DownExpand Up		@@ -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
Original file line number	Diff line number	Diff line change
Expand Up		@@ -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
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -26,26 +26,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 sessionsusing python 2.7 script:
		```
		python tests/pg_qs_test_runner.py [OPTION]...
		```
		Test using parallel sessionswith 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
Expand DownExpand Up		@@ -92,11 +107,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
		----------------
Expand All		@@ -105,7 +120,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 ]+-------------------------------------------------------------------------------------------------------------------------
Expand DownExpand Up		@@ -150,11 +165,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
Expand All		@@ -180,7 +195,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
Expand Down
Original file line number	Diff line number	Diff line change
		@@ -1,3 +1,5 @@
		#!/usr/bin/env sh

		if [ -z ${PG_VERSION+x} ]; then
		echo PG_VERSION is not set!
		exit 1
Expand All		@@ -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
Original file line number	Diff line number	Diff line change
		@@ -1,7 +1,7 @@
		#!/usr/bin/env bash

		#
		# Copyright (c)2018, Postgres Professional
		# Copyright (c)2019, Postgres Professional
		#
		# supported levels:
		#* standard
Expand DownExpand Up		@@ -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

Expand DownExpand Up		@@ -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 installPyYAML && 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

Expand All		@@ -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)
Original file line number	Diff line number	Diff line change
		@@ -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)