From 060f6006d65e6051493fc69d2ba9d0624cebd7c7 Mon Sep 17 00:00:00 2001
From: Vaibhav Kushwaha <34186745+vaibhav-yb@users.noreply.github.com>
Date: Tue, 28 Jan 2025 17:55:31 +0530
Subject: [PATCH] [DBZ-PGYB][yugabyte/yugabyte-db#25716] Set GUC to disable
 catalog version 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
---
 .../PostgresSnapshotChangeEventSource.java    | 27 +++++++++++++++++++
 1 file changed, 27 insertions(+)

diff --git a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java b/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java
index fc0924cf37d..986938ff477 100644
--- a/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java
+++ b/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);