Skip distributed schema insertion into pg_dist_schema, if already exists (#7044)

Inserting into `pg_dist_schema` causes unexpected duplicate key errors,
for distributed schemas that already exist. With this commit we skip the
insertion if the schema already exists in `pg_dist_schema`.

The error:
```sql
SET citus.enable_schema_based_sharding TO ON;
CREATE SCHEMA sc2;
CREATE SCHEMA IF NOT EXISTS sc2;
NOTICE:  schema "sc2" already exists, skipping
ERROR:  duplicate key value violates unique constraint "pg_dist_schema_pkey"
DETAIL:  Key (schemaid)=(17294) already exists.
```

fixes: #7042

src/backend/distributed/commands/schema.c

@@ -103,20 +103,29 @@ PostprocessCreateSchemaStmt(Node *node, const char *queryString)
 		}
 
 		/*
-		 * Register the tenant schema on the coordinator and save the command
+		 * Skip if the schema is already inserted into pg_dist_schema.
-		 * to register it on the workers.
+		 * This could occur when trying to create an already existing schema,
+		 * with IF NOT EXISTS clause.
 		 */
-		int shardCount = 1;
+		if (!IsTenantSchema(schemaId))
-		int replicationFactor = 1;
+		{
-		Oid distributionColumnType = InvalidOid;
+			/*
-		Oid distributionColumnCollation = InvalidOid;
+			 * Register the tenant schema on the coordinator and save the command
-		uint32 colocationId = CreateColocationGroup(
+			 * to register it on the workers.
-			shardCount, replicationFactor, distributionColumnType,
+			 */
-			distributionColumnCollation);
+			int shardCount = 1;
+			int replicationFactor = 1;
+			Oid distributionColumnType = InvalidOid;
+			Oid distributionColumnCollation = InvalidOid;
+			uint32 colocationId = CreateColocationGroup(
+				shardCount, replicationFactor, distributionColumnType,
+				distributionColumnCollation);
 
-		InsertTenantSchemaLocally(schemaId, colocationId);
+			InsertTenantSchemaLocally(schemaId, colocationId);
 
-		commands = lappend(commands, TenantSchemaInsertCommand(schemaId, colocationId));
+			commands = lappend(commands, TenantSchemaInsertCommand(schemaId,
+																   colocationId));
+		}
 	}
 
 	commands = lappend(commands, ENABLE_DDL_PROPAGATION);

src/test/regress/expected/schema_based_sharding.out

@@ -50,6 +50,8 @@ SELECT COUNT(*)=0 FROM pg_dist_schema WHERE schemaid::regnamespace::text = 'regu
 
 -- empty tenant
 CREATE SCHEMA "tenant\'_1";
+CREATE SCHEMA IF NOT EXISTS "tenant\'_1";
+NOTICE:  schema "tenant\'_1" already exists, skipping
 -- non-empty tenant
 CREATE SCHEMA "tenant\'_2";
 CREATE TABLE "tenant\'_2".test_table(a int, b text);

src/test/regress/sql/schema_based_sharding.sql

@@ -33,6 +33,7 @@ SELECT COUNT(*)=0 FROM pg_dist_schema WHERE schemaid::regnamespace::text = 'regu
 
 -- empty tenant
 CREATE SCHEMA "tenant\'_1";
+CREATE SCHEMA IF NOT EXISTS "tenant\'_1";
 
 -- non-empty tenant
 CREATE SCHEMA "tenant\'_2";