[#29215] DocDB: Fix txn_start_us DFATAL for DDL entering tablet wait-queue

Summary:
Earlier, ysql was propagating `pg_txn_start_us` for regular transactions ('kPlain`) alone, which was being used at the tablet wait-queue for prioritizing waiter resumption. This was being done for DMLs alone with the assumption that DDLs don't enter the wait-queue and instead abort active transactions (with object locking disabled). Refer https://phorge.dev.yugabyte.com/D28756 for details.

But this assumption doesn't seem to hold true with colocated database with index, as the following drop table results in a write query to the index table.
```
$CREATE DATABASE db with COLOCATION = true;
$\c db;
$CREATE TABLE test (k INT, v INT, PRIMARY KEY (k ASC));
$CREATE UNIQUE INDEX idx_1 ON test (v ASC);
$DROP INDEX idx_1;
```
and when there are active DML transactions operating at the table, the drop DDL enters the wait-queue, which leads to the below DFATAL
```
[ts-1] F1101 11:02:59.304080 661742 wait_queue.cc:268] TxnId: 6557ecea-23bc-44b9-8d55-6c264ae0b958, ReqId: 612 Expected non-zero txn_start_us
```

This revision addresses the issue by propagating the pg_txn_start_us value from ysql to docdb, and initializes `YBTransaction` with the same so that the write rpc picks this up, and the tablet wait-queue eventually honors the waiter resumption logic.

Note: With object locking enabled, the DFATAL wasn't getting triggered earlier, since the drop table gets gated the the object lock manager layer itself, and doesn't proceed to the tablet layer until the active DMLs finish. And once the active DMLs finish, it wouldn't enter the tablet wait-queue anyways.
Jira: DB-18983

Test Plan:
Jenkins

```
./yb_build.sh --cxx-test='TEST_F(PgWaitQueuesTestWithoutObjectLocking, TestDDLHaveWaitStartTimeSet)'
```

Reviewers: rthallam, sergei, pjain, patnaik.balivada

Reviewed By: pjain, patnaik.balivada

Subscribers: yql, ybase

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

Author: basavaraj29

src/yb/tserver/pg_client_session.cc

@@ -3082,7 +3082,7 @@ class PgClientSession::Impl {
       EnsureSession(kind, deadline);
       RETURN_NOT_OK(GetDdlTransactionMetadata(
           true /* use_transaction */, false /* use_regular_transaction_block */, deadline,
-          options.priority()));
+          options.priority(), options.pg_txn_start_us()));
     } else {
       DCHECK(kind == PgClientSessionKind::kPlain);
       auto& session = EnsureSession(kind, deadline);
@@ -3332,7 +3332,8 @@ class PgClientSession::Impl {
   // All DDLs use kHighestPriority unless specified otherwise.
   Result<const TransactionMetadata*> GetDdlTransactionMetadata(
       bool use_transaction, bool use_regular_transaction_block, CoarseTimePoint deadline,
-      uint64_t priority = kHighPriTxnUpperBound) {
+      uint64_t priority = kHighPriTxnUpperBound,
+      uint64_t pg_txn_start_us = 0) {
     if (!use_transaction) {
       return nullptr;
     }
@@ -3360,6 +3361,8 @@ class PgClientSession::Impl {
       const auto isolation = FLAGS_ysql_serializable_isolation_for_ddl_txn
           ? IsolationLevel::SERIALIZABLE_ISOLATION : IsolationLevel::SNAPSHOT_ISOLATION;
       txn = transaction_provider_.Take<PgClientSessionKind::kDdl>(deadline);
+      RETURN_NOT_OK(txn->SetPgTxnStart(
+          pg_txn_start_us ? pg_txn_start_us : MonoTime::Now().ToUint64()));
       RETURN_NOT_OK(txn->Init(isolation));
       txn->SetPriority(priority);
       txn->SetLogPrefixTag(kTxnLogPrefixTag, id_);

src/yb/yql/pggate/pg_txn_manager.cc

@@ -744,6 +744,7 @@ Status PgTxnManager::SetupPerformOptions(
   options->set_txn_serial_no(serial_no_.txn());
   options->set_active_sub_transaction_id(active_sub_transaction_id_);
   options->set_xcluster_target_ddl_bypass(yb_xcluster_target_ddl_bypass);
+  options->set_pg_txn_start_us(pg_txn_start_us_);
 
   if (use_saved_priority_) {
     options->set_use_existing_priority(true);
@@ -770,8 +771,6 @@ Status PgTxnManager::SetupPerformOptions(
     options->set_read_time_manipulation(
         GetActualReadTimeManipulator(isolation_level_, read_time_manipulation_, read_time_action));
     read_time_manipulation_ = tserver::ReadTimeManipulation::NONE;
-    // pg_txn_start_us is similarly only used for kPlain transactions.
-    options->set_pg_txn_start_us(pg_txn_start_us_);
     // Only clamp read-only txns/stmts.
     // Do not clamp in the serializable case since
     // - SERIALIZABLE reads do not pick read time until later.

src/yb/yql/pgwrapper/pg_wait_on_conflict-test.cc

@@ -1880,5 +1880,36 @@ TEST_F(PgWaitQueuesWithRetriesTest, TestDeadlockRetries) {
   thread_holder.WaitAndStop(30s * kTimeMultiplier);
 }
 
+#ifndef NDEBUG
+TEST_F(PgWaitQueuesTestWithoutObjectLocking, TestDDLHaveWaitStartTimeSet) {
+  constexpr auto kDbName = "colodb";
+  auto setup_conn = ASSERT_RESULT(Connect());
+  ASSERT_OK(setup_conn.ExecuteFormat("CREATE DATABASE $0 with COLOCATION = true", kDbName));
+
+  auto conn = ASSERT_RESULT(ConnectToDB(kDbName));
+  ASSERT_OK(conn.Execute(
+      "CREATE TABLE test (k INT, v INT, PRIMARY KEY (k ASC))"));
+  ASSERT_OK(conn.Execute("CREATE UNIQUE INDEX idx_1 ON test (v ASC)"));
+  ASSERT_OK(conn.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+  ASSERT_OK(conn.Execute("INSERT INTO test VALUES (1, -1)"));
+
+  yb::SyncPoint::GetInstance()->LoadDependency({
+    {"WaitQueue::Impl::SetupWaiterUnlocked:1", "TestDDLHaveWaitStartTimeSet"}});
+  yb::SyncPoint::GetInstance()->ClearTrace();
+  yb::SyncPoint::GetInstance()->EnableProcessing();
+
+  TestThreadHolder thread_holder;
+  thread_holder.AddThreadFunctor([this, &stop = thread_holder.stop_flag()] {
+    auto conn = ASSERT_RESULT(ConnectToDB(kDbName));
+    ASSERT_OK(conn.Execute("SET yb_max_query_layer_retries=10"));
+    ASSERT_OK(conn.Execute("DROP INDEX idx_1"));
+  });
+  DEBUG_ONLY_TEST_SYNC_POINT("TestDDLHaveWaitStartTimeSet");
+  ASSERT_OK(conn.CommitTransaction());
+
+  thread_holder.WaitAndStop(10s * kTimeMultiplier);
+}
+#endif
+
 } // namespace pgwrapper
 } // namespace yb