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

# 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

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

PG_CONFIG ?= pg_config

It is essential to restart the PostgreSQL instance. After that, execute the following query in psql:

CREATE EXTENSION pg_query_state;

Done!

Tests using parallel sessionsusing python 2.7 script:

Test using parallel sessionswith Python 3+ compatible script:

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:

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).

Set maximum number of parallel workers on`gather` node equals`2`:

postgres=# set max_parallel_workers_per_gather = 2;

Assume one backend with pid = 49265 performs a simple query:

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:

postgres=# \x

postgres=# select * from pg_query_state(49265);

-[ RECORD1 ]+-------------------------------------------------------------------------------------------------------------------------

`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:

postgres=# select n_join_foo_bar();

Other backend can get the follow output:

postgres=# select * from pg_query_state(49265);

-[ RECORD1 ]+------------------------------------------------------------------------------------------------------------------

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`):

postgres=# select * from pg_query_state(pid := 49265, format := 'json');

-[ RECORD1 ]+------------------------------------------------------------

pid |49265

if [-z${PG_VERSION+x} ];then

echo PG_VERSION is not set!

exit 1

LEVEL=scan-build

if [-z${USE_TPCDS+x} ];then

USE_TPCDS=0

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

if ["$LEVEL"="hardcore" ]|| \

["$LEVEL"="nightmare" ];then

["$LEVEL"="nightmare" ];then

sed -i.bak"s/\/* #define USE_VALGRIND *\//#define USE_VALGRIND/g" src/include/pg_config_manual.h

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

deactivate

set -x

set +ux

bash<(curl -s https://codecov.io/bash)

bash<(curl -s https://codecov.io/bash)

importpsycopg2.extensions

defwait(conn):

while1:

state=conn.poll()

ifstate==psycopg2.extensions.POLL_OK:

elifstate==psycopg2.extensions.POLL_WRITE:

select.select([], [conn.fileno()], [])

elifstate==psycopg2.extensions.POLL_READ:

select.select([conn.fileno()], [], [])

else:

raisepsycopg2.OperationalError("poll() returned %s"%state)

defn_async_connect(config,n=1):

aconfig=config.copy()

aconfig['async']=True

result= []

for_inrange(n):

conn=psycopg2.connect(**aconfig)

wait(conn)

result.append(conn)

defn_close(conns):

forconninconns:

conn.close()

defpg_query_state(config,pid,verbose=False,costs=False,timing=False, \

buffers=False,triggers=False,format='text'):

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()

returnresult,notices

defonetime_query_state(config,async_conn,query,args={},num_workers=0):

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)

pg_qs_args= {

'config':config,

'pid':async_conn.get_backend_pid()

}

fork,vinargs.items():

pg_qs_args[k]=v

whileTrue:

result,notices=pg_query_state(**pg_qs_args)

iflen(result)>0:

ifn_retries>=MAX_PG_QS_RETRIES:

time.sleep(DELAY_BETWEEN_RETRIES)

wait(async_conn)

set_guc(async_conn,'enable_mergejoin','on')

returnresult,notices

defset_guc(async_conn,param,value):

acurs=async_conn.cursor()

acurs.execute('set %s to %s'% (param,value))

wait(async_conn)