[#9645]: YSQL: Cleanup DDL txn state in case of query failure

Summary:
DDL transaction state is controlled by the `ddl_nesting_level` counter.
When `ddl_nesting_level` is changing from 0 to 1 by the calling of the `YBIncrementDdlNestingLevel` function new DDL transaction is started. When `ddl_nesting_level` is changing from 1 to 0 by the calling of the `YBDecrementDdlNestingLevel` function current DDL transaction is committed.

In some scenarios in case of failure the `ddl_nesting_level` counter may have inappropriate value, it is required to reset them and DDL transaction state as well.

One of such scenario is the index creation procedure. By design index creation commits current DDL transaction at the middle and starts a new one immediately. For this purpose the `YBDecrementDdlNestingLevel`/`YBIncrementDdlNestingLevel` functions are called from the `DefineIndex` function. But in case current DDL is failed to commit or new DDL transaction can't be started for some reason (due to catalog version change etc) the `YBIncrementDdlNestingLevel` function may not be called or may return error. In this case `ddl_nesting_level` will have wrong value. As a result current SQL session will not be able to start/commit further DDLs at the proper time.

Solution is to reset DDL transaction state in case current DDL query finished with an error.

Note: As far as DDL transaction state is reset in case of error it is not necessary to have `success` parameter in the `YBDecrementDdlNestingLevel` function (it is called only in case of success)

Test Plan:
New unit test was introduced

```
./yb_build.sh --gtest_filter PgDDLConcurrencyTest.IndexCreation
```

Reviewers: mihnea, jason

Reviewed By: jason

Subscribers: rsami, yql

Differential Revision: https://phabricator.dev.yugabyte.com/D12609

src/postgres/src/backend/commands/indexcmds.c

@@ -1359,8 +1359,7 @@ DefineIndex(Oid relationId,
 	 * TODO(jason): handle nested CREATE INDEX (this assumes we're at nest
 	 * level 1).
 	 */
-	YBDecrementDdlNestingLevel(true /* success */,
-	                           true /* is_catalog_version_increment */,
+	YBDecrementDdlNestingLevel(true /* is_catalog_version_increment */,
 	                           false /* is_breaking_catalog_change */);
 	CommitTransactionCommand();
 
@@ -1382,8 +1381,7 @@ DefineIndex(Oid relationId,
 	 * TODO(jason): handle nested CREATE INDEX (this assumes we're at nest
 	 * level 1).
 	 */
-	YBDecrementDdlNestingLevel(true /* success */,
-	                           true /* is_catalog_version_increment */,
+	YBDecrementDdlNestingLevel(true /* is_catalog_version_increment */,
 	                           false /* is_breaking_catalog_change */);
 	CommitTransactionCommand();
 

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

@@ -532,7 +532,7 @@ YBCAbortTransaction()
 		return;
 
 	if (YBTransactionsEnabled())
-		HandleYBStatus(YBCPgAbortTransaction());
+		YBCPgAbortTransaction();
 }
 
 void
@@ -941,6 +941,13 @@ YBIsInitDbAlreadyDone()
 static ProcessUtility_hook_type prev_ProcessUtility = NULL;
 static int ddl_nesting_level = 0;
 
+static void
+YBResetDdlState()
+{
+	ddl_nesting_level = 0;
+	YBCPgClearSeparateDdlTxnMode();
+}
+
 int
 YBGetDdlNestingLevel()
 {
@@ -958,20 +965,15 @@ YBIncrementDdlNestingLevel()
 }
 
 void
-YBDecrementDdlNestingLevel(bool success,
-                           bool is_catalog_version_increment,
-                           bool is_breaking_catalog_change)
+YBDecrementDdlNestingLevel(bool is_catalog_version_increment, bool is_breaking_catalog_change)
 {
 	ddl_nesting_level--;
 	if (ddl_nesting_level == 0)
 	{
-		bool increment_done = false;
-		if (success && is_catalog_version_increment)
-		{
-			increment_done = YbIncrementMasterCatalogVersionTableEntry(is_breaking_catalog_change);
-		}
+		const bool increment_done = is_catalog_version_increment &&
+			YbIncrementMasterCatalogVersionTableEntry(is_breaking_catalog_change);
 
-		HandleYBStatus(YBCPgExitSeparateDdlTxnMode(success));
+		HandleYBStatus(YBCPgExitSeparateDdlTxnMode());
 
 		/*
 		 * Optimization to avoid redundant cache refresh on the current session
@@ -985,28 +987,25 @@ YBDecrementDdlNestingLevel(bool success,
 			yb_catalog_cache_version += 1;
 		}
 
-		if (success)
+		List *handles = YBGetDdlHandles();
+		ListCell *lc = NULL;
+		foreach(lc, handles)
 		{
-			List *handles = YBGetDdlHandles();
-			ListCell *lc = NULL;
-			foreach(lc, handles)
-			{
-				YBCPgStatement handle = (YBCPgStatement) lfirst(lc);
-				/*
-				 * At this point we have already applied the DDL in the YSQL layer and
-				 * executing the postponed DocDB statement is not strictly required.
-				 * Ignore 'NotFound' because DocDB might already notice applied DDL.
-				 * See comment for YBGetDdlHandles in xact.h for more details.
-				 */
-				YBCStatus status = YBCPgExecPostponedDdlStmt(handle);
-				if (YBCStatusIsNotFound(status)) {
-					YBCFreeStatus(status);
-				} else {
-					HandleYBStatusAtErrorLevel(status, WARNING);
-				}
+			YBCPgStatement handle = (YBCPgStatement) lfirst(lc);
+			/*
+			 * At this point we have already applied the DDL in the YSQL layer and
+			 * executing the postponed DocDB statement is not strictly required.
+			 * Ignore 'NotFound' because DocDB might already notice applied DDL.
+			 * See comment for YBGetDdlHandles in xact.h for more details.
+			 */
+			YBCStatus status = YBCPgExecPostponedDdlStmt(handle);
+			if (YBCStatusIsNotFound(status)) {
+				YBCFreeStatus(status);
+			} else {
+				HandleYBStatusAtErrorLevel(status, WARNING);
 			}
-			YBClearDdlHandles();
 		}
+		YBClearDdlHandles();
 	}
 }
 
@@ -1328,17 +1327,17 @@ static void YBTxnDdlProcessUtility(
 	PG_CATCH();
 	{
 		if (is_txn_ddl) {
-			YBDecrementDdlNestingLevel(/* success */ false,
-			                           is_catalog_version_increment,
-			                           is_breaking_catalog_change);
+			/*
+			 * It is possible that ddl_nesting_level has wrong value due to error.
+			 * Ddl transaction state should be reset.
+			 */
+			YBResetDdlState();
 		}
 		PG_RE_THROW();
 	}
 	PG_END_TRY();
 	if (is_txn_ddl) {
-		YBDecrementDdlNestingLevel(/* success */ true,
-		                           is_catalog_version_increment,
-		                           is_breaking_catalog_change);
+		YBDecrementDdlNestingLevel(is_catalog_version_increment, is_breaking_catalog_change);
 	}
 }
 

src/postgres/src/include/pg_yb_utils.h

@@ -444,8 +444,7 @@ bool YBIsInitDbAlreadyDone();
 
 int YBGetDdlNestingLevel();
 void YBIncrementDdlNestingLevel();
-void YBDecrementDdlNestingLevel(bool success,
-                                bool is_catalog_version_increment,
+void YBDecrementDdlNestingLevel(bool is_catalog_version_increment,
                                 bool is_breaking_catalog_change);
 bool IsTransactionalDdlStatement(PlannedStmt *pstmt,
                                  bool *is_catalog_version_increment,

src/yb/yql/pggate/pg_txn_manager.cc

@@ -349,26 +349,25 @@ Status PgTxnManager::CommitTransaction() {
   return status;
 }
 
-Status PgTxnManager::AbortTransaction() {
+void PgTxnManager::AbortTransaction() {
   // If a DDL operation during a DDL txn fails the txn will be aborted before we get here.
   // However if there are failures afterwards (i.e. during COMMIT or catalog version increment),
   // then we might get here with a ddl_txn_. Clean it up in that case.
   if (ddl_txn_) {
-    RETURN_NOT_OK(ExitSeparateDdlTxnMode(false));
+    ClearSeparateDdlTxnMode();
   }
 
   if (!txn_in_progress_) {
-    return Status::OK();
+    return;
   }
   if (!txn_) {
     // This was a read-only transaction, nothing to commit.
     ResetTxnAndSession();
-    return Status::OK();
+    return;
   }
   // TODO: how do we report errors if the transaction has already committed?
   txn_->Abort();
   ResetTxnAndSession();
-  return Status::OK();
 }
 
 // TODO: dedup with similar logic in CQLServiceImpl.
@@ -427,19 +426,24 @@ Status PgTxnManager::EnterSeparateDdlTxnMode() {
   return Status::OK();
 }
 
-Status PgTxnManager::ExitSeparateDdlTxnMode(bool is_success) {
-  VLOG_TXN_STATE(2) << "is_success=" << is_success;
+Status PgTxnManager::ExitSeparateDdlTxnMode() {
+  VLOG_TXN_STATE(2);
   RSTATUS_DCHECK(
       ddl_txn_ != nullptr,
       IllegalState, "ExitSeparateDdlTxnMode called when not in a DDL transaction");
-  if (is_success) {
-    RETURN_NOT_OK(ddl_txn_->CommitFuture().get());
-  } else {
+  RETURN_NOT_OK(ddl_txn_->CommitFuture().get());
+  ddl_txn_.reset();
+  ddl_session_.reset();
+  return Status::OK();
+}
+
+void PgTxnManager::ClearSeparateDdlTxnMode() {
+  VLOG_TXN_STATE(2);
+  if (ddl_txn_) {
     ddl_txn_->Abort();
   }
   ddl_txn_.reset();
   ddl_session_.reset();
-  return Status::OK();
 }
 
 std::string PgTxnManager::TxnStateDebugStr() const {

src/yb/yql/pggate/pg_txn_manager.h

@@ -65,12 +65,13 @@ class PgTxnManager : public RefCountedThreadSafe<PgTxnManager> {
   CHECKED_STATUS RecreateTransaction();
   CHECKED_STATUS RestartTransaction();
   CHECKED_STATUS CommitTransaction();
-  CHECKED_STATUS AbortTransaction();
+  void AbortTransaction();
   CHECKED_STATUS SetIsolationLevel(int isolation);
   CHECKED_STATUS SetReadOnly(bool read_only);
   CHECKED_STATUS SetDeferrable(bool deferrable);
   CHECKED_STATUS EnterSeparateDdlTxnMode();
-  CHECKED_STATUS ExitSeparateDdlTxnMode(bool success);
+  CHECKED_STATUS ExitSeparateDdlTxnMode();
+  void ClearSeparateDdlTxnMode();
 
   // Returns the transactional session, starting a new transaction if necessary.
   yb::Result<client::YBSession*> GetTransactionalSession();

src/yb/yql/pggate/pggate.cc

@@ -1380,10 +1380,10 @@ Status PgApiImpl::CommitTransaction() {
   return pg_txn_manager_->CommitTransaction();
 }
 
-Status PgApiImpl::AbortTransaction() {
+void PgApiImpl::AbortTransaction() {
   pg_session_->InvalidateForeignKeyReferenceCache();
   pg_session_->DropBufferedOperations();
-  return pg_txn_manager_->AbortTransaction();
+  pg_txn_manager_->AbortTransaction();
 }
 
 Status PgApiImpl::SetTransactionIsolationLevel(int isolation) {
@@ -1404,23 +1404,20 @@ Status PgApiImpl::EnterSeparateDdlTxnMode() {
   return pg_txn_manager_->EnterSeparateDdlTxnMode();
 }
 
-Status PgApiImpl::ExitSeparateDdlTxnMode(bool success) {
+Status PgApiImpl::ExitSeparateDdlTxnMode() {
   // Flush all buffered operations as ddl txn use its own transaction session.
-  if (success) {
-    RETURN_NOT_OK(pg_session_->FlushBufferedOperations());
-  } else {
-    pg_session_->DropBufferedOperations();
-  }
-
-  RETURN_NOT_OK(pg_txn_manager_->ExitSeparateDdlTxnMode(success));
-  ReadHybridTime read_time;
-  if (success) {
-    // Next reads from catalog tables have to see changes made by the DDL transaction.
-    ResetCatalogReadTime();
-  }
+  RETURN_NOT_OK(pg_session_->FlushBufferedOperations());
+  RETURN_NOT_OK(pg_txn_manager_->ExitSeparateDdlTxnMode());
+  // Next reads from catalog tables have to see changes made by the DDL transaction.
+  ResetCatalogReadTime();
   return Status::OK();
 }
 
+void PgApiImpl::ClearSeparateDdlTxnMode() {
+  pg_session_->DropBufferedOperations();
+  pg_txn_manager_->ClearSeparateDdlTxnMode();
+}
+
 Status PgApiImpl::SetActiveSubTransaction(SubTransactionId id) {
   return pg_session_->SetActiveSubTransaction(id);
 }

src/yb/yql/pggate/pggate.h

@@ -469,12 +469,13 @@ class PgApiImpl {
   CHECKED_STATUS RecreateTransaction();
   CHECKED_STATUS RestartTransaction();
   CHECKED_STATUS CommitTransaction();
-  CHECKED_STATUS AbortTransaction();
+  void AbortTransaction();
   CHECKED_STATUS SetTransactionIsolationLevel(int isolation);
   CHECKED_STATUS SetTransactionReadOnly(bool read_only);
   CHECKED_STATUS SetTransactionDeferrable(bool deferrable);
   CHECKED_STATUS EnterSeparateDdlTxnMode();
-  CHECKED_STATUS ExitSeparateDdlTxnMode(bool success);
+  CHECKED_STATUS ExitSeparateDdlTxnMode();
+  void ClearSeparateDdlTxnMode();
   CHECKED_STATUS SetActiveSubTransaction(SubTransactionId id);
   CHECKED_STATUS RollbackSubTransaction(SubTransactionId id);
 

src/yb/yql/pggate/test/pggate_test.cc

@@ -190,7 +190,7 @@ void PggateTest::BeginDDLTransaction() {
 }
 
 void PggateTest::CommitDDLTransaction() {
-  CHECK_YBC_STATUS(YBCPgExitSeparateDdlTxnMode(true /* success */));
+  CHECK_YBC_STATUS(YBCPgExitSeparateDdlTxnMode());
 }
 
 void PggateTest::BeginTransaction() {

src/yb/yql/pggate/ybc_pggate.cc

@@ -888,8 +888,8 @@ YBCStatus YBCPgCommitTransaction() {
   return ToYBCStatus(pgapi->CommitTransaction());
 }
 
-YBCStatus YBCPgAbortTransaction() {
-  return ToYBCStatus(pgapi->AbortTransaction());
+void YBCPgAbortTransaction() {
+  pgapi->AbortTransaction();
 }
 
 YBCStatus YBCPgSetTransactionIsolationLevel(int isolation) {
@@ -908,8 +908,12 @@ YBCStatus YBCPgEnterSeparateDdlTxnMode() {
   return ToYBCStatus(pgapi->EnterSeparateDdlTxnMode());
 }
 
-YBCStatus YBCPgExitSeparateDdlTxnMode(bool success) {
-  return ToYBCStatus(pgapi->ExitSeparateDdlTxnMode(success));
+YBCStatus YBCPgExitSeparateDdlTxnMode() {
+  return ToYBCStatus(pgapi->ExitSeparateDdlTxnMode());
+}
+
+void YBCPgClearSeparateDdlTxnMode() {
+  pgapi->ClearSeparateDdlTxnMode();
 }
 
 YBCStatus YBCPgSetActiveSubTransaction(uint32_t id) {

src/yb/yql/pggate/ybc_pggate.h

@@ -428,12 +428,13 @@ YBCStatus YBCPgBeginTransaction();
 YBCStatus YBCPgRecreateTransaction();
 YBCStatus YBCPgRestartTransaction();
 YBCStatus YBCPgCommitTransaction();
-YBCStatus YBCPgAbortTransaction();
+void YBCPgAbortTransaction();
 YBCStatus YBCPgSetTransactionIsolationLevel(int isolation);
 YBCStatus YBCPgSetTransactionReadOnly(bool read_only);
 YBCStatus YBCPgSetTransactionDeferrable(bool deferrable);
 YBCStatus YBCPgEnterSeparateDdlTxnMode();
-YBCStatus YBCPgExitSeparateDdlTxnMode(bool success);
+YBCStatus YBCPgExitSeparateDdlTxnMode();
+void YBCPgClearSeparateDdlTxnMode();
 YBCStatus YBCPgSetActiveSubTransaction(uint32_t id);
 YBCStatus YBCPgRollbackSubTransaction(uint32_t id);
 

src/yb/yql/pgwrapper/CMakeLists.txt

@@ -56,6 +56,7 @@ ADD_YB_LIBRARY(ysql_upgrade
 
 set(YB_TEST_LINK_LIBS yb_pgwrapper yb_client ql-dml-test-base pg_wrapper_test_base
    ${YB_MIN_TEST_LIBS})
+ADD_YB_TEST(pg_ddl_concurrency-test)
 ADD_YB_TEST(pg_index_backfill-test)
 ADD_YB_TEST(pg_libpq-test)
 ADD_YB_TEST(pg_libpq_err-test)