[DBZ-PGYB][yugabyte/yugabyte-db#25716] Set GUC to disable catalog ver…

…sion check for connections (#169)

## Problem

Recent changes in YugabyteDB introduced some changes where the
`relcache/catcache` is reloaded as soon as `yb_read_time` is set. Now
that caused an issue with the snapshots in the connector i.e. whenever
the connector sets the `yb_read_time` and goes on to take the snapshot
by running a `SELECT *` query, it gets the following error:

```
Caused by: io.debezium.DebeziumException: com.yugabyte.util.PSQLException: ERROR: The catalog snapshot used for this transaction has been invalidated: expected: 1, got: 0: MISMATCHED_SCHEMA
  Where: Catalog Version Mismatch: A DDL occurred while processing this query. Try again.
```

## Solution

As a workaround, we will now bypass the catalog version check by setting
a GUC `yb_disable_catalog_version_check` at the connection level so that
we do not hit the error during snapshot.

Note that we do not require this during streaming phase and since
streaming used a separate object of `ReplicationConnection` class, we
will not be modifying it.

This closes yugabyte/yugabyte-db#25716

debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java

@@ -284,8 +284,35 @@ protected Optional<String> getSnapshotSelect(RelationalSnapshotContext<PostgresP
         return snapshotter.buildSnapshotQuery(tableId, columns);
     }
 
+    protected void disableCatalogVersionCheck() throws SQLException {
+        final String disableCatalogVersionCheckStmt =
+           "DO " +
+           "LANGUAGE plpgsql $$ " +
+           "BEGIN " +
+               "SET yb_disable_catalog_version_check = true; " +
+           "EXCEPTION " +
+               "WHEN sqlstate '42704' THEN "+
+                   "RAISE EXCEPTION 'GUC not found'; " +
+               "WHEN OTHERS THEN " +
+                   "CALL disable_catalog_version_check(); " +
+           "END $$;";
+
+        LOGGER.info("Disabling catalog version check with statement: {}", disableCatalogVersionCheckStmt);
+        try {
+            jdbcConnection.execute(disableCatalogVersionCheckStmt);
+        } catch (SQLException sqle) {
+            if (sqle.getMessage().contains("GUC not found")) {
+                LOGGER.warn("GUC not present: yb_disable_catalog_version_check");
+                jdbcConnection.execute("ABORT;");
+            } else {
+                throw sqle;
+            }
+        }
+    }
     protected void setSnapshotTransactionIsolationLevel(boolean isOnDemand) throws SQLException {
         if (!YugabyteDBServer.isEnabled() || connectorConfig.isYbConsistentSnapshotEnabled()) {
+            disableCatalogVersionCheck();
+
             LOGGER.info("Setting isolation level");
             String transactionStatement = snapshotter.snapshotTransactionIsolationLevelStatement(slotCreatedInfo, isOnDemand);
             LOGGER.info("Opening transaction with statement {}", transactionStatement);