diff --git a/src/backend/distributed/commands/alter_table.c b/src/backend/distributed/commands/alter_table.c
index d574e997a..5fa670ab2 100644
--- a/src/backend/distributed/commands/alter_table.c
+++ b/src/backend/distributed/commands/alter_table.c
@@ -248,7 +248,8 @@ undistribute_table(PG_FUNCTION_ARGS)
 
 	TableConversionParameters params = {
 		.relationId = relationId,
-		.cascadeViaForeignKeys = cascadeViaForeignKeys
+		.cascadeViaForeignKeys = cascadeViaForeignKeys,
+		.bypassTenantCheck = false
 	};
 
 	UndistributeTable(&params);
@@ -429,6 +430,55 @@ UndistributeTables(List *relationIdList)
 }
 
 
+/*
+ * EnsureUndistributeTenantTableSafe ensures that it is safe to undistribute a tenant table.
+ */
+void
+EnsureUndistributeTenantTableSafe(Oid relationId, const char *operationName)
+{
+	Oid schemaId = get_rel_namespace(relationId);
+	Assert(IsTenantSchema(schemaId));
+
+	/* We only allow undistribute while altering schema */
+	if (strcmp(operationName, TenantOperationNames[TENANT_SET_SCHEMA]) != 0)
+	{
+		ErrorIfTenantTable(relationId, operationName);
+	}
+
+	char *tableName = get_rel_name(relationId);
+	char *schemaName = get_namespace_name(schemaId);
+
+	/*
+	 * Partition table cannot be undistributed. Otherwise, its parent table would still
+	 * be a tenant table whereas partition table would be a local table.
+	 */
+	if (PartitionTable(relationId))
+	{
+		ereport(ERROR, (errmsg("%s is not allowed for partition table %s in distributed "
+							   "schema %s", operationName, tableName, schemaName),
+						errdetail("partition table should be under the same distributed "
+								  "schema as its parent and be a tenant table.")));
+	}
+
+	/*
+	 * When table is referenced by or referencing to a table in the same tenant
+	 * schema, we should disallow undistributing the table since we do not allow
+	 * foreign keys from/to Citus local or Postgres local table to/from distributed
+	 * schema.
+	 */
+	List *fkeyCommandsWithSingleShardTables =
+		GetFKeyCreationCommandsRelationInvolvedWithTableType(
+			relationId, INCLUDE_SINGLE_SHARD_TABLES);
+	if (fkeyCommandsWithSingleShardTables != NIL)
+	{
+		ereport(ERROR, (errmsg("%s is not allowed for table %s in distributed schema %s",
+							   operationName, tableName, schemaName),
+						errdetail("distributed schemas cannot have foreign keys from/to "
+								  "local tables or different schema")));
+	}
+}
+
+
 /*
  * UndistributeTable undistributes the given table. It uses ConvertTable function to
  * create a new local table and move everything to that table.
@@ -449,7 +499,13 @@ UndistributeTable(TableConversionParameters *params)
 							   "because the table is not distributed")));
 	}
 
-	ErrorIfTenantTable(params->relationId, "undistribute_table");
+	Oid schemaId = get_rel_namespace(params->relationId);
+	if (!params->bypassTenantCheck && IsTenantSchema(schemaId) &&
+		IsCitusTableType(params->relationId, SINGLE_SHARD_DISTRIBUTED))
+	{
+		EnsureUndistributeTenantTableSafe(params->relationId,
+										  TenantOperationNames[TENANT_UNDISTRIBUTE_TABLE]);
+	}
 
 	if (!params->cascadeViaForeignKeys)
 	{
@@ -506,7 +562,7 @@ AlterDistributedTable(TableConversionParameters *params)
 							   "is not distributed")));
 	}
 
-	ErrorIfTenantTable(params->relationId, "alter_distributed_table");
+	ErrorIfTenantTable(params->relationId, TenantOperationNames[TENANT_ALTER_TABLE]);
 	ErrorIfColocateWithTenantTable(params->colocateWith);
 
 	EnsureTableNotForeign(params->relationId);
@@ -1267,7 +1323,8 @@ ErrorIfColocateWithTenantTable(char *colocateWith)
 	{
 		text *colocateWithTableNameText = cstring_to_text(colocateWith);
 		Oid colocateWithTableId = ResolveRelationId(colocateWithTableNameText, false);
-		ErrorIfTenantTable(colocateWithTableId, "colocate_with");
+		ErrorIfTenantTable(colocateWithTableId,
+						   TenantOperationNames[TENANT_COLOCATE_WITH]);
 	}
 }
 
diff --git a/src/backend/distributed/commands/cascade_table_operation_for_connected_relations.c b/src/backend/distributed/commands/cascade_table_operation_for_connected_relations.c
index 1c01028d3..1102a3a51 100644
--- a/src/backend/distributed/commands/cascade_table_operation_for_connected_relations.c
+++ b/src/backend/distributed/commands/cascade_table_operation_for_connected_relations.c
@@ -468,7 +468,8 @@ ExecuteCascadeOperationForRelationIdList(List *relationIdList,
 				{
 					TableConversionParameters params = {
 						.relationId = relationId,
-						.cascadeViaForeignKeys = cascadeViaForeignKeys
+						.cascadeViaForeignKeys = cascadeViaForeignKeys,
+						.bypassTenantCheck = false
 					};
 					UndistributeTable(&params);
 				}
diff --git a/src/backend/distributed/commands/foreign_constraint.c b/src/backend/distributed/commands/foreign_constraint.c
index b48b6c54a..40ccb0ddf 100644
--- a/src/backend/distributed/commands/foreign_constraint.c
+++ b/src/backend/distributed/commands/foreign_constraint.c
@@ -1356,6 +1356,10 @@ IsTableTypeIncluded(Oid relationId, int flags)
 	{
 		return (flags & INCLUDE_LOCAL_TABLES) != 0;
 	}
+	else if (IsCitusTableType(relationId, SINGLE_SHARD_DISTRIBUTED))
+	{
+		return (flags & INCLUDE_SINGLE_SHARD_TABLES) != 0;
+	}
 	else if (IsCitusTableType(relationId, DISTRIBUTED_TABLE))
 	{
 		return (flags & INCLUDE_DISTRIBUTED_TABLES) != 0;
diff --git a/src/backend/distributed/commands/schema_based_sharding.c b/src/backend/distributed/commands/schema_based_sharding.c
index 91bae0943..b717cb5ae 100644
--- a/src/backend/distributed/commands/schema_based_sharding.c
+++ b/src/backend/distributed/commands/schema_based_sharding.c
@@ -40,6 +40,14 @@ static void EnsureSchemaExist(Oid schemaId);
 /* controlled via citus.enable_schema_based_sharding GUC */
 bool EnableSchemaBasedSharding = false;
 
+const char *TenantOperationNames[TOTAL_TENANT_OPERATION] = {
+	"undistribute_table",
+	"alter_distributed_table",
+	"colocate_with",
+	"update_distributed_table_colocation",
+	"set schema",
+};
+
 
 PG_FUNCTION_INFO_V1(citus_internal_unregister_tenant_schema_globally);
 PG_FUNCTION_INFO_V1(citus_schema_distribute);
@@ -374,12 +382,7 @@ SchemaGetNonShardTableIdList(Oid schemaId)
  *  - Schema name is in the allowed-list,
  *  - Schema does not depend on an extension (created by extension),
  *  - No extension depends on the schema (CREATE EXTENSION <ext> SCHEMA <schema>),
- *  - Current user should be the owner of tables under the schema,
- *  - Table kinds are supported,
- *  - Referencing and referenced foreign keys for the tables under the schema are
- *    supported,
- *	- Tables under the schema are not owned by an extension,
- *  - Only Citus local and Postgres local tables exist under the schema.
+ *	- Some checks for the table for being a valid tenant table.
  */
 static void
 EnsureSchemaCanBeDistributed(Oid schemaId, List *schemaTableIdList)
@@ -409,39 +412,55 @@ EnsureSchemaCanBeDistributed(Oid schemaId, List *schemaTableIdList)
 	Oid relationId = InvalidOid;
 	foreach_oid(relationId, schemaTableIdList)
 	{
-		/* Ensure table owner */
-		EnsureTableOwner(relationId);
+		EnsureTenantTable(relationId, "citus_schema_distribute");
+	}
+}
 
-		/* Check relation kind */
-		EnsureTableKindSupportedForTenantSchema(relationId);
 
-		/* Check foreign keys */
-		EnsureFKeysForTenantTable(relationId);
+/*
+ * EnsureTenantTable ensures the table can be a valid tenant table.
+ *  - Current user should be the owner of table,
+ *  - Table kind is supported,
+ *  - Referencing and referenced foreign keys for the table are supported,
+ *	- Table is not owned by an extension,
+ *  - Table should be Citus local or Postgres local table.
+ */
+void
+EnsureTenantTable(Oid relationId, char *operationName)
+{
+	/* Ensure table owner */
+	EnsureTableOwner(relationId);
 
-		/* Check table not owned by an extension */
-		ObjectAddress *tableAddress = palloc0(sizeof(ObjectAddress));
-		ObjectAddressSet(*tableAddress, RelationRelationId, relationId);
-		if (IsAnyObjectAddressOwnedByExtension(list_make1(tableAddress), NULL))
-		{
-			char *tableName = get_namespace_name(schemaId);
-			ereport(ERROR, (errmsg("schema cannot be distributed since it has "
-								   "table %s which is owned by an extension",
-								   tableName)));
-		}
+	/* Check relation kind */
+	EnsureTableKindSupportedForTenantSchema(relationId);
 
-		/* Postgres local tables are allowed */
-		if (!IsCitusTable(relationId))
-		{
-			continue;
-		}
+	/* Check foreign keys */
+	EnsureFKeysForTenantTable(relationId);
 
-		/* Only Citus local tables, amongst Citus table types, are allowed */
-		if (!IsCitusTableType(relationId, CITUS_LOCAL_TABLE))
-		{
-			ereport(ERROR, (errmsg("schema already has distributed tables"),
-							errhint("Undistribute distributed tables under "
-									"the schema before distributing the schema.")));
-		}
+	/* Check table not owned by an extension */
+	ObjectAddress *tableAddress = palloc0(sizeof(ObjectAddress));
+	ObjectAddressSet(*tableAddress, RelationRelationId, relationId);
+	if (IsAnyObjectAddressOwnedByExtension(list_make1(tableAddress), NULL))
+	{
+		Oid schemaId = get_rel_namespace(relationId);
+		char *tableName = get_namespace_name(schemaId);
+		ereport(ERROR, (errmsg("schema cannot be distributed since it has "
+							   "table %s which is owned by an extension",
+							   tableName)));
+	}
+
+	/* Postgres local tables are allowed */
+	if (!IsCitusTable(relationId))
+	{
+		return;
+	}
+
+	/* Only Citus local tables, amongst Citus table types, are allowed */
+	if (!IsCitusTableType(relationId, CITUS_LOCAL_TABLE))
+	{
+		ereport(ERROR, (errmsg("distributed schema cannot have distributed tables"),
+						errhint("Undistribute distributed tables before "
+								"'%s'.", operationName)));
 	}
 }
 
@@ -743,7 +762,7 @@ citus_schema_undistribute(PG_FUNCTION_ARGS)
  * if the given relation is a tenant table.
  */
 void
-ErrorIfTenantTable(Oid relationId, char *operationName)
+ErrorIfTenantTable(Oid relationId, const char *operationName)
 {
 	if (IsTenantSchema(get_rel_namespace(relationId)))
 	{
@@ -753,20 +772,3 @@ ErrorIfTenantTable(Oid relationId, char *operationName)
 							   operationName)));
 	}
 }
-
-
-/*
- * ErrorIfTenantSchema errors out with the given operation name,
- * if the given schema is a tenant schema.
- */
-void
-ErrorIfTenantSchema(Oid nspOid, char *operationName)
-{
-	if (IsTenantSchema(nspOid))
-	{
-		ereport(ERROR, (errmsg(
-							"%s is not allowed for %s because it is a distributed schema",
-							get_namespace_name(nspOid),
-							operationName)));
-	}
-}
diff --git a/src/backend/distributed/commands/table.c b/src/backend/distributed/commands/table.c
index a5e997969..4ea28c71d 100644
--- a/src/backend/distributed/commands/table.c
+++ b/src/backend/distributed/commands/table.c
@@ -41,6 +41,7 @@
 #include "distributed/resource_lock.h"
 #include "distributed/version_compat.h"
 #include "distributed/worker_shard_visibility.h"
+#include "distributed/tenant_schema_metadata.h"
 #include "foreign/foreign.h"
 #include "lib/stringinfo.h"
 #include "nodes/parsenodes.h"
@@ -2310,9 +2311,52 @@ PreprocessAlterTableSchemaStmt(Node *node, const char *queryString,
 		return NIL;
 	}
 
-	ErrorIfTenantTable(relationId, "ALTER TABLE SET SCHEMA");
-	ErrorIfTenantSchema(get_namespace_oid(stmt->newschema, false),
-						"ALTER TABLE SET SCHEMA");
+	Oid oldSchemaId = get_rel_namespace(relationId);
+	Oid newSchemaId = get_namespace_oid(stmt->newschema, stmt->missing_ok);
+	if (!OidIsValid(oldSchemaId) || !OidIsValid(newSchemaId))
+	{
+		return NIL;
+	}
+
+	/*  Do nothing if new schema is the same as old schema */
+	if (newSchemaId == oldSchemaId)
+	{
+		return NIL;
+	}
+
+	/* Undistribute table if its old schema is a tenant schema */
+	if (IsTenantSchema(oldSchemaId) && IsCoordinator())
+	{
+		EnsureUndistributeTenantTableSafe(relationId,
+										  TenantOperationNames[TENANT_SET_SCHEMA]);
+
+		char *oldSchemaName = get_namespace_name(oldSchemaId);
+		char *tableName = stmt->relation->relname;
+		ereport(NOTICE, (errmsg("undistributing table %s in distributed schema %s "
+								"before altering its schema", tableName, oldSchemaName)));
+
+		/* Undistribute tenant table by suppressing weird notices */
+		TableConversionParameters params = {
+			.relationId = relationId,
+			.cascadeViaForeignKeys = false,
+			.bypassTenantCheck = true,
+			.suppressNoticeMessages = true,
+		};
+		UndistributeTable(&params);
+
+		/* relation id changes after undistribute_table */
+		relationId = get_relname_relid(tableName, oldSchemaId);
+
+		/*
+		 * After undistribution, the table could be Citus table or Postgres table.
+		 * If it is Postgres table, do not propagate the `ALTER TABLE SET SCHEMA`
+		 * command to workers.
+		 */
+		if (!IsCitusTable(relationId))
+		{
+			return NIL;
+		}
+	}
 
 	DDLJob *ddlJob = palloc0(sizeof(DDLJob));
 	QualifyTreeNode((Node *) stmt);
@@ -4166,3 +4210,61 @@ ConvertNewTableIfNecessary(Node *createStmt)
 		CreateCitusLocalTable(createdRelationId, cascade, autoConverted);
 	}
 }
+
+
+/*
+ * ConvertToTenantTableIfNecessary converts given relation to a tenant table if its
+ * schema changed to a distributed schema.
+ */
+void
+ConvertToTenantTableIfNecessary(AlterObjectSchemaStmt *stmt)
+{
+	Assert(stmt->objectType == OBJECT_TABLE || stmt->objectType == OBJECT_FOREIGN_TABLE);
+
+	if (!IsCoordinator())
+	{
+		return;
+	}
+
+	/*
+	 * We will let Postgres deal with missing_ok
+	 */
+	List *tableAddresses = GetObjectAddressListFromParseTree((Node *) stmt, true, true);
+
+	/*  the code-path only supports a single object */
+	Assert(list_length(tableAddresses) == 1);
+
+	/* We have already asserted that we have exactly 1 address in the addresses. */
+	ObjectAddress *tableAddress = linitial(tableAddresses);
+	char relKind = get_rel_relkind(tableAddress->objectId);
+	if (relKind == RELKIND_SEQUENCE || relKind == RELKIND_VIEW)
+	{
+		return;
+	}
+
+	Oid relationId = tableAddress->objectId;
+	Oid schemaId = get_namespace_oid(stmt->newschema, stmt->missing_ok);
+	if (!OidIsValid(schemaId))
+	{
+		return;
+	}
+
+	/*
+	 * Make table a tenant table when its schema actually changed. When its schema
+	 * is not changed as in `ALTER TABLE <tbl> SET SCHEMA <same_schema>`, we detect
+	 * that by seeing the table is still a single shard table. (i.e. not undistributed
+	 * at `preprocess` step)
+	 */
+	if (!IsCitusTableType(relationId, SINGLE_SHARD_DISTRIBUTED) &&
+		IsTenantSchema(schemaId))
+	{
+		EnsureTenantTable(relationId, "ALTER TABLE SET SCHEMA");
+
+		char *schemaName = get_namespace_name(schemaId);
+		char *tableName = stmt->relation->relname;
+		ereport(NOTICE, (errmsg("converting table %s to a tenant table in distributed "
+								"schema %s", tableName, schemaName)));
+
+		CreateTenantSchemaTable(relationId);
+	}
+}
diff --git a/src/backend/distributed/commands/utility_hook.c b/src/backend/distributed/commands/utility_hook.c
index 6393fdd71..ad2f5de1b 100644
--- a/src/backend/distributed/commands/utility_hook.c
+++ b/src/backend/distributed/commands/utility_hook.c
@@ -370,6 +370,18 @@ multi_ProcessUtility(PlannedStmt *pstmt,
 
 				ConvertNewTableIfNecessary(createStmt);
 			}
+
+			if (context == PROCESS_UTILITY_TOPLEVEL &&
+				IsA(parsetree, AlterObjectSchemaStmt))
+			{
+				AlterObjectSchemaStmt *alterSchemaStmt = castNode(AlterObjectSchemaStmt,
+																  parsetree);
+				if (alterSchemaStmt->objectType == OBJECT_TABLE ||
+					alterSchemaStmt->objectType == OBJECT_FOREIGN_TABLE)
+				{
+					ConvertToTenantTableIfNecessary(alterSchemaStmt);
+				}
+			}
 		}
 
 		UtilityHookLevel--;
@@ -999,7 +1011,8 @@ UndistributeDisconnectedCitusLocalTables(void)
 		TableConversionParameters params = {
 			.relationId = citusLocalTableId,
 			.cascadeViaForeignKeys = true,
-			.suppressNoticeMessages = true
+			.suppressNoticeMessages = true,
+			.bypassTenantCheck = false
 		};
 		UndistributeTable(&params);
 	}
diff --git a/src/backend/distributed/utils/colocation_utils.c b/src/backend/distributed/utils/colocation_utils.c
index dba791ba4..62a13af3a 100644
--- a/src/backend/distributed/utils/colocation_utils.c
+++ b/src/backend/distributed/utils/colocation_utils.c
@@ -116,7 +116,7 @@ update_distributed_table_colocation(PG_FUNCTION_ARGS)
 	text *colocateWithTableNameText = PG_GETARG_TEXT_P(1);
 
 	EnsureTableOwner(targetRelationId);
-	ErrorIfTenantTable(targetRelationId, "update_distributed_table_colocation");
+	ErrorIfTenantTable(targetRelationId, TenantOperationNames[TENANT_UPDATE_COLOCATION]);
 
 	char *colocateWithTableName = text_to_cstring(colocateWithTableNameText);
 	if (IsColocateWithNone(colocateWithTableName))
@@ -127,7 +127,8 @@ update_distributed_table_colocation(PG_FUNCTION_ARGS)
 	else
 	{
 		Oid colocateWithTableId = ResolveRelationId(colocateWithTableNameText, false);
-		ErrorIfTenantTable(colocateWithTableId, "colocate_with");
+		ErrorIfTenantTable(colocateWithTableId,
+						   TenantOperationNames[TENANT_COLOCATE_WITH]);
 		EnsureTableOwner(colocateWithTableId);
 		MarkTablesColocated(colocateWithTableId, targetRelationId);
 	}
diff --git a/src/include/distributed/commands.h b/src/include/distributed/commands.h
index 442e58058..a013f3977 100644
--- a/src/include/distributed/commands.h
+++ b/src/include/distributed/commands.h
@@ -118,7 +118,7 @@ typedef enum ExtractForeignKeyConstraintsMode
 	/* exclude the self-referencing foreign keys */
 	EXCLUDE_SELF_REFERENCES = 1 << 2,
 
-	/* any combination of the 4 flags below is supported */
+	/* any combination of the 5 flags below is supported */
 	/* include foreign keys when the other table is a distributed table*/
 	INCLUDE_DISTRIBUTED_TABLES = 1 << 3,
 
@@ -131,9 +131,13 @@ typedef enum ExtractForeignKeyConstraintsMode
 	/* include foreign keys when the other table is a Postgres local table*/
 	INCLUDE_LOCAL_TABLES = 1 << 6,
 
+	/* include foreign keys when the other table is a single shard table*/
+	INCLUDE_SINGLE_SHARD_TABLES = 1 << 7,
+
 	/* include foreign keys regardless of the other table's type */
 	INCLUDE_ALL_TABLE_TYPES = INCLUDE_DISTRIBUTED_TABLES | INCLUDE_REFERENCE_TABLES |
-							  INCLUDE_CITUS_LOCAL_TABLES | INCLUDE_LOCAL_TABLES
+							  INCLUDE_CITUS_LOCAL_TABLES | INCLUDE_LOCAL_TABLES |
+							  INCLUDE_SINGLE_SHARD_TABLES
 } ExtractForeignKeyConstraintMode;
 
 
@@ -155,6 +159,19 @@ typedef enum SearchForeignKeyColumnFlags
 	/* callers can also pass union of above flags */
 } SearchForeignKeyColumnFlags;
 
+
+typedef enum TenantOperation
+{
+	TENANT_UNDISTRIBUTE_TABLE = 0,
+	TENANT_ALTER_TABLE,
+	TENANT_COLOCATE_WITH,
+	TENANT_UPDATE_COLOCATION,
+	TENANT_SET_SCHEMA,
+} TenantOperation;
+
+#define TOTAL_TENANT_OPERATION 5
+extern const char *TenantOperationNames[TOTAL_TENANT_OPERATION];
+
 /* begin.c - forward declarations */
 extern void SaveBeginCommandProperties(TransactionStmt *transactionStmt);
 
@@ -593,6 +610,7 @@ extern char * GetAlterColumnWithNextvalDefaultCmd(Oid sequenceOid, Oid relationI
 
 extern void ErrorIfTableHasIdentityColumn(Oid relationId);
 extern void ConvertNewTableIfNecessary(Node *createStmt);
+extern void ConvertToTenantTableIfNecessary(AlterObjectSchemaStmt *alterObjectSchemaStmt);
 
 /* text_search.c - forward declarations */
 extern List * GetCreateTextSearchConfigStatements(const ObjectAddress *address);
@@ -792,11 +810,11 @@ extern void UpdateAutoConvertedForConnectedRelations(List *relationId, bool
 extern bool ShouldUseSchemaBasedSharding(char *schemaName);
 extern bool ShouldCreateTenantSchemaTable(Oid relationId);
 extern bool IsTenantSchema(Oid schemaId);
+extern void EnsureTenantTable(Oid relationId, char *operationName);
 extern void ErrorIfIllegalPartitioningInTenantSchema(Oid parentRelationId,
 													 Oid partitionRelationId);
 extern void CreateTenantSchemaTable(Oid relationId);
-extern void ErrorIfTenantTable(Oid relationId, char *operationName);
-extern void ErrorIfTenantSchema(Oid nspOid, char *operationName);
+extern void ErrorIfTenantTable(Oid relationId, const char *operationName);
 extern uint32 CreateTenantSchemaColocationId(void);
 
 #endif /*CITUS_COMMANDS_H */
diff --git a/src/include/distributed/metadata_utility.h b/src/include/distributed/metadata_utility.h
index 473d105e8..6536e89bc 100644
--- a/src/include/distributed/metadata_utility.h
+++ b/src/include/distributed/metadata_utility.h
@@ -172,6 +172,12 @@ typedef struct TableConversionParameters
 	 * messages that we explicitly issue
 	 */
 	bool suppressNoticeMessages;
+
+	/*
+	 * bypassTenantCheck skips tenant table checks to allow some internal
+	 * operations which are normally disallowed
+	 */
+	bool bypassTenantCheck;
 } TableConversionParameters;
 
 typedef struct TableConversionReturn
@@ -363,6 +369,7 @@ extern void CreateDistributedTable(Oid relationId, char *distributionColumnName,
 								   bool shardCountIsStrict, char *colocateWithTableName);
 extern void CreateReferenceTable(Oid relationId);
 extern void CreateTruncateTrigger(Oid relationId);
+extern void EnsureUndistributeTenantTableSafe(Oid relationId, const char *operationName);
 extern TableConversionReturn * UndistributeTable(TableConversionParameters *params);
 extern void UndistributeTables(List *relationIdList);
 
diff --git a/src/test/regress/expected/citus_schema_distribute_undistribute.out b/src/test/regress/expected/citus_schema_distribute_undistribute.out
index 8e6803b6f..ae08b6c6a 100644
--- a/src/test/regress/expected/citus_schema_distribute_undistribute.out
+++ b/src/test/regress/expected/citus_schema_distribute_undistribute.out
@@ -482,8 +482,8 @@ SELECT create_distributed_table('tenant1.dist', 'id');
 (1 row)
 
 SELECT citus_schema_distribute('tenant1');
-ERROR:  schema already has distributed tables
-HINT:  Undistribute distributed tables under the schema before distributing the schema.
+ERROR:  distributed schema cannot have distributed tables
+HINT:  Undistribute distributed tables before 'citus_schema_distribute'.
 SELECT undistribute_table('tenant1.dist');
  undistribute_table
 ---------------------------------------------------------------------
@@ -510,8 +510,8 @@ SELECT create_reference_table('tenant1.ref2');
 (1 row)
 
 SELECT citus_schema_distribute('tenant1');
-ERROR:  schema already has distributed tables
-HINT:  Undistribute distributed tables under the schema before distributing the schema.
+ERROR:  distributed schema cannot have distributed tables
+HINT:  Undistribute distributed tables before 'citus_schema_distribute'.
 SELECT undistribute_table('tenant1.ref2');
  undistribute_table
 ---------------------------------------------------------------------
@@ -766,8 +766,8 @@ SELECT create_distributed_table('tenant1.new_dist', 'id');
 (1 row)
 
 SELECT citus_schema_distribute('tenant1');
-ERROR:  schema already has distributed tables
-HINT:  Undistribute distributed tables under the schema before distributing the schema.
+ERROR:  distributed schema cannot have distributed tables
+HINT:  Undistribute distributed tables before 'citus_schema_distribute'.
 SELECT undistribute_table('tenant1.new_dist');
  undistribute_table
 ---------------------------------------------------------------------
@@ -795,8 +795,8 @@ SELECT create_distributed_table('tenant1.single_shard_t', NULL);
 (1 row)
 
 SELECT citus_schema_distribute('tenant1');
-ERROR:  schema already has distributed tables
-HINT:  Undistribute distributed tables under the schema before distributing the schema.
+ERROR:  distributed schema cannot have distributed tables
+HINT:  Undistribute distributed tables before 'citus_schema_distribute'.
 SELECT undistribute_table('tenant1.single_shard_t');
  undistribute_table
 ---------------------------------------------------------------------
diff --git a/src/test/regress/expected/schema_based_sharding.out b/src/test/regress/expected/schema_based_sharding.out
index 951fc05e7..b041708cf 100644
--- a/src/test/regress/expected/schema_based_sharding.out
+++ b/src/test/regress/expected/schema_based_sharding.out
@@ -80,24 +80,148 @@ ERROR:  tenant_2.test_table is not allowed for update_distributed_table_colocati
 -- verify we also don't allow colocate_with a tenant table
 SELECT update_distributed_table_colocation('regular_schema.test_table', colocate_with => 'tenant_2.test_table');
 ERROR:  tenant_2.test_table is not allowed for colocate_with because it belongs to a distributed schema
--- verify we don't allow undistribute_table for tenant tables
-SELECT undistribute_table('tenant_2.test_table');
-ERROR:  tenant_2.test_table is not allowed for undistribute_table because it belongs to a distributed schema
+-- verify we do not allow undistribute_table for tenant tables
+CREATE TABLE tenant_2.undist_table(id int);
+SELECT undistribute_table('tenant_2.undist_table');
+ERROR:  tenant_2.undist_table is not allowed for undistribute_table because it belongs to a distributed schema
 -- verify we don't allow alter_distributed_table for tenant tables
 SELECT alter_distributed_table('tenant_2.test_table', colocate_with => 'none');
 ERROR:  tenant_2.test_table is not allowed for alter_distributed_table because it belongs to a distributed schema
 -- verify we also don't allow colocate_with a tenant table
 SELECT alter_distributed_table('regular_schema.test_table', colocate_with => 'tenant_2.test_table');
 ERROR:  tenant_2.test_table is not allowed for colocate_with because it belongs to a distributed schema
--- verify we don't allow ALTER TABLE SET SCHEMA for tenant tables
-ALTER TABLE tenant_2.test_table SET SCHEMA regular_schema;
-ERROR:  tenant_2.test_table is not allowed for ALTER TABLE SET SCHEMA because it belongs to a distributed schema
--- verify we don't allow ALTER TABLE SET SCHEMA for tenant schemas
-ALTER TABLE regular_schema.test_table SET SCHEMA tenant_2;
-ERROR:  tenant_2 is not allowed for ALTER TABLE SET SCHEMA because it is a distributed schema
--- the same, from tenant schema to tenant schema
-ALTER TABLE tenant_2.test_table SET SCHEMA tenant_3;
-ERROR:  tenant_2.test_table is not allowed for ALTER TABLE SET SCHEMA because it belongs to a distributed schema
+-- verify we can set tenant table's schema to regular schema
+CREATE TABLE tenant_2.test_table2(id int);
+ALTER TABLE tenant_2.test_table2 SET SCHEMA regular_schema;
+NOTICE:  undistributing table test_table2 in distributed schema tenant_2 before altering its schema
+-- verify that regular_schema.test_table2 does not exist in pg_dist_partition
+SELECT COUNT(*)=0 FROM pg_dist_partition
+WHERE logicalrelid = 'regular_schema.test_table2'::regclass AND
+      partmethod = 'n' AND repmodel = 's' AND colocationid > 0;
+ ?column?
+---------------------------------------------------------------------
+ t
+(1 row)
+
+-- verify that tenant_2.test_table2 does not exist
+SELECT * FROM tenant_2.test_table2;
+ERROR:  relation "tenant_2.test_table2" does not exist
+-- verify we can set regular table's schema to distributed schema
+CREATE TABLE regular_schema.test_table3(id int);
+ALTER TABLE regular_schema.test_table3 SET SCHEMA tenant_2;
+NOTICE:  converting table test_table3 to a tenant table in distributed schema tenant_2
+-- verify that tenant_2.test_table3 is recorded in pg_dist_partition as a single-shard table.
+SELECT COUNT(*)=1 FROM pg_dist_partition
+WHERE logicalrelid = 'tenant_2.test_table3'::regclass AND
+      partmethod = 'n' AND repmodel = 's' AND colocationid > 0;
+ ?column?
+---------------------------------------------------------------------
+ t
+(1 row)
+
+-- verify that regular_schema.test_table3 does not exist
+SELECT * FROM regular_schema.test_table3;
+ERROR:  relation "regular_schema.test_table3" does not exist
+-- verify we can set tenant table's schema to another distributed schema
+CREATE TABLE tenant_2.test_table4(id int);
+ALTER TABLE tenant_2.test_table4 SET SCHEMA tenant_3;
+NOTICE:  undistributing table test_table4 in distributed schema tenant_2 before altering its schema
+NOTICE:  converting table test_table4 to a tenant table in distributed schema tenant_3
+-- verify that tenant_3.test_table4 is recorded in pg_dist_partition as a single-shard table.
+SELECT COUNT(*)=1 FROM pg_dist_partition
+WHERE logicalrelid = 'tenant_3.test_table4'::regclass AND
+      partmethod = 'n' AND repmodel = 's' AND colocationid > 0;
+ ?column?
+---------------------------------------------------------------------
+ t
+(1 row)
+
+-- verify that tenant_2.test_table4 does not exist
+SELECT * FROM tenant_2.test_table4;
+ERROR:  relation "tenant_2.test_table4" does not exist
+-- verify that we can put a local table in regular schema into distributed schema
+CREATE TABLE regular_schema.pg_local_tbl(id int);
+ALTER TABLE regular_schema.pg_local_tbl SET SCHEMA tenant_2;
+NOTICE:  converting table pg_local_tbl to a tenant table in distributed schema tenant_2
+-- verify that we can put a Citus local table in regular schema into distributed schema
+CREATE TABLE regular_schema.citus_local_tbl(id int);
+SELECT citus_add_local_table_to_metadata('regular_schema.citus_local_tbl');
+ citus_add_local_table_to_metadata
+---------------------------------------------------------------------
+
+(1 row)
+
+ALTER TABLE regular_schema.citus_local_tbl SET SCHEMA tenant_2;
+NOTICE:  converting table citus_local_tbl to a tenant table in distributed schema tenant_2
+-- verify that we do not allow a hash distributed table in regular schema into distributed schema
+CREATE TABLE regular_schema.hash_dist_tbl(id int);
+SELECT create_distributed_table('regular_schema.hash_dist_tbl', 'id');
+ create_distributed_table
+---------------------------------------------------------------------
+
+(1 row)
+
+ALTER TABLE regular_schema.hash_dist_tbl SET SCHEMA tenant_2;
+ERROR:  distributed schema cannot have distributed tables
+HINT:  Undistribute distributed tables before 'ALTER TABLE SET SCHEMA'.
+-- verify that we do not allow a reference table in regular schema into distributed schema
+CREATE TABLE regular_schema.ref_tbl(id int  PRIMARY KEY);
+SELECT create_reference_table('regular_schema.ref_tbl');
+ create_reference_table
+---------------------------------------------------------------------
+
+(1 row)
+
+ALTER TABLE regular_schema.ref_tbl SET SCHEMA tenant_2;
+ERROR:  distributed schema cannot have distributed tables
+HINT:  Undistribute distributed tables before 'ALTER TABLE SET SCHEMA'.
+-- verify that we can put a table in tenant schema into regular schema
+CREATE TABLE tenant_2.tenant_tbl(id int);
+ALTER TABLE tenant_2.tenant_tbl SET SCHEMA regular_schema;
+NOTICE:  undistributing table tenant_tbl in distributed schema tenant_2 before altering its schema
+-- verify that we can put a table in tenant schema into another tenant schema
+CREATE TABLE tenant_2.tenant_tbl2(id int);
+ALTER TABLE tenant_2.tenant_tbl2 SET SCHEMA tenant_3;
+NOTICE:  undistributing table tenant_tbl2 in distributed schema tenant_2 before altering its schema
+NOTICE:  converting table tenant_tbl2 to a tenant table in distributed schema tenant_3
+-- verify that we do not allow a local table in regular schema into distributed schema if it has foreign key to a non-reference table in another schema
+CREATE TABLE regular_schema.pg_local_tbl1(id int PRIMARY KEY);
+CREATE TABLE regular_schema.pg_local_tbl2(id int REFERENCES regular_schema.pg_local_tbl1(id));
+ALTER TABLE regular_schema.pg_local_tbl2 SET SCHEMA tenant_2;
+ERROR:  foreign keys from distributed schemas can only point to the same distributed schema or reference tables in regular schemas
+DETAIL:  "tenant_2.pg_local_tbl2" references "regular_schema.pg_local_tbl1" via foreign key constraint "pg_local_tbl2_id_fkey"
+-- verify that we allow a local table in regular schema into distributed schema if it has foreign key to a reference table in another schema
+CREATE TABLE regular_schema.pg_local_tbl3(id int REFERENCES regular_schema.ref_tbl(id));
+ALTER TABLE regular_schema.pg_local_tbl3 SET SCHEMA tenant_2;
+NOTICE:  converting table pg_local_tbl3 to a tenant table in distributed schema tenant_2
+-- verify that we do not allow a table in tenant schema into regular schema if it has foreign key to/from another table in the same schema
+CREATE TABLE tenant_2.tenant_tbl1(id int PRIMARY KEY);
+CREATE TABLE tenant_2.tenant_tbl2(id int REFERENCES tenant_2.tenant_tbl1(id));
+ALTER TABLE tenant_2.tenant_tbl1 SET SCHEMA regular_schema;
+ERROR:  set schema is not allowed for table tenant_tbl1 in distributed schema tenant_2
+DETAIL:  distributed schemas cannot have foreign keys from/to local tables or different schema
+ALTER TABLE tenant_2.tenant_tbl2 SET SCHEMA regular_schema;
+ERROR:  set schema is not allowed for table tenant_tbl2 in distributed schema tenant_2
+DETAIL:  distributed schemas cannot have foreign keys from/to local tables or different schema
+-- verify that we do not allow a table in distributed schema into another distributed schema if it has foreign key to/from another table in the same schema
+CREATE TABLE tenant_2.tenant_tbl3(id int PRIMARY KEY);
+CREATE TABLE tenant_2.tenant_tbl4(id int REFERENCES tenant_2.tenant_tbl3(id));
+ALTER TABLE tenant_2.tenant_tbl3 SET SCHEMA tenant_3;
+ERROR:  set schema is not allowed for table tenant_tbl3 in distributed schema tenant_2
+DETAIL:  distributed schemas cannot have foreign keys from/to local tables or different schema
+ALTER TABLE tenant_2.tenant_tbl4 SET SCHEMA tenant_3;
+ERROR:  set schema is not allowed for table tenant_tbl4 in distributed schema tenant_2
+DETAIL:  distributed schemas cannot have foreign keys from/to local tables or different schema
+-- alter set non-existent schema
+ALTER TABLE tenant_2.test_table SET SCHEMA ghost_schema;
+ERROR:  schema "ghost_schema" does not exist
+ALTER TABLE IF EXISTS tenant_2.test_table SET SCHEMA ghost_schema;
+ERROR:  schema "ghost_schema" does not exist
+-- alter set non-existent table
+ALTER TABLE tenant_2.ghost_table SET SCHEMA ghost_schema;
+ERROR:  relation "tenant_2.ghost_table" does not exist
+ALTER TABLE IF EXISTS tenant_2.ghost_table SET SCHEMA ghost_schema;
+NOTICE:  relation "ghost_table" does not exist, skipping
 -- (on coordinator) verify that colocation id is set for empty tenants too
 SELECT colocationid > 0 FROM pg_dist_schema
 WHERE schemaid::regnamespace::text IN ('tenant_1', 'tenant_3');
@@ -265,8 +389,8 @@ SELECT EXISTS(
 (1 row)
 
 INSERT INTO tenant_4.another_partitioned_table VALUES (1, 'a');
-ERROR:  insert or update on table "another_partitioned_table_child_1920040" violates foreign key constraint "another_partitioned_table_a_fkey_1920039"
-DETAIL:  Key (a)=(1) is not present in table "partitioned_table_1920037".
+ERROR:  insert or update on table "another_partitioned_table_child_1920090" violates foreign key constraint "another_partitioned_table_a_fkey_1920089"
+DETAIL:  Key (a)=(1) is not present in table "partitioned_table_1920087".
 CONTEXT:  while executing command on localhost:xxxxx
 INSERT INTO tenant_4.partitioned_table VALUES (1, 'a');
 INSERT INTO tenant_4.another_partitioned_table VALUES (1, 'a');
diff --git a/src/test/regress/sql/schema_based_sharding.sql b/src/test/regress/sql/schema_based_sharding.sql
index 88ca77a40..7b16ba904 100644
--- a/src/test/regress/sql/schema_based_sharding.sql
+++ b/src/test/regress/sql/schema_based_sharding.sql
@@ -59,18 +59,100 @@ SELECT citus_add_local_table_to_metadata('tenant_2.test_table');
 SELECT update_distributed_table_colocation('tenant_2.test_table', colocate_with => 'none');
 -- verify we also don't allow colocate_with a tenant table
 SELECT update_distributed_table_colocation('regular_schema.test_table', colocate_with => 'tenant_2.test_table');
--- verify we don't allow undistribute_table for tenant tables
-SELECT undistribute_table('tenant_2.test_table');
+
+-- verify we do not allow undistribute_table for tenant tables
+CREATE TABLE tenant_2.undist_table(id int);
+SELECT undistribute_table('tenant_2.undist_table');
+
 -- verify we don't allow alter_distributed_table for tenant tables
 SELECT alter_distributed_table('tenant_2.test_table', colocate_with => 'none');
 -- verify we also don't allow colocate_with a tenant table
 SELECT alter_distributed_table('regular_schema.test_table', colocate_with => 'tenant_2.test_table');
--- verify we don't allow ALTER TABLE SET SCHEMA for tenant tables
-ALTER TABLE tenant_2.test_table SET SCHEMA regular_schema;
--- verify we don't allow ALTER TABLE SET SCHEMA for tenant schemas
-ALTER TABLE regular_schema.test_table SET SCHEMA tenant_2;
--- the same, from tenant schema to tenant schema
-ALTER TABLE tenant_2.test_table SET SCHEMA tenant_3;
+
+-- verify we can set tenant table's schema to regular schema
+CREATE TABLE tenant_2.test_table2(id int);
+ALTER TABLE tenant_2.test_table2 SET SCHEMA regular_schema;
+-- verify that regular_schema.test_table2 does not exist in pg_dist_partition
+SELECT COUNT(*)=0 FROM pg_dist_partition
+WHERE logicalrelid = 'regular_schema.test_table2'::regclass AND
+      partmethod = 'n' AND repmodel = 's' AND colocationid > 0;
+-- verify that tenant_2.test_table2 does not exist
+SELECT * FROM tenant_2.test_table2;
+
+-- verify we can set regular table's schema to distributed schema
+CREATE TABLE regular_schema.test_table3(id int);
+ALTER TABLE regular_schema.test_table3 SET SCHEMA tenant_2;
+-- verify that tenant_2.test_table3 is recorded in pg_dist_partition as a single-shard table.
+SELECT COUNT(*)=1 FROM pg_dist_partition
+WHERE logicalrelid = 'tenant_2.test_table3'::regclass AND
+      partmethod = 'n' AND repmodel = 's' AND colocationid > 0;
+-- verify that regular_schema.test_table3 does not exist
+SELECT * FROM regular_schema.test_table3;
+
+-- verify we can set tenant table's schema to another distributed schema
+CREATE TABLE tenant_2.test_table4(id int);
+ALTER TABLE tenant_2.test_table4 SET SCHEMA tenant_3;
+-- verify that tenant_3.test_table4 is recorded in pg_dist_partition as a single-shard table.
+SELECT COUNT(*)=1 FROM pg_dist_partition
+WHERE logicalrelid = 'tenant_3.test_table4'::regclass AND
+      partmethod = 'n' AND repmodel = 's' AND colocationid > 0;
+-- verify that tenant_2.test_table4 does not exist
+SELECT * FROM tenant_2.test_table4;
+
+-- verify that we can put a local table in regular schema into distributed schema
+CREATE TABLE regular_schema.pg_local_tbl(id int);
+ALTER TABLE regular_schema.pg_local_tbl SET SCHEMA tenant_2;
+
+-- verify that we can put a Citus local table in regular schema into distributed schema
+CREATE TABLE regular_schema.citus_local_tbl(id int);
+SELECT citus_add_local_table_to_metadata('regular_schema.citus_local_tbl');
+ALTER TABLE regular_schema.citus_local_tbl SET SCHEMA tenant_2;
+
+-- verify that we do not allow a hash distributed table in regular schema into distributed schema
+CREATE TABLE regular_schema.hash_dist_tbl(id int);
+SELECT create_distributed_table('regular_schema.hash_dist_tbl', 'id');
+ALTER TABLE regular_schema.hash_dist_tbl SET SCHEMA tenant_2;
+
+-- verify that we do not allow a reference table in regular schema into distributed schema
+CREATE TABLE regular_schema.ref_tbl(id int  PRIMARY KEY);
+SELECT create_reference_table('regular_schema.ref_tbl');
+ALTER TABLE regular_schema.ref_tbl SET SCHEMA tenant_2;
+
+-- verify that we can put a table in tenant schema into regular schema
+CREATE TABLE tenant_2.tenant_tbl(id int);
+ALTER TABLE tenant_2.tenant_tbl SET SCHEMA regular_schema;
+
+-- verify that we can put a table in tenant schema into another tenant schema
+CREATE TABLE tenant_2.tenant_tbl2(id int);
+ALTER TABLE tenant_2.tenant_tbl2 SET SCHEMA tenant_3;
+
+-- verify that we do not allow a local table in regular schema into distributed schema if it has foreign key to a non-reference table in another schema
+CREATE TABLE regular_schema.pg_local_tbl1(id int PRIMARY KEY);
+CREATE TABLE regular_schema.pg_local_tbl2(id int REFERENCES regular_schema.pg_local_tbl1(id));
+ALTER TABLE regular_schema.pg_local_tbl2 SET SCHEMA tenant_2;
+
+-- verify that we allow a local table in regular schema into distributed schema if it has foreign key to a reference table in another schema
+CREATE TABLE regular_schema.pg_local_tbl3(id int REFERENCES regular_schema.ref_tbl(id));
+ALTER TABLE regular_schema.pg_local_tbl3 SET SCHEMA tenant_2;
+
+-- verify that we do not allow a table in tenant schema into regular schema if it has foreign key to/from another table in the same schema
+CREATE TABLE tenant_2.tenant_tbl1(id int PRIMARY KEY);
+CREATE TABLE tenant_2.tenant_tbl2(id int REFERENCES tenant_2.tenant_tbl1(id));
+ALTER TABLE tenant_2.tenant_tbl1 SET SCHEMA regular_schema;
+ALTER TABLE tenant_2.tenant_tbl2 SET SCHEMA regular_schema;
+
+-- verify that we do not allow a table in distributed schema into another distributed schema if it has foreign key to/from another table in the same schema
+CREATE TABLE tenant_2.tenant_tbl3(id int PRIMARY KEY);
+CREATE TABLE tenant_2.tenant_tbl4(id int REFERENCES tenant_2.tenant_tbl3(id));
+ALTER TABLE tenant_2.tenant_tbl3 SET SCHEMA tenant_3;
+ALTER TABLE tenant_2.tenant_tbl4 SET SCHEMA tenant_3;
+
+-- alter set non-existent schema
+ALTER TABLE tenant_2.test_table SET SCHEMA ghost_schema;
+ALTER TABLE IF EXISTS tenant_2.test_table SET SCHEMA ghost_schema;
+-- alter set non-existent table
+ALTER TABLE tenant_2.ghost_table SET SCHEMA ghost_schema;
+ALTER TABLE IF EXISTS tenant_2.ghost_table SET SCHEMA ghost_schema;
 
 -- (on coordinator) verify that colocation id is set for empty tenants too
 SELECT colocationid > 0 FROM pg_dist_schema