Properly handle error at owner check (#6984)

We did not properly handle the error at ownership check method, which
causes `max stack depth for errors` as in
https://github.com/citusdata/citus/issues/6980.

**Fix:**
In case of an error, we should rollback subtransaction and throw the
message with log level to `LOG_SERVER_ONLY`.

Note: We prevent logs from the client to prevent pg vanilla test
failures due to Citus logs which differs from the actual Postgres logs.
(For context: https://github.com/citusdata/citus/pull/6130)

I also needed to fix a flaky test: `multi_schema_support`

DESCRIPTION: Fixes a bug related to non-existent objects in DDL
commands.

Fixes https://github.com/citusdata/citus/issues/6980

(cherry picked from commit 565c5260fd)

src/backend/distributed/utils/citus_depended_object.c

@@ -462,21 +462,25 @@ HasDropCommandViolatesOwnership(Node *node)
 static bool
 AnyObjectViolatesOwnership(DropStmt *dropStmt)
 {
+	bool hasOwnershipViolation = false;
 	volatile ObjectAddress objectAddress = { 0 };
 	Relation relation = NULL;
-	bool objectViolatesOwnership = false;
 	ObjectType objectType = dropStmt->removeType;
 	bool missingOk = dropStmt->missing_ok;
 
-	Node *object = NULL;
-	foreach_ptr(object, dropStmt->objects)
+	MemoryContext savedContext = CurrentMemoryContext;
+	ResourceOwner savedOwner = CurrentResourceOwner;
+	BeginInternalSubTransaction(NULL);
+	MemoryContextSwitchTo(savedContext);
+
+	PG_TRY();
 	{
-		PG_TRY();
+		Node *object = NULL;
+		foreach_ptr(object, dropStmt->objects)
 		{
 			objectAddress = get_object_address(objectType, object,
 											   &relation, AccessShareLock, missingOk);
 
-
 			if (OidIsValid(objectAddress.objectId))
 			{
 				/*
@@ -487,29 +491,39 @@ AnyObjectViolatesOwnership(DropStmt *dropStmt)
 									   objectAddress,
 									   object, relation);
 			}
-		}
-		PG_CATCH();
-		{
-			if (OidIsValid(objectAddress.objectId))
+
+			if (relation != NULL)
 			{
-				/* ownership violation */
-				objectViolatesOwnership = true;
+				relation_close(relation, NoLock);
+				relation = NULL;
 			}
 		}
-		PG_END_TRY();
 
+		ReleaseCurrentSubTransaction();
+		MemoryContextSwitchTo(savedContext);
+		CurrentResourceOwner = savedOwner;
+	}
+	PG_CATCH();
+	{
+		MemoryContextSwitchTo(savedContext);
+		ErrorData *edata = CopyErrorData();
+		FlushErrorState();
+
+		hasOwnershipViolation = true;
 		if (relation != NULL)
 		{
-			relation_close(relation, AccessShareLock);
+			relation_close(relation, NoLock);
 			relation = NULL;
 		}
+		RollbackAndReleaseCurrentSubTransaction();
+		MemoryContextSwitchTo(savedContext);
+		CurrentResourceOwner = savedOwner;
 
-		/* we found ownership violation, so can return here */
-		if (objectViolatesOwnership)
-		{
-			return true;
-		}
+		/* Rethrow error with LOG_SERVER_ONLY to prevent log to be sent to client */
+		edata->elevel = LOG_SERVER_ONLY;
+		ThrowErrorData(edata);
 	}
+	PG_END_TRY();
 
-	return false;
+	return hasOwnershipViolation;
 }

src/test/regress/expected/multi_schema_support.out

@@ -1095,6 +1095,9 @@ ALTER TABLE IF EXISTS non_existent_table SET SCHEMA non_existent_schema;
 NOTICE:  relation "non_existent_table" does not exist, skipping
 DROP SCHEMA existing_schema, another_existing_schema CASCADE;
 NOTICE:  drop cascades to table existing_schema.table_set_schema
+-- test DROP SCHEMA with nonexisting schemas
+DROP SCHEMA ax, bx, cx, dx, ex, fx, gx, jx;
+ERROR:  schema "ax" does not exist
 -- test ALTER TABLE SET SCHEMA with interesting names
 CREATE SCHEMA "cItuS.T E E N'sSchema";
 CREATE SCHEMA "citus-teen's scnd schm.";
@@ -1361,6 +1364,7 @@ SELECT pg_identify_object_as_address(classid, objid, objsubid) FROM pg_catalog.p
  (schema,{run_test_schema},{})
 (1 row)
 
+DROP TABLE public.nation_local;
 DROP SCHEMA run_test_schema, test_schema_support_join_1, test_schema_support_join_2, "Citus'Teen123", "CiTUS.TEEN2", bar, test_schema_support CASCADE;
 -- verify that the dropped schema is removed from worker's pg_dist_object
 SELECT pg_identify_object_as_address(classid, objid, objsubid) FROM pg_catalog.pg_dist_object

src/test/regress/sql/multi_schema_support.sql

@@ -802,6 +802,9 @@ ALTER TABLE IF EXISTS non_existent_table SET SCHEMA non_existent_schema;
 DROP SCHEMA existing_schema, another_existing_schema CASCADE;
 
 
+-- test DROP SCHEMA with nonexisting schemas
+DROP SCHEMA ax, bx, cx, dx, ex, fx, gx, jx;
+
 -- test ALTER TABLE SET SCHEMA with interesting names
 CREATE SCHEMA "cItuS.T E E N'sSchema";
 CREATE SCHEMA "citus-teen's scnd schm.";
@@ -968,6 +971,7 @@ SET client_min_messages TO WARNING;
 
 SELECT pg_identify_object_as_address(classid, objid, objsubid) FROM pg_catalog.pg_dist_object
     WHERE classid=2615 and objid IN (select oid from pg_namespace where nspname='run_test_schema');
+DROP TABLE public.nation_local;
 DROP SCHEMA run_test_schema, test_schema_support_join_1, test_schema_support_join_2, "Citus'Teen123", "CiTUS.TEEN2", bar, test_schema_support CASCADE;
 -- verify that the dropped schema is removed from worker's pg_dist_object
 SELECT pg_identify_object_as_address(classid, objid, objsubid) FROM pg_catalog.pg_dist_object