[BACKPORT 2024.2][#25929] Docdb: Disable catalog version check when reading sequences as of time

Summary:
Original commit: 7fc910d / D41911
Currently, clone is failing when the database has a sequence, and the time we are cloning to is a time before a DDL that increments the last_breaking_version of the source database. Clone is failing in ysql_dump phase when executing the following query to get the sequence data: `SELECT last_value, is_called FROM public.seq_name`. The error is:
 `ERROR:  The catalog snapshot used for this transaction has been invalidated: expected: 2, got: 1: MISMATCHED_SCHEMA`.

Generally, ysql_dump sets the GUC `yb_disable_catalog_version_check` to true, which should not set the catalog version in all the read requests that ysql_dump is issuing, and thus no catalog version check is performed.
However, as sequences have their own read path, the GUC `yb_disable_catalog_version_check` is not taking effect, and the catalog version check is returning a catalog version mismatch.

Fixed by extending the role of `yb_disable_catalog_version_check` to cover the sequences' read path. When set to true, we don't send the catalog version with the read request to the tserver and thus disable the catalog version check for such a sequence read query. This allows ysql_dump to read the sequence as of a time regardless of the catalog version at that time and thus, being able to clone sequences to a time before a DDL happened.
Jira: DB-15245

Test Plan: ./yb_build.sh --cxx-test integration-tests_minicluster-snapshot-test --gtest_filter Colocation/PgCloneTestWithColocatedDBParam.CloneWithSequencesAndDdl/0

Reviewers: asrivastava

Reviewed By: asrivastava

Subscribers: slingam, yql, ybase

Differential Revision: https://phorge.dev.yugabyte.com/D42696

Author: yamen-haddad

src/postgres/src/backend/utils/misc/pg_yb_utils.c

@@ -1458,7 +1458,6 @@ bool yb_enable_fkey_catcache = true;
 int yb_insert_on_conflict_read_batch_size = 1024;
 bool yb_enable_inplace_index_update = true;
 bool yb_enable_nop_alter_role_optimization = true;
-bool yb_disable_catalog_version_check = false;
 
 YBUpdateOptimizationOptions yb_update_optimization_options = {
 	.has_infra = true,

src/postgres/src/include/pg_yb_utils.h

@@ -582,8 +582,6 @@ extern int yb_insert_on_conflict_read_batch_size;
  */
 extern bool yb_enable_nop_alter_role_optimization;
 
-extern bool yb_disable_catalog_version_check;
-
 //------------------------------------------------------------------------------
 // GUC variables needed by YB via their YB pointers.
 extern int StatementTimeout;

src/yb/integration-tests/minicluster-snapshot-test.cc

@@ -975,6 +975,23 @@ TEST_P(PgCloneTestWithColocatedDBParam, YB_DISABLE_TEST_IN_SANITIZERS(CloneWithS
   ASSERT_EQ(row, kRows[3]);
 }
 
+TEST_P(PgCloneTestWithColocatedDBParam, YB_DISABLE_TEST_IN_SANITIZERS(CloneWithSequencesAndDdl)) {
+  auto seq_table_name = "table_with_sequence";
+  ASSERT_OK(
+      source_conn_->ExecuteFormat("CREATE TABLE $0 (id INT, i2 SERIAL, c1 INT)", seq_table_name));
+  ASSERT_OK(source_conn_->ExecuteFormat("INSERT INTO $0 (id,c1) VALUES (11,22)", seq_table_name));
+  auto clone_to_time = ASSERT_RESULT(GetCurrentTime()).ToInt64();
+  // Run a DDL that increments the last_breaking_version of pg_yb_catalog.
+  ASSERT_OK(source_conn_->ExecuteFormat("ALTER TABLE $0 RENAME COLUMN c1 TO c2", seq_table_name));
+  ASSERT_OK(source_conn_->ExecuteFormat(
+      "CREATE DATABASE $0 TEMPLATE $1 AS OF $2", kTargetNamespaceName1, kSourceNamespaceName,
+      clone_to_time));
+  auto target_conn = ASSERT_RESULT(ConnectToDB(kTargetNamespaceName1));
+  auto row = ASSERT_RESULT(
+      (target_conn.FetchRow<int64_t>(Format("SELECT count(*) FROM $0", seq_table_name))));
+  ASSERT_EQ(row, 1);
+}
+
 // Test yb_database_clones (ysql function to list clones)
 TEST_F(PgCloneTest, YB_DISABLE_TEST_IN_SANITIZERS(YsqlListClonesAPI)) {
   std::string list_clones_query =

src/yb/yql/pggate/pg_client.cc

@@ -688,17 +688,19 @@ class PgClient::Impl : public BigDataFetcher {
   }
 
   Result<std::pair<int64_t, bool>> ReadSequenceTuple(
-      int64_t db_oid, int64_t seq_oid, uint64_t ysql_catalog_version,
+      int64_t db_oid, int64_t seq_oid, std::optional<uint64_t> ysql_catalog_version,
       bool is_db_catalog_version_mode, std::optional<uint64_t> read_time) {
     tserver::PgReadSequenceTupleRequestPB req;
     req.set_session_id(session_id_);
     req.set_db_oid(db_oid);
     req.set_seq_oid(seq_oid);
-    if (is_db_catalog_version_mode) {
-      DCHECK(FLAGS_ysql_enable_db_catalog_version_mode);
-      req.set_ysql_db_catalog_version(ysql_catalog_version);
-    } else {
-      req.set_ysql_catalog_version(ysql_catalog_version);
+    if (ysql_catalog_version) {
+      if (is_db_catalog_version_mode) {
+        DCHECK(FLAGS_ysql_enable_db_catalog_version_mode);
+        req.set_ysql_db_catalog_version(*ysql_catalog_version);
+      } else {
+        req.set_ysql_catalog_version(*ysql_catalog_version);
+      }
     }
 
     if (read_time) {
@@ -1644,8 +1646,8 @@ Result<std::pair<int64_t, int64_t>> PgClient::FetchSequenceTuple(int64_t db_oid,
 }
 
 Result<std::pair<int64_t, bool>> PgClient::ReadSequenceTuple(
-    int64_t db_oid, int64_t seq_oid, uint64_t ysql_catalog_version, bool is_db_catalog_version_mode,
-    std::optional<uint64_t> read_time) {
+    int64_t db_oid, int64_t seq_oid, std::optional<uint64_t> ysql_catalog_version,
+    bool is_db_catalog_version_mode, std::optional<uint64_t> read_time) {
   return impl_->ReadSequenceTuple(
       db_oid, seq_oid, ysql_catalog_version, is_db_catalog_version_mode, read_time);
 }

src/yb/yql/pggate/pg_client.h

@@ -211,7 +211,7 @@ class PgClient {
                                                          bool cycle);
 
   Result<std::pair<int64_t, bool>> ReadSequenceTuple(
-      int64_t db_oid, int64_t seq_oid, uint64_t ysql_catalog_version,
+      int64_t db_oid, int64_t seq_oid, std::optional<uint64_t> ysql_catalog_version,
       bool is_db_catalog_version_mode, std::optional<uint64_t> read_time = std::nullopt);
 
   Status DeleteSequenceTuple(int64_t db_oid, int64_t seq_oid);

src/yb/yql/pggate/pg_session.cc

@@ -491,12 +491,17 @@ Result<std::pair<int64_t, bool>> PgSession::ReadSequenceTuple(int64_t db_oid,
                                                               int64_t seq_oid,
                                                               uint64_t ysql_catalog_version,
                                                               bool is_db_catalog_version_mode) {
+  std::optional<uint64_t> optional_ysql_catalog_version = std::nullopt;
+  std::optional<uint64_t> optional_yb_read_time = std::nullopt;
+  if (!yb_disable_catalog_version_check) {
+    optional_ysql_catalog_version = ysql_catalog_version;
+  }
   if (yb_read_time != 0) {
-    return pg_client_.ReadSequenceTuple(
-        db_oid, seq_oid, ysql_catalog_version, is_db_catalog_version_mode, yb_read_time);
+    optional_yb_read_time = yb_read_time;
   }
   return pg_client_.ReadSequenceTuple(
-      db_oid, seq_oid, ysql_catalog_version, is_db_catalog_version_mode);
+      db_oid, seq_oid, optional_ysql_catalog_version, is_db_catalog_version_mode,
+      optional_yb_read_time);
 }
 
 //--------------------------------------------------------------------------------------------------

src/yb/yql/pggate/util/yb_guc.cc

@@ -81,6 +81,7 @@ int yb_explicit_row_locking_batch_size = 1;
 
 uint64_t yb_read_time = 0;
 bool yb_is_read_time_ht = false;
+bool yb_disable_catalog_version_check = false;
 
 int yb_read_after_commit_visibility = 0;
 

src/yb/yql/pggate/util/yb_guc.h

@@ -154,6 +154,8 @@ extern int yb_xcluster_consistency_level;
 
 extern uint64_t yb_read_time;
 extern bool yb_is_read_time_ht;
+extern bool yb_disable_catalog_version_check;
+
 /*
  * Allows for customizing the number of rows to be prefetched.
  */

src/yb/yql/pggate/util/ybc_util.h

@@ -19,6 +19,7 @@
 #include <stddef.h>
 #include <stdint.h>
 
+#include "yb/yql/pggate/util/yb_guc.h"
 #include "yb/yql/pggate/ybc_pg_typedefs.h"
 
 #ifdef __cplusplus