[yugabyte#26481] DocDB: Handle exclusive object lock(s) release for non schema changing DDL/DML(s)

Summary:
Commit yugabyte@47c5cf2 handled acquire/release for schema changing DDLs alone.

Problems:
1. In YB, we seem to have another category of DDLs which don't perform schema changes, but might take exclusive object locks. These aren't tracked by master's background DDL verification task.
2. pggate doesn't seem to tag statements like 'BACKFILL INDEX' as DDL, but they end up taking exclusive object locks.

When table locking feature is enabled, the release of exclusive locks taken by above type transactions wasn't handled in the above referenced commit. This diff addresses the issue by making `PgClientSession` responsible for triggering release of exclusive locks whenever necessary.

Note: Failure handling on the exclusive locks release path from the tserver side can go in a subsequent change. Tracked by yugabyte#26498
Jira: DB-15848

Test Plan:
Jenkins

./yb_build.sh --cxx-test='TEST_F(PgObjectLocksTest, BackfillIndexSanityTest) {'
./yb_build.sh --cxx-test='TEST_F(PgObjectLocksTest, ReleaseExpiredLocksInvalidatesCatalogCache) {'

Reviewers: rthallam, amitanand, zdrudi, myang, esheng

Reviewed By: amitanand, zdrudi

Subscribers: ybase, yql

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

Author: basavaraj29

src/yb/integration-tests/object_lock-test.cc

@@ -81,7 +81,6 @@ MATCHER_P(EqualsStatus, expected_status, "") {
 
 constexpr uint64_t kDefaultMasterYSQLLeaseTTLMilli = 5 * 1000;
 constexpr uint64_t kDefaultYSQLLeaseRefreshIntervalMilli = 500;
-constexpr size_t kPgConnectionTimeoutSecs = 10 * kTimeMultiplier;
 const std::string kTServerYsqlLeaseRefreshFlagName = "TEST_tserver_enable_ysql_lease_refresh";
 
 class ObjectLockTest : public MiniClusterTestWithClient<MiniCluster> {
@@ -1368,59 +1367,4 @@ Status ExternalObjectLockTest::WaitForTServerLease(const std::string& ts_uuid, M
       timeout, "Wait for master to establish tserver's lease");
 }
 
-TEST_F(ExternalObjectLockTest, ExclusiveLockReleaseInvalidatesCatalogCache) {
-  const auto ts1_idx = 1;
-  const auto ts2_idx = 2;
-  auto* ts1 = cluster_->tablet_server(ts1_idx);
-  auto* ts2 = cluster_->tablet_server(ts2_idx);
-
-  ts2->Shutdown();
-  LogWaiter log_waiter(ts2, "Received new lease epoch");
-  ASSERT_OK(ts2->Restart(
-      ExternalMiniClusterOptions::kDefaultStartCqlProxy,
-      {std::make_pair("TEST_ysql_disable_transparent_cache_refresh_retry", "true")}));
-  ASSERT_OK(log_waiter.WaitFor(MonoDelta::FromSeconds(kTimeMultiplier * 10)));
-
-  auto conn1 = ASSERT_RESULT(ConnectToTabletServer(ts1, kPgConnectionTimeoutSecs));
-  auto conn2 = ASSERT_RESULT(ConnectToTabletServer(ts2, kPgConnectionTimeoutSecs));
-
-  ASSERT_OK(conn1.Execute("CREATE TABLE test(k INT PRIMARY KEY, v INT)"));
-  ASSERT_OK(conn1.Execute("INSERT INTO test SELECT generate_series(1,11), 0"));
-
-  ASSERT_OK(conn2.FetchMatrix("SELECT * FROM test WHERE k=1", 1 /* rows */, 2 /* columns */));
-
-  // Disable catalog cache invalidation on tserver-master heartbeat path. Set it after the tserver
-  // boots up since setting it as part of initialization seems to error.
-  ASSERT_OK(cluster_->SetFlag(
-      ts2,
-      "TEST_tserver_disable_catalog_refresh_on_heartbeat",
-      "true"));
-
-  // Release of exclusive locks of the below DDL should invalidate catalog cache on all tservers.
-  ASSERT_OK(conn1.Execute("ALTER TABLE test ADD COLUMN v1 INT"));
-  // The DML should now see the updated schema and not hit a catalog cache/schema version mismatch.
-  ASSERT_OK(conn2.FetchMatrix("SELECT * FROM test WHERE k=1", 1 /* rows */, 3 /* columns */));
-}
-
-TEST_F(ExternalObjectLockTest, ConsecutiveAltersSucceedWithoutCatalogVersionIssues) {
-  const auto ts1_idx = 1;
-  const auto ts2_idx = 2;
-  auto* ts1 = cluster_->tablet_server(ts1_idx);
-  auto* ts2 = cluster_->tablet_server(ts2_idx);
-
-  auto conn1 = ASSERT_RESULT(ConnectToTabletServer(ts1, kPgConnectionTimeoutSecs));
-  auto conn2 = ASSERT_RESULT(ConnectToTabletServer(ts2, kPgConnectionTimeoutSecs));
-
-  ASSERT_OK(conn1.Execute("CREATE TABLE t1(c1 INT, c2 INT)"));
-  ASSERT_OK(conn2.Fetch("SELECT * FROM t1"));
-
-  ASSERT_OK(cluster_->SetFlag(
-      ts2,
-      "TEST_tserver_disable_catalog_refresh_on_heartbeat",
-      "true"));
-
-  ASSERT_OK(conn1.Execute("ALTER TABLE t1 ADD COLUMN c3 INT"));
-  ASSERT_OK(conn2.Execute("ALTER TABLE t1 ADD COLUMN c4 INT"));
-}
-
 }  // namespace yb

src/yb/master/object_lock_info_manager.cc

@@ -94,8 +94,9 @@ class ObjectLockInfoManager::Impl {
       const tserver::AcquireObjectLockRequestPB& req, rpc::RpcContext context,
       StdStatusCallback callback);
 
+  void PopulateDbCatalogVersionCache(ReleaseObjectLockRequestPB& req);
   void UnlockObject(
-      ReleaseObjectLockRequestPB req, ReleaseObjectLocksGlobalResponsePB* resp,
+      ReleaseObjectLockRequestPB&& req, ReleaseObjectLocksGlobalResponsePB* resp,
       rpc::RpcContext rpc);
   void UnlockObject(
       const tserver::ReleaseObjectLockRequestPB& req, std::optional<rpc::RpcContext> context,
@@ -602,9 +603,35 @@ void ObjectLockInfoManager::Impl::LockObject(
   lock_objects->Launch();
 }
 
+void ObjectLockInfoManager::Impl::PopulateDbCatalogVersionCache(ReleaseObjectLockRequestPB& req) {
+  // TODO: Currently, we fetch and send catalog version of all dbs because the cache invalidation
+  // logic on the tserver side expects a full report. Fix it and then optimize the below to only
+  // send the catalog version of the db being operated on by the txn.
+  DbOidToCatalogVersionMap versions;
+  uint64_t fingerprint;
+  auto s = catalog_manager_->GetYsqlAllDBCatalogVersions(
+      FLAGS_enable_heartbeat_pg_catalog_versions_cache, &versions, &fingerprint);
+  if (!s.ok()) {
+    // In this case, we fallback to delayed cache invalidation on tserver-master heartbeat path.
+    LOG(WARNING) << "Couldn't populate catalog version on exclusive lock release: " << s;
+    return;
+  }
+  if (versions.empty()) {
+    return;
+  }
+  auto* db_catalog_version_data = req.mutable_db_catalog_version_data();
+  for (const auto& it : versions) {
+    auto* const catalog_version_pb = db_catalog_version_data->add_db_catalog_versions();
+    catalog_version_pb->set_db_oid(it.first);
+    catalog_version_pb->set_current_version(it.second.current_version);
+    catalog_version_pb->set_last_breaking_version(it.second.last_breaking_version);
+  }
+}
+
 void ObjectLockInfoManager::Impl::UnlockObject(
-    ReleaseObjectLockRequestPB req, ReleaseObjectLocksGlobalResponsePB* resp,
+    ReleaseObjectLockRequestPB&& req, ReleaseObjectLocksGlobalResponsePB* resp,
     rpc::RpcContext context) {
+  PopulateDbCatalogVersionCache(req);
   UnlockObject(req, std::move(context), std::nullopt, [resp, clock = clock_](const Status& s) {
     resp->set_propagated_hybrid_time(clock->Now().ToUint64());
     FillErrorIfRequired(s, resp);
@@ -636,26 +663,7 @@ void ObjectLockInfoManager::Impl::UnlockObject(const TransactionId& txn_id) {
     req.set_lease_epoch(it->second.lease_epoch);
   }
 
-  // TODO: Currently, we fetch and send catalog version of all dbs because the cache invalidation
-  // logic on the tserver side expects a full report. Fix it and then optimize the below to only
-  // send the catalog version of the db being operated on by the txn.
-  DbOidToCatalogVersionMap versions;
-  uint64_t fingerprint;
-  auto s = catalog_manager_->GetYsqlAllDBCatalogVersions(
-      FLAGS_enable_heartbeat_pg_catalog_versions_cache, &versions, &fingerprint);
-  if (s.ok()) {
-    auto* db_catalog_version_data = req.mutable_db_catalog_version_data();
-    for (const auto& it : versions) {
-      auto* const catalog_version_pb = db_catalog_version_data->add_db_catalog_versions();
-      catalog_version_pb->set_db_oid(it.first);
-      catalog_version_pb->set_current_version(it.second.current_version);
-      catalog_version_pb->set_last_breaking_version(it.second.last_breaking_version);
-    }
-  } else {
-    // In this case, we fallback to delayed cache invalidation on tserver-master heartbeat path.
-    LOG(WARNING) << "Couldn't populate catalog version on exclusive lock release: " << s;
-  }
-
+  PopulateDbCatalogVersionCache(req);
   return UnlockObject(
       std::move(req), std::nullopt /* context */, std::nullopt /* leader epoch */,
       [txn_id](Status s) {
@@ -683,6 +691,11 @@ void ObjectLockInfoManager::Impl::ReleaseLocksHeldByExpiredLeaseEpoch(
         auto txn_id = CHECK_RESULT(TransactionId::FromString(txn_id_str));
         request->set_txn_id(txn_id.data(), txn_id.size());
         request->set_lease_epoch(max_lease_epoch_to_release + 1);
+        if (requests_per_txn.empty()) {
+          // Set the db catalog cache on just one of the unlock requests, since it would be the same
+          // unless a new DDL modified it, in which case it's release would set the latest cache.
+          PopulateDbCatalogVersionCache(*request.get());
+        }
         requests_per_txn.push_back(request);
       }
     }

src/yb/master/ysql_ddl_handler.cc

@@ -49,6 +49,10 @@ DEFINE_test_flag(double, ysql_ddl_rollback_failure_probability, 0.0,
 DEFINE_test_flag(double, ysql_ddl_verification_failure_probability, 0.0,
     "Inject random failure of ddl verification operations");
 
+DEFINE_test_flag(bool, disable_release_object_locks_on_ddl_verification, false,
+    "When set, skip release object lock rpcs to tservers triggered at the end of DDL verification, "
+    "that release object locks acquired by the DDL.");
+
 DECLARE_bool(TEST_enable_object_locking_for_table_locks);
 
 using namespace std::placeholders;
@@ -631,11 +635,14 @@ void CatalogManager::RemoveDdlTransactionStateUnlocked(
       // 1. Either the alter waits inline successfully before issuing the commit,
       // 2. or when the above times out, this branch is involed by the multi step
       //    TableSchemaVerificationTask's callback post the schema changes have been applied.
-      WARN_NOT_OK(
-          background_tasks_thread_pool_->SubmitFunc([this, txn_id]() {
-            DoReleaseObjectLocksIfNecessary(txn_id);
-          }),
-          Format("Failed to submit task for releasing exclusive object locks of txn $0", txn_id));
+      if (FLAGS_TEST_enable_object_locking_for_table_locks &&
+          !FLAGS_TEST_disable_release_object_locks_on_ddl_verification) {
+        WARN_NOT_OK(
+            background_tasks_thread_pool_->SubmitFunc([this, txn_id]() {
+              DoReleaseObjectLocksIfNecessary(txn_id);
+            }),
+            Format("Failed to submit task for releasing exclusive object locks of txn $0", txn_id));
+      }
     } else {
       VLOG(1) << "DDL Verification state for " << txn_id << " has "
               << tables.size() << " tables remaining";
@@ -791,9 +798,6 @@ void CatalogManager::ScheduleTriggerDdlVerificationIfNeeded(
 }
 
 void CatalogManager::DoReleaseObjectLocksIfNecessary(const TransactionId& txn_id) {
-  if (!PREDICT_FALSE(FLAGS_TEST_enable_object_locking_for_table_locks)) {
-    return;
-  }
   DEBUG_ONLY_TEST_SYNC_POINT("DoReleaseObjectLocksIfNecessary");
   object_lock_info_manager_->ReleaseLocksForTxn(txn_id);
 }