Fixes for logical replication

.github/composite-actions/install-extensions/action.yml

@@ -13,6 +13,14 @@ inputs:
     description: 'Postgres Parallel Query Mode'
     required: no
     default: false
+  psql_port:
+    description: "Postgres port"
+    required: false
+    default: 'null'
+  tsql_port:
+    description: "TDS port"
+    required: false
+    default: 1433
 
 runs:
   using: "composite"
@@ -22,11 +30,17 @@ runs:
         ulimit -c unlimited
         cd ~
         export PATH=/opt/mssql-tools/bin:$PATH
-        ~/${{inputs.install_dir}}/bin/initdb -D ~/${{inputs.install_dir}}/data/
-        ~/${{inputs.install_dir}}/bin/pg_ctl -c -D ~/${{inputs.install_dir}}/data/ -l logfile start
-        cd ${{inputs.install_dir}}/data
+
+        # add port as suffix in data directory name to make it unique
+        if [[ '${{ inputs.psql_port }}' != 'null' ]];then
+          export DATADIR=data_${{ inputs.psql_port }}
+        fi
+
+        ~/${{inputs.install_dir}}/bin/initdb -D ~/${{inputs.install_dir}}/$DATADIR/
+        cd ${{inputs.install_dir}}/$DATADIR
         sudo sed -i "s/#listen_addresses = 'localhost'/listen_addresses = '*'/g" postgresql.conf
         sudo sed -i "s/#shared_preload_libraries = ''/shared_preload_libraries = 'babelfishpg_tds, pg_stat_statements'/g" postgresql.conf
+        sudo sed -i "s/#port = 5432/port = ${{ inputs.psql_port }}/g" postgresql.conf
         ipaddress=$(ifconfig eth0 | grep 'inet ' | cut -d: -f2 | awk '{ print $2}')
         # Allow only runner to have trust authentication, all other users must provide a password
         {
@@ -37,8 +51,9 @@ runs:
           sudo echo "host    all             all             0.0.0.0/0                md5"
           sudo echo "host    all             all             ::/0                     md5"
         } > pg_hba.conf
-        ~/${{inputs.install_dir}}/bin/pg_ctl -c -D ~/${{inputs.install_dir}}/data/ -l logfile restart
+        ~/${{inputs.install_dir}}/bin/pg_ctl -c -D ~/${{inputs.install_dir}}/$DATADIR/ -l logfile start
         cd ~/work/babelfish_extensions/babelfish_extensions/
-        sudo ~/${{inputs.install_dir}}/bin/psql -d postgres -U runner -v user="jdbc_user" -v db="jdbc_testdb" -v migration_mode=${{inputs.migration_mode}} -v parallel_query_mode=${{inputs.parallel_query_mode}} -f .github/scripts/create_extension.sql
-        sqlcmd -S localhost -U "jdbc_user" -P 12345678 -Q "SELECT @@version GO"
+        sudo ~/${{inputs.install_dir}}/bin/psql -d postgres -U runner -v user="jdbc_user" -v db="jdbc_testdb" -v migration_mode=${{inputs.migration_mode}} -v tsql_port=${{inputs.tsql_port}} -v parallel_query_mode=${{inputs.parallel_query_mode}} -f .github/scripts/create_extension.sql
+        ~/${{inputs.install_dir}}/bin/pg_ctl -c -D ~/${{inputs.install_dir}}/$DATADIR/ -l logfile restart
+        sqlcmd -S localhost,${{inputs.tsql_port}} -U "jdbc_user" -P 12345678 -Q "SELECT @@version GO"
       shell: bash

.github/composite-actions/run-jdbc-tests/action.yml

@@ -1,13 +1,20 @@
 name: 'Run JDBC Tests'
 description: 'Run Babel JDBC test framework'
 
+inputs:
+  input_dir:
+    description: 'Test input directory'
+    required: no
+    default: input
+
 runs:
   using: "composite"
   steps:
     - name: Run JDBC Tests
       run: |
           export PATH=~/${{env.INSTALL_DIR}}/bin:$PATH
           export PG_SRC=~/work/babelfish_extensions/postgresql_modified_for_babelfish
+          export inputFilesPath=${{inputs.input_dir}}
           cd test/JDBC/
           mvn test
       shell: bash

.github/scripts/create_extension.sql

@@ -8,6 +8,7 @@ GRANT ALL ON SCHEMA sys to :user;
 ALTER USER :user CREATEDB;
 ALTER SYSTEM SET babelfishpg_tsql.database_name = :db;
 ALTER SYSTEM SET babelfishpg_tsql.migration_mode = :'migration_mode';
+ALTER SYSTEM SET babelfishpg_tds.port = :tsql_port;
 
 \if :parallel_query_mode
     ALTER SYSTEM SET parallel_setup_cost = 0;

.github/workflows/logical-replication-test.yml

@@ -0,0 +1,110 @@
+name: Logical Replication Test
+on: [push, pull_request]
+
+jobs:
+  run-logical-replication-tests:
+    env:
+      INSTALL_DIR: psql
+    runs-on: ubuntu-20.04
+    steps:
+      - uses: actions/checkout@v2
+        id: checkout
+
+      - name: Install Dependencies
+        id: install-dependencies
+        if: always()
+        uses: ./.github/composite-actions/install-dependencies
+
+      - name: Build Modified Postgres
+        id: build-modified-postgres
+        if: always() && steps.install-dependencies.outcome == 'success'
+        uses: ./.github/composite-actions/build-modified-postgres
+        with:
+          install_dir: 'psql'
+          code_coverage: 'yes'
+
+      - name: Compile ANTLR
+        id: compile-antlr
+        if: always() && steps.build-modified-postgres.outcome == 'success'
+        uses: ./.github/composite-actions/compile-antlr
+
+      - name: Build Extensions
+        id: build-extensions
+        if: always() && steps.compile-antlr.outcome == 'success'
+        uses: ./.github/composite-actions/build-extensions
+
+      - name: Build tds_fdw Extension
+        id: build-tds_fdw-extension
+        if: always() && steps.build-extensions.outcome == 'success'
+        uses: ./.github/composite-actions/build-tds_fdw-extension
+
+      - name: Build PostGIS Extension
+        id: build-postgis-extension
+        if: always() && steps.build-tds_fdw-extension.outcome == 'success'
+        uses: ./.github/composite-actions/build-postgis-extension
+
+      - name: Start Publisher server
+        id: start-publisher
+        if: always() && steps.build-postgis-extension.outcome == 'success'
+        uses: ./.github/composite-actions/install-extensions
+
+      - name: Start Subscriber server
+        id: start-subscriber
+        if: always() && steps.start-publisher.outcome == 'success'
+        uses: ./.github/composite-actions/install-extensions
+        with:
+          psql_port: 5433
+          tsql_port: 8199
+
+      - name: Setup Publication and Subscription
+        id: setup-pub-sub
+        if: always() && steps.start-subscriber.outcome == 'success'
+        run: |
+          ~/${{inputs.install_dir}}/bin/psql -d jdbc_testdb -U runner -c "CREATE PUBLICATION my_pub;"
+          ~/${{inputs.install_dir}}/bin/psql -d jdbc_testdb -U runner -p 5433 -c "CREATE SUBSCRIPTION my_sub CONNECTION 'host=localhost port=5432 user=jdbc_user dbname=jdbc_testdb password=12345678' PUBLICATION my_pub;"
+
+      - name: Run Replication Tests
+        id: replication
+        if: always() && steps.setup-pub-sub.outcome == 'success'
+        timeout-minutes: 60
+        uses: ./.github/composite-actions/run-jdbc-tests
+        with:
+          input_dir: 'replication'
+
+      - name: Upload Log
+        if: always() && steps.replication.outcome == 'failure'
+        uses: actions/upload-artifact@v2
+        with:
+          name: postgres-log-replication
+          path: |
+            ~/psql/data/logfile
+            ~/psql/data2/logfile
+
+      # The test summary files contain paths with ':' characters, which is not allowed with the upload-artifact actions
+      - name: Rename Test Summary Files
+        id: test-file-rename
+        if: always() && steps.replication.outcome == 'failure'
+        run: |
+          cd test/JDBC/Info
+          timestamp=`ls -Art | tail -n 1`
+          cd $timestamp
+          mv $timestamp.diff ../output-diff.diff
+          mv "$timestamp"_runSummary.log ../run-summary.log
+
+      - name: Upload Run Summary 
+        if: always() && steps.test-file-rename.outcomeg == 'success'
+        uses: actions/upload-artifact@v2
+        with:
+          name: run-summary.log
+          path: test/JDBC/Info/run-summary.log
+
+      - name: Upload Output Diff
+        if: always() && steps.replication.outcome == 'failure'
+        uses: actions/upload-artifact@v2
+        with:
+          name: output-diff.diff
+          path: test/JDBC/Info/output-diff.diff
+
+      - name: Check and upload coredumps
+        if: always() && steps.replication.outcome == 'failure'
+        uses: ./.github/composite-actions/upload-coredump

contrib/babelfishpg_common/src/sqlvariant.c

@@ -29,6 +29,7 @@
 #include "parser/parse_coerce.h"
 #include "parser/parse_oper.h"
 #include "instr.h"
+#include "replication/slot.h"
 #include "utils/builtins.h"
 #include "utils/elog.h"
 #include "utils/guc.h"
@@ -79,6 +80,10 @@ sqlvariantin(PG_FUNCTION_ARGS)
 	Oid			typIOParam;
 	svhdr_5B_t *svhdr;
 
+	/* Input as a bytea instead if it is a walsender process. */
+	if (SessionReplicationRole == SESSION_REPLICATION_ROLE_REPLICA)
+		PG_RETURN_DATUM(byteain(fcinfo));
+
 	getTypeInputInfo(type, &input_func, &typIOParam);
 	/* evalute input fuction */
 	data_val = (text *) OidInputFunctionCall(input_func, str, typelem, atttypmod);
@@ -130,6 +135,16 @@ sqlvariantout(PG_FUNCTION_ARGS)
 	size_t		data_len = VARSIZE_ANY_EXHDR(vlena) - svhdr_size;
 	Datum	   *output_datum = palloc0(SIZEOF_DATUM);
 
+	/*
+	 * Output as a bytea instead if we are in a logical decoding context.
+	 * There are two ways it possible:
+	 * 1. MyReplicationSlot is logical.
+	 * 2. This is a logical walsender process.
+	 */
+	if ((MyReplicationSlot != NULL && SlotIsLogical(MyReplicationSlot)) ||
+		am_db_walsender)
+		PG_RETURN_DATUM(byteaout(fcinfo));
+
 	if (!get_typbyval(type))	/* pass by reference */
 		*output_datum = SV_DATUM(vlena, svhdr_size);
 	else						/* pass by value */

contrib/babelfishpg_common/src/varbinary.c

@@ -18,6 +18,7 @@
 #include "catalog/pg_collation.h"
 #include "catalog/pg_type.h"
 #include "collation.h"
+#include "commands/trigger.h"
 #include "common/int.h"
 #include "encoding/encoding.h"
 #include "lib/hyperloglog.h"
@@ -177,10 +178,15 @@ varbinaryin(PG_FUNCTION_ARGS)
 
 	len = strlen(inputText);
 
+	/*
+	 * Assume that input string is already hex encoded for following cases:
+	 * 1. Typmode is TSQLHexConstTypmod
+	 * 2. dump_restore GUC is set.
+	 * 3. This is logical replication applyworker, i.e. replication role is replica
+	 */
 	if (typmod == TSQLHexConstTypmod ||
-		(dump_restore && strcmp(dump_restore, "on") == 0))	/* Treat input string as
-															 * T-SQL hex constant
-															 * during restore */
+		(dump_restore && strcmp(dump_restore, "on") == 0) ||
+		 SessionReplicationRole == SESSION_REPLICATION_ROLE_REPLICA)
 	{
 		/*
 		 * calculate length of the binary code e.g. 0xFF should be 1 byte