Remove dependencies on placement caches from DROP

When dropping objects, citus_drop_trigger should not depend on the
validity of the caches for placements as it causes may cause error
messages from time to time. With this commit we remove the depencency on
the validity on caches for shard placements.

Fixes: #5780

src/backend/distributed/metadata/metadata_cache.c

@@ -1167,6 +1167,39 @@ LookupNodeForGroup(int32 groupId)
 }
 
 
+/*
+ * ShardPlacementListViaCatalog returns the list of placements for the given shard from
+ * the cache.
+ *
+ * This function is intended for DROP operations can potentially drop placements, and
+ * therefore invalidate caches for placements. We continue accessing caches for worker
+ * nodes, and expect that they will not get invalidated by a concurrent process.
+ */
+List *
+ShardPlacementListViaCatalog(uint64 shardId)
+{
+	List *groupShardPlacementList = BuildShardPlacementList(shardId);
+	List *shardPlacementList = NIL;
+
+	GroupShardPlacement *groupShardPlacement = NULL;
+	foreach_ptr(groupShardPlacement, groupShardPlacementList)
+	{
+		WorkerNode *worker = LookupNodeForGroup(groupShardPlacement->groupId);
+		ShardPlacement *placement = CitusMakeNode(ShardPlacement);
+		placement->shardId = groupShardPlacement->shardId;
+		placement->shardLength = groupShardPlacement->shardLength;
+		placement->nodeId = worker->nodeId;
+		placement->nodeName = pstrdup(worker->workerName);
+		placement->nodePort = worker->workerPort;
+		placement->placementId = groupShardPlacement->placementId;
+
+		shardPlacementList = lappend(shardPlacementList, placement);
+	}
+
+	return SortList(shardPlacementList, CompareShardPlacements);
+}
+
+
 /*
  * ShardPlacementList returns the list of placements for the given shard from
  * the cache.

src/backend/distributed/metadata/metadata_utility.c

@@ -1466,7 +1466,7 @@ ActiveShardPlacementListOnGroup(uint64 shardId, int32 groupId)
 
 /*
  * ActiveShardPlacementList finds shard placements for the given shardId from
- * system catalogs, chooses placements that are in active state, and returns
+ * metadata cache, chooses placements that are in active state, and returns
  * these shard placements in a new list.
  */
 List *

src/backend/distributed/operations/delete_protocol.c

@@ -373,7 +373,7 @@ DropTaskList(Oid relationId, char *schemaName, char *relationName,
 		task->dependentTaskList = NULL;
 		task->replicationModel = REPLICATION_MODEL_INVALID;
 		task->anchorShardId = shardId;
-		task->taskPlacementList = ShardPlacementList(shardId);
+		task->taskPlacementList = ShardPlacementListViaCatalog(shardId);
 
 		taskList = lappend(taskList, task);
 	}

src/include/distributed/metadata_cache.h

@@ -175,6 +175,7 @@ extern int32 GetLocalNodeId(void);
 extern void CitusTableCacheFlushInvalidatedEntries(void);
 extern Oid LookupShardRelationFromCatalog(int64 shardId, bool missing_ok);
 extern List * ShardPlacementList(uint64 shardId);
+extern List * ShardPlacementListViaCatalog(uint64 shardId);
 extern void CitusInvalidateRelcacheByRelid(Oid relationId);
 extern void CitusInvalidateRelcacheByShardId(int64 shardId);
 extern void InvalidateForeignKeyGraph(void);

src/test/regress/expected/distributed_types.out

@@ -578,6 +578,22 @@ DETAIL:  "type temp_type" will be created only locally
 CREATE TYPE pg_temp.temp_enum AS ENUM ('one', 'two', 'three');
 WARNING:  "type temp_enum" has dependency on unsupported object "schema pg_temp_xxx"
 DETAIL:  "type temp_enum" will be created only locally
+-- check that dropping a schema that has a type used in a distribution column does not fail
+CREATE SCHEMA schema_with_custom_distribution_type;
+SET search_path TO schema_with_custom_distribution_type;
+CREATE TYPE my_type AS (int_field int);
+CREATE TABLE tbl (a schema_with_custom_distribution_type.my_type);
+SELECT create_distributed_table('tbl', 'a');
+ create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+drop schema schema_with_custom_distribution_type cascade;
+NOTICE:  drop cascades to 2 other objects
+DETAIL:  drop cascades to type my_type
+drop cascades to table tbl
+SET search_path TO type_tests;
 -- clear objects
 SET client_min_messages TO error; -- suppress cascading objects dropping
 DROP SCHEMA type_tests CASCADE;

src/test/regress/sql/distributed_types.sql

@@ -350,6 +350,19 @@ SELECT create_distributed_table('table_text_local_def','id');
 CREATE TYPE pg_temp.temp_type AS (int_field int);
 CREATE TYPE pg_temp.temp_enum AS ENUM ('one', 'two', 'three');
 
+-- check that dropping a schema that has a type used in a distribution column does not fail
+
+CREATE SCHEMA schema_with_custom_distribution_type;
+SET search_path TO schema_with_custom_distribution_type;
+
+CREATE TYPE my_type AS (int_field int);
+CREATE TABLE tbl (a schema_with_custom_distribution_type.my_type);
+SELECT create_distributed_table('tbl', 'a');
+
+drop schema schema_with_custom_distribution_type cascade;
+
+SET search_path TO type_tests;
+
 -- clear objects
 SET client_min_messages TO error; -- suppress cascading objects dropping
 DROP SCHEMA type_tests CASCADE;