Add:

1) Drop colocated shards
2) Swap Metadata
3) Drop dummy shards

src/backend/distributed/operations/citus_split_shard_by_split_points.c

@@ -81,6 +81,7 @@ LookupSplitMode(Oid shardTransferModeOid)
 	{
 		shardSplitMode = BLOCKING_SPLIT;
 	}
+
 	/* Extend with other modes as we support them */
 	else if (strncmp(enumLabel, "auto", NAMEDATALEN) == 0 ||
 			 strncmp(enumLabel, "force_logical", NAMEDATALEN) == 0)

src/backend/distributed/operations/shard_split.c

@@ -94,6 +94,10 @@ static void TryDropSplitShardsOnFailure(List *shardGroupSplitIntervalListList,
 static Task * CreateTaskForDDLCommandList(uint64 jobId, List *ddlCommandList,
 										  WorkerNode *workerNode);
 
+static void AddDummyShardEntryInMap(uint32 targetNodeId, ShardInterval *shardInterval);
+static void DropDummyShards(void);
+void TryDropShard(MultiConnection *connection, ShardInterval *shardInterval);
+
 /* Customize error message strings based on operation type */
 static const char *const SplitOperationName[] =
 {
@@ -106,6 +110,9 @@ static const char *const SplitTargetName[] =
 	[ISOLATE_TENANT_TO_NEW_SHARD] = "tenant",
 };
 
+/* Map containing list of dummy shards created on target nodes */
+static HTAB *DummyShardInfoHashMap = NULL;
+
 /* Function definitions */
 
 /*
@@ -1039,7 +1046,7 @@ TryDropSplitShardsOnFailure(List *shardGroupSplitIntervalListList,
 
 			int connectionFlags = FOR_DDL;
 			connectionFlags |= OUTSIDE_TRANSACTION;
-			MultiConnection *connnection = GetNodeUserDatabaseConnection(
+			MultiConnection *connection = GetNodeUserDatabaseConnection(
 				connectionFlags,
 				workerPlacementNode->workerName,
 				workerPlacementNode->workerPort,
@@ -1051,13 +1058,14 @@ TryDropSplitShardsOnFailure(List *shardGroupSplitIntervalListList,
 			 * The shard may or may not exist and the connection could have died.
 			 */
 			ExecuteOptionalRemoteCommand(
-				connnection,
+				connection,
 				dropShardQuery->data,
 				NULL /* pgResult */);
 		}
 	}
 }
 
+
 /*
  * SplitShard API to split a given shard (or shard group) in blocking fashion
  * based on specified split points to a set of destination nodes.
@@ -1107,12 +1115,33 @@ NonBlockingShardSplit(SplitOperation splitOperation,
 			sourceShardToCopyNode,
 			workersForPlacementList);
 
+		/*TODO: Refactor this method. BlockWrites is within this as of now, take it out */
 		SplitShardReplicationSetup(
 			sourceColocatedShardIntervalList,
 			shardGroupSplitIntervalListList,
 			sourceShardToCopyNode,
 			workersForPlacementList);
 
+		/*
+		 * Drop old shards and delete related metadata. Have to do that before
+		 * creating the new shard metadata, because there's cross-checks
+		 * preventing inconsistent metadata (like overlapping shards).
+		 */
+		DropShardList(sourceColocatedShardIntervalList);
+
+		/* Insert new shard and placement metdata */
+		InsertSplitChildrenShardMetadata(shardGroupSplitIntervalListList,
+										 workersForPlacementList);
+
+		/*
+		 * Create foreign keys if exists after the metadata changes happening in
+		 * DropShardList() and InsertSplitChildrenShardMetadata() because the foreign
+		 * key creation depends on the new metadata.
+		 */
+		CreateForeignKeyConstraints(shardGroupSplitIntervalListList,
+									workersForPlacementList);
+
+		DropDummyShards();
 	}
 	PG_CATCH();
 	{
@@ -1120,11 +1149,14 @@ NonBlockingShardSplit(SplitOperation splitOperation,
 		TryDropSplitShardsOnFailure(shardGroupSplitIntervalListList,
 									workersForPlacementList);
 
+		DropDummyShards();
+
 		PG_RE_THROW();
 	}
 	PG_END_TRY();
 }
 
+
 /* Create ShardGroup split children on a list of corresponding workers. */
 static void
 CreateSplitShardsForShardGroupTwo(WorkerNode *sourceShardNode,
@@ -1151,18 +1183,25 @@ CreateSplitShardsForShardGroupTwo(WorkerNode *sourceShardNode,
 				splitShardCreationCommandList,
 				shardInterval->shardId);
 
-			/* Create new split child shard on the specified placement list */
+			/* Create new split child shard on the specified worker node */
 			CreateObjectOnPlacement(splitShardCreationCommandList, workerPlacementNode);
 		}
 	}
 }
 
+
 static void
 CreateDummyShardsForShardGroup(List *sourceColocatedShardIntervalList,
 							   List *shardGroupSplitIntervalListList,
 							   WorkerNode *sourceWorkerNode,
 							   List *workersForPlacementList)
 {
+	/*
+	 * Setup a hash map to store list of dummy shards created on nodes.
+	 * This will facilitate easy cleanup.
+	 */
+	DummyShardInfoHashMap = SetupHashMapForShardInfo();
+
 	/*
 	 * Statisfy Constraint 1: Create dummy source shard(s) on all destination nodes.
 	 * If source node is also in desintation, skip dummy shard creation(see Note 1 from function description).
@@ -1194,8 +1233,11 @@ CreateDummyShardsForShardGroup(List *sourceColocatedShardIntervalList,
 				splitShardCreationCommandList,
 				shardInterval->shardId);
 
-			/* Create new split child shard on the specified placement list */
+			/* Create dummy source shard on the specified placement list */
 			CreateObjectOnPlacement(splitShardCreationCommandList, workerPlacementNode);
+
+			/* Add dummy source shard entry created for placement node in map */
+			AddDummyShardEntryInMap(workerPlacementNode->nodeId, shardInterval);
 		}
 	}
 
@@ -1224,12 +1266,16 @@ CreateDummyShardsForShardGroup(List *sourceColocatedShardIntervalList,
 				splitShardCreationCommandList,
 				shardInterval->shardId);
 
-			/* Create new split child shard on the specified placement list */
+			/* Create dummy split child shard on source worker node */
 			CreateObjectOnPlacement(splitShardCreationCommandList, sourceWorkerNode);
+
+			/* Add dummy split child shard entry created on source node */
+			AddDummyShardEntryInMap(sourceWorkerNode->nodeId, shardInterval);
 		}
 	}
 }
 
+
 static HTAB *
 CreateWorkerForPlacementSet(List *workersForPlacementList)
 {
@@ -1257,13 +1303,14 @@ CreateWorkerForPlacementSet(List *workersForPlacementList)
 }
 
 
-static void SplitShardReplicationSetup(List *sourceColocatedShardIntervalList,
+static void
+SplitShardReplicationSetup(List *sourceColocatedShardIntervalList,
 						   List *shardGroupSplitIntervalListList,
 						   WorkerNode *sourceWorkerNode,
 						   List *destinationWorkerNodesList)
 {
-
+	StringInfo splitShardReplicationUDF = CreateSplitShardReplicationSetupUDF(
-	StringInfo splitShardReplicationUDF = CreateSplitShardReplicationSetupUDF(sourceColocatedShardIntervalList,
+		sourceColocatedShardIntervalList,
 		shardGroupSplitIntervalListList,
 		destinationWorkerNodesList);
 
@@ -1279,7 +1326,9 @@ static void SplitShardReplicationSetup(List *sourceColocatedShardIntervalList,
 	ClaimConnectionExclusively(sourceConnection);
 
 	PGresult *result = NULL;
-	int queryResult = ExecuteOptionalRemoteCommand(sourceConnection, splitShardReplicationUDF->data, &result);
+	int queryResult = ExecuteOptionalRemoteCommand(sourceConnection,
+												   splitShardReplicationUDF->data,
+												   &result);
 	if (queryResult != RESPONSE_OKAY || !IsResponseOK(result))
 	{
 		ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE),
@@ -1292,12 +1341,89 @@ static void SplitShardReplicationSetup(List *sourceColocatedShardIntervalList,
 	/* Get replication slot information */
 	List *replicationSlotInfoList = ParseReplicationSlotInfoFromResult(result);
 
-	List * shardSplitPubSubMetadata = CreateShardSplitPubSubMetadataList(sourceColocatedShardIntervalList,
+	List *shardSplitPubSubMetadata = CreateShardSplitPubSubMetadataList(
+		sourceColocatedShardIntervalList,
 		shardGroupSplitIntervalListList,
 		destinationWorkerNodesList,
 		replicationSlotInfoList);
 
 
-	LogicallyReplicateSplitShards(sourceWorkerNode, shardSplitPubSubMetadata, sourceColocatedShardIntervalList, shardGroupSplitIntervalListList, destinationWorkerNodesList);
+	LogicallyReplicateSplitShards(sourceWorkerNode, shardSplitPubSubMetadata,
+								  sourceColocatedShardIntervalList,
+								  shardGroupSplitIntervalListList,
+								  destinationWorkerNodesList);
 }
 
+
+static void
+AddDummyShardEntryInMap(uint32 targetNodeId, ShardInterval *shardInterval)
+{
+	NodeShardMappingKey key;
+	key.nodeId = targetNodeId;
+	key.tableOwnerId = TableOwnerOid(shardInterval->relationId);
+
+	bool found = false;
+	NodeShardMappingEntry *nodeMappingEntry =
+		(NodeShardMappingEntry *) hash_search(DummyShardInfoHashMap, &key, HASH_ENTER,
+											  &found);
+	if (!found)
+	{
+		nodeMappingEntry->shardSplitInfoList = NIL;
+	}
+
+	nodeMappingEntry->shardSplitInfoList =
+		lappend(nodeMappingEntry->shardSplitInfoList, (ShardInterval *) shardInterval);
+}
+
+
+static void
+DropDummyShards()
+{
+	HASH_SEQ_STATUS status;
+	hash_seq_init(&status, DummyShardInfoHashMap);
+
+	NodeShardMappingEntry *entry = NULL;
+	while ((entry = (NodeShardMappingEntry *) hash_seq_search(&status)) != NULL)
+	{
+		uint32 nodeId = entry->key.nodeId;
+		WorkerNode *shardToBeDroppedNode = FindNodeWithNodeId(nodeId,
+															  false /* missingOk */);
+
+		int connectionFlags = FOR_DDL;
+		connectionFlags |= OUTSIDE_TRANSACTION;
+		MultiConnection *connection = GetNodeUserDatabaseConnection(
+			connectionFlags,
+			shardToBeDroppedNode->workerName,
+			shardToBeDroppedNode->workerPort,
+			CurrentUserName(),
+			NULL /* databaseName */);
+
+		List *dummyShardIntervalList = entry->shardSplitInfoList;
+		ShardInterval *shardInterval = NULL;
+		foreach_ptr(shardInterval, dummyShardIntervalList)
+		{
+			TryDropShard(connection, shardInterval);
+		}
+	}
+}
+
+
+void
+TryDropShard(MultiConnection *connection, ShardInterval *shardInterval)
+{
+	char *qualifiedShardName = ConstructQualifiedShardName(shardInterval);
+	StringInfo dropShardQuery = makeStringInfo();
+
+	/* Caller enforces that foreign tables cannot be split (use DROP_REGULAR_TABLE_COMMAND) */
+	appendStringInfo(dropShardQuery, DROP_REGULAR_TABLE_COMMAND,
+					 qualifiedShardName);
+
+	/*
+	 * Perform a drop in best effort manner.
+	 * The shard may or may not exist and the connection could have died.
+	 */
+	ExecuteOptionalRemoteCommand(
+		connection,
+		dropShardQuery->data,
+		NULL /* pgResult */);
+}

src/backend/distributed/operations/split_shard_replication_setup.c

@@ -48,7 +48,8 @@ static void AddShardSplitInfoEntryForNodeInMap(ShardSplitInfo *shardSplitInfo);
 static void PopulateShardSplitInfoInSM(ShardSplitInfoSMHeader *shardSplitInfoSMHeader,
 									   HTAB *shardInfoHashMap);
 
-static void ReturnReplicationSlotInfo(HTAB *shardInfoHashMap, Tuplestorestate *tupleStore, TupleDesc tupleDescriptor);
+static void ReturnReplicationSlotInfo(HTAB *shardInfoHashMap, Tuplestorestate *tupleStore,
+									  TupleDesc tupleDescriptor);
 
 static void CreatePublishersForSplitChildren(HTAB *shardInfoHashMap);
 StringInfo GetSoureAndDestinationShardNames(List *shardSplitInfoList);
@@ -420,9 +421,10 @@ ParseShardSplitInfoFromDatum(Datum shardSplitInfoDatum,
 	*nodeId = DatumGetInt32(nodeIdDatum);
 }
 
-static void CreatePublishersForSplitChildren(HTAB *shardInfoHashMap)
-{
 
+static void
+CreatePublishersForSplitChildren(HTAB *shardInfoHashMap)
+{
 	HASH_SEQ_STATUS status;
 	hash_seq_init(&status, shardInfoHashMap);
 
@@ -441,16 +443,20 @@ static void CreatePublishersForSplitChildren(HTAB *shardInfoHashMap)
 																		  CitusExtensionOwnerName(),
 																		  get_database_name(
 																			  MyDatabaseId));
-		StringInfo shardNamesForPublication = GetSoureAndDestinationShardNames(entry->shardSplitInfoList);
+		StringInfo shardNamesForPublication = GetSoureAndDestinationShardNames(
+			entry->shardSplitInfoList);
 
 		StringInfo command = makeStringInfo();
-		appendStringInfo(command, "CREATE PUBLICATION sameerpub_%u_%u FOR TABLE %s", nodeId, tableOwnerId,shardNamesForPublication->data);
+		appendStringInfo(command, "CREATE PUBLICATION sameerpub_%u_%u FOR TABLE %s",
+						 nodeId, tableOwnerId, shardNamesForPublication->data);
 		ExecuteCriticalRemoteCommand(sourceConnection, command->data);
 		printf("Sameer UserName: %s \n", GetUserNameFromId(tableOwnerId, false));
 	}
 }
 
-StringInfo GetSoureAndDestinationShardNames(List* shardSplitInfoList)
+
+StringInfo
+GetSoureAndDestinationShardNames(List *shardSplitInfoList)
 {
 	HASHCTL info;
 	int flags = HASH_ELEM | HASH_CONTEXT;
@@ -507,6 +513,7 @@ StringInfo GetSoureAndDestinationShardNames(List* shardSplitInfoList)
 	return allShardNames;
 }
 
+
 char *
 ConstructFullyQualifiedSplitChildShardName(ShardSplitInfo *shardSplitInfo)
 {
@@ -521,7 +528,10 @@ ConstructFullyQualifiedSplitChildShardName(ShardSplitInfo* shardSplitInfo)
 	return shardName;
 }
 
-static void ReturnReplicationSlotInfo(HTAB *shardInfoHashMap, Tuplestorestate *tupleStore, TupleDesc tupleDescriptor)
+
+static void
+ReturnReplicationSlotInfo(HTAB *shardInfoHashMap, Tuplestorestate *tupleStore, TupleDesc
+						  tupleDescriptor)
 {
 	HASH_SEQ_STATUS status;
 	hash_seq_init(&status, shardInfoHashMap);
@@ -540,7 +550,8 @@ static void ReturnReplicationSlotInfo(HTAB *shardInfoHashMap, Tuplestorestate *t
 		char *tableOwnerName = GetUserNameFromId(entry->key.tableOwnerId, false);
 		values[1] = CStringGetTextDatum(tableOwnerName);
 
-		char * slotName = encode_replication_slot(entry->key.nodeId, entry->key.tableOwnerId);
+		char *slotName = encode_replication_slot(entry->key.nodeId,
+												 entry->key.tableOwnerId);
 		values[2] = CStringGetTextDatum(slotName);
 
 		tuplestore_putvalues(tupleStore, tupleDescriptor, values, nulls);

src/backend/distributed/replication/multi_logical_replication.c

@@ -136,10 +136,12 @@ static XLogRecPtr GetRemoteLSN(MultiConnection *connection, char *command);
 static uint64 TotalRelationSizeForSubscription(MultiConnection *connection,
 											   char *command);
 static bool RelationSubscriptionsAreReady(MultiConnection *targetConnection,
-										  Bitmapset *tableOwnerIds, char * operationPrefix);
+										  Bitmapset *tableOwnerIds,
+										  char *operationPrefix);
 static void WaitForMiliseconds(long timeout);
 static XLogRecPtr GetSubscriptionPosition(MultiConnection *connection,
-										  Bitmapset *tableOwnerIds, char * operationPrefix);
+										  Bitmapset *tableOwnerIds,
+										  char *operationPrefix);
 static char * ShardMovePublicationName(Oid ownerId);
 static char * ShardSubscriptionName(Oid ownerId, char *operationPrefix);
 static void AcquireLogicalReplicationLock(void);
@@ -148,7 +150,8 @@ static void DropAllShardMoveSubscriptions(MultiConnection *connection);
 static void DropAllShardMoveReplicationSlots(MultiConnection *connection);
 static void DropAllShardMovePublications(MultiConnection *connection);
 static void DropAllShardMoveUsers(MultiConnection *connection);
-static char * ShardSubscriptionNamesValueList(Bitmapset *tableOwnerIds, char * operationPrefix);
+static char * ShardSubscriptionNamesValueList(Bitmapset *tableOwnerIds,
+											  char *operationPrefix);
 static void DropShardMoveSubscription(MultiConnection *connection,
 									  char *subscriptionName);
 static void DropShardMoveReplicationSlot(MultiConnection *connection,
@@ -227,14 +230,16 @@ LogicallyReplicateShards(List *shardList, char *sourceNodeName, int sourceNodePo
 		 * subscription is not ready. There is no point of locking the shards before the
 		 * subscriptions for each relation becomes ready, so wait for it.
 		 */
-		WaitForRelationSubscriptionsBecomeReady(targetConnection, tableOwnerIds, SHARD_MOVE_SUBSCRIPTION_PREFIX);
+		WaitForRelationSubscriptionsBecomeReady(targetConnection, tableOwnerIds,
+												SHARD_MOVE_SUBSCRIPTION_PREFIX);
 
 		/*
 		 * Wait until the subscription is caught up to changes that has happened
 		 * after the initial COPY on the shards.
 		 */
 		XLogRecPtr sourcePosition = GetRemoteLogPosition(sourceConnection);
-		WaitForShardSubscriptionToCatchUp(targetConnection, sourcePosition, tableOwnerIds, SHARD_MOVE_SUBSCRIPTION_PREFIX);
+		WaitForShardSubscriptionToCatchUp(targetConnection, sourcePosition, tableOwnerIds,
+										  SHARD_MOVE_SUBSCRIPTION_PREFIX);
 
 		/*
 		 * Now lets create the post-load objects, such as the indexes, constraints
@@ -244,7 +249,8 @@ LogicallyReplicateShards(List *shardList, char *sourceNodeName, int sourceNodePo
 		CreatePostLogicalReplicationDataLoadObjects(shardList, targetNodeName,
 													targetNodePort);
 		sourcePosition = GetRemoteLogPosition(sourceConnection);
-		WaitForShardSubscriptionToCatchUp(targetConnection, sourcePosition, tableOwnerIds, SHARD_MOVE_SUBSCRIPTION_PREFIX);
+		WaitForShardSubscriptionToCatchUp(targetConnection, sourcePosition, tableOwnerIds,
+										  SHARD_MOVE_SUBSCRIPTION_PREFIX);
 
 		/*
 		 * We're almost done, we'll block the writes to the shards that we're
@@ -257,7 +263,8 @@ LogicallyReplicateShards(List *shardList, char *sourceNodeName, int sourceNodePo
 		BlockWritesToShardList(shardList);
 
 		sourcePosition = GetRemoteLogPosition(sourceConnection);
-		WaitForShardSubscriptionToCatchUp(targetConnection, sourcePosition, tableOwnerIds, SHARD_MOVE_SUBSCRIPTION_PREFIX);
+		WaitForShardSubscriptionToCatchUp(targetConnection, sourcePosition, tableOwnerIds,
+										  SHARD_MOVE_SUBSCRIPTION_PREFIX);
 
 		/*
 		 * We're creating the foreign constraints to reference tables after the
@@ -1085,7 +1092,8 @@ DropShardMovePublications(MultiConnection *connection, Bitmapset *tableOwnerIds)
 		 * If replication slot can not be dropped while dropping the subscriber, drop
 		 * it here.
 		 */
-		DropShardMoveReplicationSlot(connection, ShardSubscriptionName(ownerId, SHARD_MOVE_SUBSCRIPTION_PREFIX));
+		DropShardMoveReplicationSlot(connection, ShardSubscriptionName(ownerId,
+																	   SHARD_MOVE_SUBSCRIPTION_PREFIX));
 		DropShardMovePublication(connection, ShardMovePublicationName(ownerId));
 	}
 }
@@ -1317,7 +1325,8 @@ DropShardMoveSubscriptions(MultiConnection *connection, Bitmapset *tableOwnerIds
 	int ownerId = -1;
 	while ((ownerId = bms_next_member(tableOwnerIds, ownerId)) >= 0)
 	{
-		DropShardMoveSubscription(connection, ShardSubscriptionName(ownerId, SHARD_MOVE_SUBSCRIPTION_PREFIX));
+		DropShardMoveSubscription(connection, ShardSubscriptionName(ownerId,
+																	SHARD_MOVE_SUBSCRIPTION_PREFIX));
 		DropShardMoveUser(connection, ShardMoveSubscriptionRole(ownerId));
 	}
 }
@@ -1491,7 +1500,8 @@ CreateShardMoveSubscriptions(MultiConnection *connection, char *sourceNodeName,
 		appendStringInfo(createSubscriptionCommand,
 						 "CREATE SUBSCRIPTION %s CONNECTION %s PUBLICATION %s "
 						 "WITH (citus_use_authinfo=true, enabled=false)",
-						 quote_identifier(ShardSubscriptionName(ownerId, SHARD_MOVE_SUBSCRIPTION_PREFIX)),
+						 quote_identifier(ShardSubscriptionName(ownerId,
+																SHARD_MOVE_SUBSCRIPTION_PREFIX)),
 						 quote_literal_cstr(conninfo->data),
 						 quote_identifier(ShardMovePublicationName(ownerId)));
 
@@ -1500,7 +1510,8 @@ CreateShardMoveSubscriptions(MultiConnection *connection, char *sourceNodeName,
 		pfree(createSubscriptionCommand);
 		ExecuteCriticalRemoteCommand(connection, psprintf(
 										 "ALTER SUBSCRIPTION %s OWNER TO %s",
-										 ShardSubscriptionName(ownerId, SHARD_MOVE_SUBSCRIPTION_PREFIX),
+										 ShardSubscriptionName(ownerId,
+															   SHARD_MOVE_SUBSCRIPTION_PREFIX),
 										 ShardMoveSubscriptionRole(ownerId)
 										 ));
 
@@ -1519,7 +1530,8 @@ CreateShardMoveSubscriptions(MultiConnection *connection, char *sourceNodeName,
 
 		ExecuteCriticalRemoteCommand(connection, psprintf(
 										 "ALTER SUBSCRIPTION %s ENABLE",
-										 ShardSubscriptionName(ownerId, SHARD_MOVE_SUBSCRIPTION_PREFIX)
+										 ShardSubscriptionName(ownerId,
+															   SHARD_MOVE_SUBSCRIPTION_PREFIX)
 										 ));
 	}
 }
@@ -1655,7 +1667,8 @@ WaitForRelationSubscriptionsBecomeReady(MultiConnection *targetConnection,
 	while (true)
 	{
 		/* we're done, all relations are ready */
-		if (RelationSubscriptionsAreReady(targetConnection, tableOwnerIds, operationPrefix))
+		if (RelationSubscriptionsAreReady(targetConnection, tableOwnerIds,
+										  operationPrefix))
 		{
 			ereport(LOG, (errmsg("The states of the relations belonging to the "
 								 "subscriptions became READY on the "
@@ -1665,7 +1678,8 @@ WaitForRelationSubscriptionsBecomeReady(MultiConnection *targetConnection,
 
 			break;
 		}
-		char *subscriptionValueList = ShardSubscriptionNamesValueList(tableOwnerIds, operationPrefix);
+		char *subscriptionValueList = ShardSubscriptionNamesValueList(tableOwnerIds,
+																	  operationPrefix);
 
 		/* Get the current total size of tables belonging to the subscriber */
 		uint64 currentTotalRelationSize =
@@ -1842,7 +1856,8 @@ ShardSubscriptionNamesValueList(Bitmapset *tableOwnerIds, char * operationPrefix
 			first = false;
 		}
 		appendStringInfoString(subscriptionValueList,
-							   quote_literal_cstr(ShardSubscriptionName(ownerId, operationPrefix)));
+							   quote_literal_cstr(ShardSubscriptionName(ownerId,
+																		operationPrefix)));
 	}
 	appendStringInfoString(subscriptionValueList, ")");
 	return subscriptionValueList->data;
@@ -1859,7 +1874,8 @@ RelationSubscriptionsAreReady(MultiConnection *targetConnection,
 {
 	bool raiseInterrupts = false;
 
-	char *subscriptionValueList = ShardSubscriptionNamesValueList(tableOwnerIds, operationPrefix);
+	char *subscriptionValueList = ShardSubscriptionNamesValueList(tableOwnerIds,
+																  operationPrefix);
 	char *query = psprintf(
 		"SELECT count(*) FROM pg_subscription_rel, pg_stat_subscription "
 		"WHERE srsubid = subid AND srsubstate != 'r' AND subname IN %s",
@@ -1910,7 +1926,8 @@ RelationSubscriptionsAreReady(MultiConnection *targetConnection,
  * The function also reports its progress in every logicalReplicationProgressReportTimeout.
  */
 void
-WaitForShardSubscriptionToCatchUp(MultiConnection *targetConnection, XLogRecPtr sourcePosition,
+WaitForShardSubscriptionToCatchUp(MultiConnection *targetConnection, XLogRecPtr
+								  sourcePosition,
 								  Bitmapset *tableOwnerIds, char *operationPrefix)
 {
 	XLogRecPtr previousTargetPosition = 0;
@@ -2050,9 +2067,11 @@ WaitForMiliseconds(long timeout)
  * replication.
  */
 static XLogRecPtr
-GetSubscriptionPosition(MultiConnection *connection, Bitmapset *tableOwnerIds, char * operationPrefix)
+GetSubscriptionPosition(MultiConnection *connection, Bitmapset *tableOwnerIds,
+						char *operationPrefix)
 {
-	char *subscriptionValueList = ShardSubscriptionNamesValueList(tableOwnerIds, operationPrefix);
+	char *subscriptionValueList = ShardSubscriptionNamesValueList(tableOwnerIds,
+																  operationPrefix);
 	return GetRemoteLSN(connection, psprintf(
 							"SELECT min(latest_end_lsn) FROM pg_stat_subscription "
 							"WHERE subname IN %s", subscriptionValueList));
@@ -2066,7 +2085,6 @@ CreateShardSubscription(MultiConnection *connection, char *sourceNodeName,
 						char *publicationName,
 						Oid ownerId)
 {
-
 	StringInfo createSubscriptionCommand = makeStringInfo();
 	StringInfo conninfo = makeStringInfo();
 
@@ -2092,7 +2110,8 @@ CreateShardSubscription(MultiConnection *connection, char *sourceNodeName,
 	appendStringInfo(createSubscriptionCommand,
 					 "CREATE SUBSCRIPTION %s CONNECTION %s PUBLICATION %s "
 					 "WITH (citus_use_authinfo=true, enabled=false)",
-                        quote_identifier(ShardSubscriptionName(ownerId, SHARD_SPLIT_SUBSCRIPTION_PREFIX)),
+					 quote_identifier(ShardSubscriptionName(ownerId,
+															SHARD_SPLIT_SUBSCRIPTION_PREFIX)),
 					 quote_literal_cstr(conninfo->data),
 					 quote_identifier(publicationName));
 
@@ -2101,7 +2120,8 @@ CreateShardSubscription(MultiConnection *connection, char *sourceNodeName,
 	pfree(createSubscriptionCommand);
 	ExecuteCriticalRemoteCommand(connection, psprintf(
 									 "ALTER SUBSCRIPTION %s OWNER TO %s",
-                                        ShardSubscriptionName(ownerId, SHARD_SPLIT_SUBSCRIPTION_PREFIX),
+									 ShardSubscriptionName(ownerId,
+														   SHARD_SPLIT_SUBSCRIPTION_PREFIX),
 									 ShardMoveSubscriptionRole(ownerId)
 									 ));
 
@@ -2120,6 +2140,7 @@ CreateShardSubscription(MultiConnection *connection, char *sourceNodeName,
 
 	ExecuteCriticalRemoteCommand(connection, psprintf(
 									 "ALTER SUBSCRIPTION %s ENABLE",
-                                        ShardSubscriptionName(ownerId, SHARD_SPLIT_SUBSCRIPTION_PREFIX)
+									 ShardSubscriptionName(ownerId,
+														   SHARD_SPLIT_SUBSCRIPTION_PREFIX)
 									 ));
 }

src/backend/distributed/shardsplit/shardsplit_logical_replication.c

@@ -26,30 +26,39 @@
 static HTAB *ShardInfoHashMapForPublications = NULL;
 
 /* function declarations */
-static void AddShardEntryInMap(Oid tableOwner, uint32 nodeId, ShardInterval * shardInterval, bool isChildShardInterval);
+static void AddPublishableShardEntryInMap(uint32 targetNodeId,
-ShardSplitPubSubMetadata * CreateShardSplitPubSubMetadata(Oid tableOwnerId, uint32 nodeId, List * shardIdList, List * replicationSlotInfoList);
+										  ShardInterval *shardInterval, bool
+										  isChildShardInterval);
+ShardSplitPubSubMetadata * CreateShardSplitPubSubMetadata(Oid tableOwnerId, uint32 nodeId,
+														  List *shardIdList,
+														  List *replicationSlotInfoList);
 
-static void
+static void CreateShardSplitPublicationForNode(MultiConnection *connection,
-CreateShardSplitPublicationForNode(MultiConnection *connection, List *shardList,
+											   List *shardList,
-							uint32_t publicationForTargetNodeId, Oid tableOwner);
+											   uint32_t publicationForTargetNodeId, Oid
-static void
+											   tableOwner);
-CreateShardSplitPublications(MultiConnection *sourceConnection, List * shardSplitPubSubMetadataList);
+static void CreateShardSplitPublications(MultiConnection *sourceConnection,
-static void
+										 List *shardSplitPubSubMetadataList);
-CreateShardSplitSubscriptions(List * targetNodeConnectionList, List * shardSplitPubSubMetadataList, WorkerNode * sourceWorkerNode, char * superUser, char * databaseName);
+static void CreateShardSplitSubscriptions(List *targetNodeConnectionList,
-static void
+										  List *shardSplitPubSubMetadataList,
-WaitForShardSplitRelationSubscriptionsBecomeReady(List * targetNodeConnectionList, List * shardSplitPubSubMetadataList);
+										  WorkerNode *sourceWorkerNode, char *superUser,
+										  char *databaseName);
+static void WaitForShardSplitRelationSubscriptionsBecomeReady(
+	List *targetNodeConnectionList, List *shardSplitPubSubMetadataList);
 
-static void
+static void WaitForShardSplitRelationSubscriptionsToBeCaughtUp(XLogRecPtr sourcePosition,
-WaitForShardSplitRelationSubscriptionsToBeCaughtUp(XLogRecPtr sourcePosition, List * targetNodeConnectionList, List * shardSplitPubSubMetadataList);
+															   List *
+															   targetNodeConnectionList,
+															   List *
+															   shardSplitPubSubMetadataList);
 
-
+static char * ShardSplitPublicationName(uint32_t nodeId, Oid ownerId);
-static char *
-ShardSplitPublicationName(uint32_t nodeId, Oid ownerId);
 
 /*used for debuggin. Remove later*/
 void PrintShardSplitPubSubMetadata(ShardSplitPubSubMetadata *shardSplitMetadata);
 
-StringInfo CreateSplitShardReplicationSetupUDF(List *sourceColocatedShardIntervalList,
+StringInfo
+CreateSplitShardReplicationSetupUDF(List *sourceColocatedShardIntervalList,
 									List *shardGroupSplitIntervalListList,
 									List *destinationWorkerNodesList)
 {
@@ -74,10 +83,12 @@ StringInfo CreateSplitShardReplicationSetupUDF(List *sourceColocatedShardInterva
 			}
 
 			StringInfo minValueString = makeStringInfo();
-			appendStringInfo(minValueString, "%d", DatumGetInt32(splitChildShardInterval->minValue));
+			appendStringInfo(minValueString, "%d", DatumGetInt32(
+								 splitChildShardInterval->minValue));
 
 			StringInfo maxValueString = makeStringInfo();
-			appendStringInfo(maxValueString, "%d", DatumGetInt32(splitChildShardInterval->maxValue));
+			appendStringInfo(maxValueString, "%d", DatumGetInt32(
+								 splitChildShardInterval->maxValue));
 
 			appendStringInfo(splitChildrenRows,
 							 "ROW(%lu, %lu, %s, %s, %u)::citus.split_shard_info",
@@ -93,12 +104,15 @@ StringInfo CreateSplitShardReplicationSetupUDF(List *sourceColocatedShardInterva
 
 	StringInfo splitShardReplicationUDF = makeStringInfo();
 	appendStringInfo(splitShardReplicationUDF,
-		"SELECT * FROM worker_split_shard_replication_setup(ARRAY[%s])", splitChildrenRows->data);
+					 "SELECT * FROM worker_split_shard_replication_setup(ARRAY[%s])",
+					 splitChildrenRows->data);
 
 	return splitShardReplicationUDF;
 }
 
-List * ParseReplicationSlotInfoFromResult(PGresult * result)
+
+List *
+ParseReplicationSlotInfoFromResult(PGresult *result)
 {
 	int64 rowCount = PQntuples(result);
 	int64 colCount = PQnfields(result);
@@ -106,7 +120,8 @@ List * ParseReplicationSlotInfoFromResult(PGresult * result)
 	List *replicationSlotInfoList = NIL;
 	for (int64 rowIndex = 0; rowIndex < rowCount; rowIndex++)
 	{
-        ReplicationSlotInfo * replicationSlotInfo = (ReplicationSlotInfo *)palloc0(sizeof(ReplicationSlotInfo));
+		ReplicationSlotInfo *replicationSlotInfo = (ReplicationSlotInfo *) palloc0(
+			sizeof(ReplicationSlotInfo));
 
 		char *targeNodeIdString = PQgetvalue(result, rowIndex, 0);
 		replicationSlotInfo->targetNodeId = strtoul(targeNodeIdString, NULL, 10);
@@ -124,7 +139,8 @@ List * ParseReplicationSlotInfoFromResult(PGresult * result)
 }
 
 
-List * CreateShardSplitPubSubMetadataList(List *sourceColocatedShardIntervalList,
+List *
+CreateShardSplitPubSubMetadataList(List *sourceColocatedShardIntervalList,
 								   List *shardGroupSplitIntervalListList,
 								   List *destinationWorkerNodesList,
 								   List *replicationSlotInfoList)
@@ -140,15 +156,17 @@ List * CreateShardSplitPubSubMetadataList(List *sourceColocatedShardIntervalList
 		forboth_ptr(splitChildShardInterval, splitChildShardIntervalList,
 					destinationWorkerNode, destinationWorkerNodesList)
 		{
-            /* Table owner is same for both parent and child shard */
-            Oid tableOwnerId = TableOwnerOid(sourceShardIntervalToCopy->relationId);
 			uint32 destinationWorkerNodeId = destinationWorkerNode->nodeId;
 
 			/* Add split child shard interval */
-            AddShardEntryInMap(tableOwnerId, destinationWorkerNodeId, splitChildShardInterval, true /*isChildShardInterval*/);
+			AddPublishableShardEntryInMap(destinationWorkerNodeId,
+										  splitChildShardInterval,
+										  true /*isChildShardInterval*/);
 
 			/* Add parent shard interval if not already added */
-            AddShardEntryInMap(tableOwnerId, destinationWorkerNodeId, sourceShardIntervalToCopy, false /*isChildShardInterval*/);
+			AddPublishableShardEntryInMap(destinationWorkerNodeId,
+										  sourceShardIntervalToCopy,
+										  false /*isChildShardInterval*/);
 		}
 	}
 
@@ -162,23 +180,31 @@ List * CreateShardSplitPubSubMetadataList(List *sourceColocatedShardIntervalList
 	{
 		uint32_t nodeId = entry->key.nodeId;
 		uint32_t tableOwnerId = entry->key.tableOwnerId;
-        ShardSplitPubSubMetadata * shardSplitPubSubMetadata = CreateShardSplitPubSubMetadata(tableOwnerId, nodeId, entry->shardSplitInfoList, replicationSlotInfoList);
+		ShardSplitPubSubMetadata *shardSplitPubSubMetadata =
+			CreateShardSplitPubSubMetadata(tableOwnerId, nodeId,
+										   entry->shardSplitInfoList,
+										   replicationSlotInfoList);
 
-        shardSplitPubSubMetadataList = lappend(shardSplitPubSubMetadataList, shardSplitPubSubMetadata);
+		shardSplitPubSubMetadataList = lappend(shardSplitPubSubMetadataList,
+											   shardSplitPubSubMetadata);
 	}
 
 	return shardSplitPubSubMetadataList;
 }
 
-static void AddShardEntryInMap(Oid tableOwnerId, uint32 nodeId, ShardInterval * shardInterval, bool isChildShardInterval)
+
+static void
+AddPublishableShardEntryInMap(uint32 targetNodeId, ShardInterval *shardInterval, bool
+							  isChildShardInterval)
 {
 	NodeShardMappingKey key;
-	key.nodeId = nodeId;
+	key.nodeId = targetNodeId;
-	key.tableOwnerId = tableOwnerId;
+	key.tableOwnerId = TableOwnerOid(shardInterval->relationId);
 
 	bool found = false;
 	NodeShardMappingEntry *nodeMappingEntry =
-		(NodeShardMappingEntry *) hash_search(ShardInfoHashMapForPublications, &key, HASH_ENTER,
+		(NodeShardMappingEntry *) hash_search(ShardInfoHashMapForPublications, &key,
+											  HASH_ENTER,
 											  &found);
 	if (!found)
 	{
@@ -188,7 +214,8 @@ static void AddShardEntryInMap(Oid tableOwnerId, uint32 nodeId, ShardInterval *
 	if (isChildShardInterval)
 	{
 		nodeMappingEntry->shardSplitInfoList =
-            lappend(nodeMappingEntry->shardSplitInfoList, (ShardInterval *) shardInterval);
+			lappend(nodeMappingEntry->shardSplitInfoList,
+					(ShardInterval *) shardInterval);
 		return;
 	}
 
@@ -205,14 +232,15 @@ static void AddShardEntryInMap(Oid tableOwnerId, uint32 nodeId, ShardInterval *
 	/* Add parent shard Interval */
 	nodeMappingEntry->shardSplitInfoList =
 		lappend(nodeMappingEntry->shardSplitInfoList, (ShardInterval *) shardInterval);
-
 }
 
 
-ShardSplitPubSubMetadata * CreateShardSplitPubSubMetadata(Oid tableOwnerId, uint32 nodeId, List * shardIntervalList, List * replicationSlotInfoList)
+ShardSplitPubSubMetadata *
+CreateShardSplitPubSubMetadata(Oid tableOwnerId, uint32 nodeId, List *shardIntervalList,
+							   List *replicationSlotInfoList)
 {
-
+	ShardSplitPubSubMetadata *shardSplitPubSubMetadata = palloc0(
-    ShardSplitPubSubMetadata * shardSplitPubSubMetadata = palloc0(sizeof(ShardSplitPubSubMetadata));
+		sizeof(ShardSplitPubSubMetadata));
 	shardSplitPubSubMetadata->shardIntervalListForSubscription = shardIntervalList;
 	shardSplitPubSubMetadata->tableOwnerId = tableOwnerId;
 
@@ -232,7 +260,8 @@ ShardSplitPubSubMetadata * CreateShardSplitPubSubMetadata(Oid tableOwnerId, uint
 }
 
 
-void LogicallyReplicateSplitShards(WorkerNode *sourceWorkerNode,
+void
+LogicallyReplicateSplitShards(WorkerNode *sourceWorkerNode,
 							  List *shardSplitPubSubMetadataList,
 							  List *sourceColocatedShardIntervalList,
 							  List *shardGroupSplitIntervalListList,
@@ -244,7 +273,8 @@ void LogicallyReplicateSplitShards(WorkerNode *sourceWorkerNode,
 
 	/* Get source node connection */
 	MultiConnection *sourceConnection =
-        GetNodeUserDatabaseConnection(connectionFlags, sourceWorkerNode->workerName, sourceWorkerNode->workerPort,
+		GetNodeUserDatabaseConnection(connectionFlags, sourceWorkerNode->workerName,
+									  sourceWorkerNode->workerPort,
 									  superUser, databaseName);
 
 	ClaimConnectionExclusively(sourceConnection);
@@ -257,7 +287,8 @@ void LogicallyReplicateSplitShards(WorkerNode *sourceWorkerNode,
 		WorkerNode *targetWorkerNode = FindNodeWithNodeId(targetWorkerNodeId, false);
 
 		MultiConnection *targetConnection =
-        GetNodeUserDatabaseConnection(connectionFlags, targetWorkerNode->workerName, targetWorkerNode->workerPort,
+			GetNodeUserDatabaseConnection(connectionFlags, targetWorkerNode->workerName,
+										  targetWorkerNode->workerPort,
 										  superUser, databaseName);
 		ClaimConnectionExclusively(targetConnection);
 
@@ -273,26 +304,36 @@ void LogicallyReplicateSplitShards(WorkerNode *sourceWorkerNode,
 								  superUser,
 								  databaseName);
 
-   WaitForShardSplitRelationSubscriptionsBecomeReady(targetNodeConnectionList, shardSplitPubSubMetadataList);
+	WaitForShardSplitRelationSubscriptionsBecomeReady(targetNodeConnectionList,
+													  shardSplitPubSubMetadataList);
 
 	XLogRecPtr sourcePosition = GetRemoteLogPosition(sourceConnection);
-   WaitForShardSplitRelationSubscriptionsToBeCaughtUp(sourcePosition, targetNodeConnectionList, shardSplitPubSubMetadataList);
+	WaitForShardSplitRelationSubscriptionsToBeCaughtUp(sourcePosition,
+													   targetNodeConnectionList,
+													   shardSplitPubSubMetadataList);
 
-   CreateAuxiliaryStructuresForShardGroup(shardGroupSplitIntervalListList, destinationWorkerNodesList);
+	CreateAuxiliaryStructuresForShardGroup(shardGroupSplitIntervalListList,
+										   destinationWorkerNodesList);
 
 	sourcePosition = GetRemoteLogPosition(sourceConnection);
-   WaitForShardSplitRelationSubscriptionsToBeCaughtUp(sourcePosition, targetNodeConnectionList, shardSplitPubSubMetadataList);
+	WaitForShardSplitRelationSubscriptionsToBeCaughtUp(sourcePosition,
+													   targetNodeConnectionList,
+													   shardSplitPubSubMetadataList);
 
 	BlockWritesToShardList(sourceColocatedShardIntervalList);
 
 	sourcePosition = GetRemoteLogPosition(sourceConnection);
-   WaitForShardSplitRelationSubscriptionsToBeCaughtUp(sourcePosition, targetNodeConnectionList, shardSplitPubSubMetadataList);
+	WaitForShardSplitRelationSubscriptionsToBeCaughtUp(sourcePosition,
+													   targetNodeConnectionList,
+													   shardSplitPubSubMetadataList);
+
 
 	/*TOOD : Create foreign key constraints and handle partitioned tables*/
 }
 
 
-void PrintShardSplitPubSubMetadata(ShardSplitPubSubMetadata * shardSplitMetadata)
+void
+PrintShardSplitPubSubMetadata(ShardSplitPubSubMetadata *shardSplitMetadata)
 {
 	printf("sameer: ShardSplitPubSbuMetadata\n");
 	ReplicationSlotInfo *replicationInfo = shardSplitMetadata->slotInfo;
@@ -305,13 +346,16 @@ void PrintShardSplitPubSubMetadata(ShardSplitPubSubMetadata * shardSplitMetadata
 		printf("%ld ", shardInterval->shardId);
 	}
 
-    printf("\nManual Username from OID at source: %s \n", GetUserNameFromId(shardSplitMetadata->tableOwnerId, false));
+	printf("\nManual Username from OID at source: %s \n", GetUserNameFromId(
-    printf("slotname:%s  targetNode:%u tableOwner:%s \n", replicationInfo->slotName, replicationInfo->targetNodeId, replicationInfo->tableOwnerName);
+			   shardSplitMetadata->tableOwnerId, false));
+	printf("slotname:%s  targetNode:%u tableOwner:%s \n", replicationInfo->slotName,
+		   replicationInfo->targetNodeId, replicationInfo->tableOwnerName);
 }
 
 
 static void
-CreateShardSplitPublications(MultiConnection *sourceConnection, List *shardSplitPubSubMetadataList)
+CreateShardSplitPublications(MultiConnection *sourceConnection,
+							 List *shardSplitPubSubMetadataList)
 {
 	ShardSplitPubSubMetadata *shardSplitPubSubMetadata = NULL;
 	foreach_ptr(shardSplitPubSubMetadata, shardSplitPubSubMetadataList)
@@ -320,12 +364,14 @@ CreateShardSplitPublications(MultiConnection *sourceConnection, List *shardSplit
 		Oid tableOwnerId = shardSplitPubSubMetadata->tableOwnerId;
 
 		CreateShardSplitPublicationForNode(sourceConnection,
-                                     shardSplitPubSubMetadata->shardIntervalListForSubscription,
+										   shardSplitPubSubMetadata->
+										   shardIntervalListForSubscription,
 										   publicationForNodeId,
 										   tableOwnerId);
 	}
 }
 
+
 static void
 CreateShardSplitSubscriptions(List *targetNodeConnectionList,
 							  List *shardSplitPubSubMetadataList,
@@ -355,7 +401,6 @@ static void
 CreateShardSplitPublicationForNode(MultiConnection *connection, List *shardList,
 								   uint32_t publicationForTargetNodeId, Oid ownerId)
 {
-
 	StringInfo createPublicationCommand = makeStringInfo();
 	bool prefixWithComma = false;
 
@@ -390,7 +435,8 @@ ShardSplitPublicationName(uint32_t nodeId, Oid ownerId)
 
 
 static void
-WaitForShardSplitRelationSubscriptionsBecomeReady(List * targetNodeConnectionList, List * shardSplitPubSubMetadataList)
+WaitForShardSplitRelationSubscriptionsBecomeReady(List *targetNodeConnectionList,
+												  List *shardSplitPubSubMetadataList)
 {
 	MultiConnection *targetConnection = NULL;
 	ShardSplitPubSubMetadata *shardSplitPubSubMetadata = NULL;
@@ -398,13 +444,18 @@ WaitForShardSplitRelationSubscriptionsBecomeReady(List * targetNodeConnectionLis
 				shardSplitPubSubMetadata, shardSplitPubSubMetadataList)
 	{
 		Bitmapset *tableOwnerIds = NULL;
-        tableOwnerIds = bms_add_member(tableOwnerIds, shardSplitPubSubMetadata->tableOwnerId);
+		tableOwnerIds = bms_add_member(tableOwnerIds,
-        WaitForRelationSubscriptionsBecomeReady(targetConnection, tableOwnerIds, SHARD_SPLIT_SUBSCRIPTION_PREFIX);
+									   shardSplitPubSubMetadata->tableOwnerId);
+		WaitForRelationSubscriptionsBecomeReady(targetConnection, tableOwnerIds,
+												SHARD_SPLIT_SUBSCRIPTION_PREFIX);
 	}
 }
 
+
 static void
-WaitForShardSplitRelationSubscriptionsToBeCaughtUp(XLogRecPtr sourcePosition, List * targetNodeConnectionList, List * shardSplitPubSubMetadataList)
+WaitForShardSplitRelationSubscriptionsToBeCaughtUp(XLogRecPtr sourcePosition,
+												   List *targetNodeConnectionList,
+												   List *shardSplitPubSubMetadataList)
 {
 	MultiConnection *targetConnection = NULL;
 	ShardSplitPubSubMetadata *shardSplitPubSubMetadata = NULL;
@@ -412,7 +463,9 @@ WaitForShardSplitRelationSubscriptionsToBeCaughtUp(XLogRecPtr sourcePosition, Li
 				shardSplitPubSubMetadata, shardSplitPubSubMetadataList)
 	{
 		Bitmapset *tableOwnerIds = NULL;
-        tableOwnerIds = bms_add_member(tableOwnerIds, shardSplitPubSubMetadata->tableOwnerId);
+		tableOwnerIds = bms_add_member(tableOwnerIds,
-        WaitForShardSubscriptionToCatchUp(targetConnection, sourcePosition, tableOwnerIds, SHARD_SPLIT_SUBSCRIPTION_PREFIX);
+									   shardSplitPubSubMetadata->tableOwnerId);
+		WaitForShardSubscriptionToCatchUp(targetConnection, sourcePosition, tableOwnerIds,
+										  SHARD_SPLIT_SUBSCRIPTION_PREFIX);
 	}
 }

src/include/distributed/multi_logical_replication.h

@@ -28,14 +28,15 @@ extern void LogicallyReplicateShards(List *shardList, char *sourceNodeName,
 
 extern XLogRecPtr GetRemoteLogPosition(MultiConnection *connection);
 
-extern void
+extern void CreateShardSubscription(MultiConnection *connection, char *sourceNodeName,
-CreateShardSubscription(MultiConnection *connection, char *sourceNodeName,
+									int sourceNodePort, char *userName,
-							 int sourceNodePort, char *userName, char *databaseName,
+									char *databaseName,
 									char *publicationName,
 									Oid ownerId);
 
 extern void WaitForRelationSubscriptionsBecomeReady(MultiConnection *targetConnection,
-													Bitmapset *tableOwnerIds, char * operationPrefix);
+													Bitmapset *tableOwnerIds,
+													char *operationPrefix);
 extern void WaitForShardSubscriptionToCatchUp(MultiConnection *targetConnection,
 											  XLogRecPtr sourcePosition,
 											  Bitmapset *tableOwnerIds,

src/include/distributed/shardsplit_logical_replication.h

@@ -44,8 +44,10 @@ HTAB * SetupHashMapForShardInfo(void);
 
 List * ParseReplicationSlotInfoFromResult(PGresult *result);
 
-extern StringInfo CreateSplitShardReplicationSetupUDF(List *sourceColocatedShardIntervalList,
+extern StringInfo CreateSplitShardReplicationSetupUDF(
-									   List *shardGroupSplitIntervalListList,
+	List *sourceColocatedShardIntervalList,
+	List *
+	shardGroupSplitIntervalListList,
 	List *destinationWorkerNodesList);
 
 extern List *  CreateShardSplitPubSubMetadataList(List *sourceColocatedShardIntervalList,

src/test/regress/expected/citus_sameer.out

@@ -113,8 +113,8 @@ SELECT nodeid AS worker_2_node FROM pg_dist_node WHERE nodeport=:worker_2_port \
 SELECT citus_split_shard_by_split_points(
 	1,
 	ARRAY['-1073741826'],
-	ARRAY[:worker_1_node, :worker_2_node],
+	ARRAY[:worker_2_node, :worker_2_node],
-	'non_blocking');
+	'force_logical');
 NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing SELECT worker_apply_shard_ddl_command (100, 'CREATE TABLE citus_split_shard_by_split_points_negative.table_to_split (id bigint NOT NULL, value character(1)) ')
@@ -165,6 +165,14 @@ NOTICE:  issuing COMMIT
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT worker_apply_shard_ddl_command (100, 'CREATE TABLE citus_split_shard_by_split_points_negative.table_to_split (id bigint NOT NULL, value character(1)) ')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT worker_apply_shard_ddl_command (100, 'ALTER TABLE citus_split_shard_by_split_points_negative.table_to_split OWNER TO postgres')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing COMMIT
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing SELECT worker_apply_shard_ddl_command (101, 'CREATE TABLE citus_split_shard_by_split_points_negative.table_to_split (id bigint NOT NULL, value character(1)) ')
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing SELECT worker_apply_shard_ddl_command (101, 'ALTER TABLE citus_split_shard_by_split_points_negative.table_to_split OWNER TO postgres')
@@ -173,14 +181,121 @@ NOTICE:  issuing COMMIT
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT worker_apply_shard_ddl_command (102, 'CREATE TABLE citus_split_shard_by_split_points_negative.table_second (id bigint NOT NULL, value character(1)) ')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT worker_apply_shard_ddl_command (102, 'ALTER TABLE citus_split_shard_by_split_points_negative.table_second OWNER TO postgres')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing COMMIT
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing SELECT worker_apply_shard_ddl_command (103, 'CREATE TABLE citus_split_shard_by_split_points_negative.table_second (id bigint NOT NULL, value character(1)) ')
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing SELECT worker_apply_shard_ddl_command (103, 'ALTER TABLE citus_split_shard_by_split_points_negative.table_second OWNER TO postgres')
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing COMMIT
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
-NOTICE:  issuing SELECT * FROM worker_split_shard_replication_setup(ARRAY[ROW(1, 100, '-2147483648', '-1073741826', 16)::citus.split_shard_info,ROW(1, 101, '-1073741825', '2147483647', 18)::citus.split_shard_info,ROW(2, 102, '-2147483648', '-1073741826', 16)::citus.split_shard_info,ROW(2, 103, '-1073741825', '2147483647', 18)::citus.split_shard_info])
+NOTICE:  issuing SELECT * FROM worker_split_shard_replication_setup(ARRAY[ROW(1, 100, '-2147483648', '-1073741826', 18)::citus.split_shard_info,ROW(1, 101, '-1073741825', '2147483647', 18)::citus.split_shard_info,ROW(2, 102, '-2147483648', '-1073741826', 18)::citus.split_shard_info,ROW(2, 103, '-1073741825', '2147483647', 18)::citus.split_shard_info])
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing CREATE PUBLICATION citus_shard_split_publication_18_10 FOR TABLE citus_split_shard_by_split_points_negative.table_to_split_100,citus_split_shard_by_split_points_negative.table_to_split_1,citus_split_shard_by_split_points_negative.table_to_split_101,citus_split_shard_by_split_points_negative.table_second_102,citus_split_shard_by_split_points_negative.table_second_2,citus_split_shard_by_split_points_negative.table_second_103
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SET LOCAL citus.enable_ddl_propagation TO OFF;
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing CREATE USER citus_shard_move_subscription_role_10 SUPERUSER IN ROLE postgres
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing COMMIT
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing CREATE SUBSCRIPTION citus_shard_split_subscription_10 CONNECTION 'host=''localhost'' port=xxxxx user=''postgres'' dbname=''regression'' connect_timeout=20' PUBLICATION citus_shard_split_publication_18_10 WITH (citus_use_authinfo=true, enabled=false)
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER SUBSCRIPTION citus_shard_split_subscription_10 OWNER TO citus_shard_move_subscription_role_10
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SET LOCAL citus.enable_ddl_propagation TO OFF;
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER ROLE citus_shard_move_subscription_role_10 NOSUPERUSER
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing COMMIT
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER SUBSCRIPTION citus_shard_split_subscription_10 ENABLE
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT count(*) FROM pg_subscription_rel, pg_stat_subscription WHERE srsubid = subid AND srsubstate != 'r' AND subname IN ('citus_shard_split_subscription_10')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT sum(pg_total_relation_size(srrelid)) FROM pg_subscription_rel, pg_stat_subscription WHERE srsubid = subid AND subname IN ('citus_shard_split_subscription_10')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT count(*) FROM pg_subscription_rel, pg_stat_subscription WHERE srsubid = subid AND srsubstate != 'r' AND subname IN ('citus_shard_split_subscription_10')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT pg_current_wal_lsn()
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT min(latest_end_lsn) FROM pg_stat_subscription WHERE subname IN ('citus_shard_split_subscription_10')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT worker_apply_shard_ddl_command (100, 'ALTER TABLE citus_split_shard_by_split_points_negative.table_to_split ADD CONSTRAINT table_to_split_pkey PRIMARY KEY (id)')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT worker_apply_shard_ddl_command (101, 'ALTER TABLE citus_split_shard_by_split_points_negative.table_to_split ADD CONSTRAINT table_to_split_pkey PRIMARY KEY (id)')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT worker_apply_shard_ddl_command (102, 'ALTER TABLE citus_split_shard_by_split_points_negative.table_second ADD CONSTRAINT table_second_pkey PRIMARY KEY (id)')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT worker_apply_shard_ddl_command (103, 'ALTER TABLE citus_split_shard_by_split_points_negative.table_second ADD CONSTRAINT table_second_pkey PRIMARY KEY (id)')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT pg_current_wal_lsn()
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT min(latest_end_lsn) FROM pg_stat_subscription WHERE subname IN ('citus_shard_split_subscription_10')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT lock_shard_metadata(7, ARRAY[1, 2])
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT lock_shard_metadata(7, ARRAY[1, 2])
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT pg_current_wal_lsn()
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT min(latest_end_lsn) FROM pg_stat_subscription WHERE subname IN ('citus_shard_split_subscription_10')
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT citus_internal_delete_shard_metadata(1);
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT citus_internal_delete_shard_metadata(1);
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing DROP TABLE IF EXISTS citus_split_shard_by_split_points_negative.table_to_split_1 CASCADE
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT citus_internal_delete_shard_metadata(2);
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing SELECT citus_internal_delete_shard_metadata(2);
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing DROP TABLE IF EXISTS citus_split_shard_by_split_points_negative.table_second_2 CASCADE
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing WITH shard_data(relationname, shardid, storagetype, shardminvalue, shardmaxvalue)  AS (VALUES ('citus_split_shard_by_split_points_negative.table_to_split'::regclass, 100, 't'::"char", '-2147483648', '-1073741826'), ('citus_split_shard_by_split_points_negative.table_to_split'::regclass, 101, 't'::"char", '-1073741825', '2147483647'), ('citus_split_shard_by_split_points_negative.table_second'::regclass, 102, 't'::"char", '-2147483648', '-1073741826'), ('citus_split_shard_by_split_points_negative.table_second'::regclass, 103, 't'::"char", '-1073741825', '2147483647')) SELECT citus_internal_add_shard_metadata(relationname, shardid, storagetype, shardminvalue, shardmaxvalue) FROM shard_data;
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing WITH shard_data(relationname, shardid, storagetype, shardminvalue, shardmaxvalue)  AS (VALUES ('citus_split_shard_by_split_points_negative.table_to_split'::regclass, 100, 't'::"char", '-2147483648', '-1073741826'), ('citus_split_shard_by_split_points_negative.table_to_split'::regclass, 101, 't'::"char", '-1073741825', '2147483647'), ('citus_split_shard_by_split_points_negative.table_second'::regclass, 102, 't'::"char", '-2147483648', '-1073741826'), ('citus_split_shard_by_split_points_negative.table_second'::regclass, 103, 't'::"char", '-1073741825', '2147483647')) SELECT citus_internal_add_shard_metadata(relationname, shardid, storagetype, shardminvalue, shardmaxvalue) FROM shard_data;
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing WITH placement_data(shardid, shardstate, shardlength, groupid, placementid)  AS (VALUES (100, 1, 0, 16, 85), (101, 1, 0, 16, 86), (102, 1, 0, 16, 87), (103, 1, 0, 16, 88)) SELECT citus_internal_add_placement_metadata(shardid, shardstate, shardlength, groupid, placementid) FROM placement_data;
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing WITH placement_data(shardid, shardstate, shardlength, groupid, placementid)  AS (VALUES (100, 1, 0, 16, 85), (101, 1, 0, 16, 86), (102, 1, 0, 16, 87), (103, 1, 0, 16, 88)) SELECT citus_internal_add_placement_metadata(shardid, shardstate, shardlength, groupid, placementid) FROM placement_data;
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing DROP TABLE IF EXISTS citus_split_shard_by_split_points_negative.table_to_split_100 CASCADE
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing DROP TABLE IF EXISTS citus_split_shard_by_split_points_negative.table_to_split_101 CASCADE
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing DROP TABLE IF EXISTS citus_split_shard_by_split_points_negative.table_second_102 CASCADE
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing DROP TABLE IF EXISTS citus_split_shard_by_split_points_negative.table_second_103 CASCADE
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing DROP TABLE IF EXISTS citus_split_shard_by_split_points_negative.table_to_split_1 CASCADE
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing DROP TABLE IF EXISTS citus_split_shard_by_split_points_negative.table_second_2 CASCADE
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing PREPARE TRANSACTION 'citus_xx_xx_xx_xx'
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing PREPARE TRANSACTION 'citus_xx_xx_xx_xx'
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing COMMIT PREPARED 'citus_xx_xx_xx_xx'
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing COMMIT PREPARED 'citus_xx_xx_xx_xx'
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+WARNING:  connection claimed exclusively at transaction commit
 WARNING:  connection claimed exclusively at transaction commit
 WARNING:  connection claimed exclusively at transaction commit
  citus_split_shard_by_split_points
@@ -196,10 +311,10 @@ SELECT * FROM show_catalog;
                    Schema                   |        Name        |  Owner
 ---------------------------------------------------------------------
  citus_split_shard_by_split_points_negative | table_second       | postgres
+ citus_split_shard_by_split_points_negative | table_second_102   | postgres
  citus_split_shard_by_split_points_negative | table_second_103   | postgres
- citus_split_shard_by_split_points_negative | table_second_2     | postgres
  citus_split_shard_by_split_points_negative | table_to_split     | postgres
- citus_split_shard_by_split_points_negative | table_to_split_1   | postgres
+ citus_split_shard_by_split_points_negative | table_to_split_100 | postgres
  citus_split_shard_by_split_points_negative | table_to_split_101 | postgres
 (6 rows)
 
@@ -209,14 +324,8 @@ SELECT * FROM show_catalog;
                    Schema                   |      Name      |  Owner
 ---------------------------------------------------------------------
  citus_split_shard_by_split_points_negative | table_second   | postgres
- citus_split_shard_by_split_points_negative | table_second_102   | postgres
- citus_split_shard_by_split_points_negative | table_second_103   | postgres
- citus_split_shard_by_split_points_negative | table_second_2     | postgres
  citus_split_shard_by_split_points_negative | table_to_split | postgres
- citus_split_shard_by_split_points_negative | table_to_split_1   | postgres
+(2 rows)
- citus_split_shard_by_split_points_negative | table_to_split_100 | postgres
- citus_split_shard_by_split_points_negative | table_to_split_101 | postgres
-(8 rows)
 
 SELECT * FROM pg_publication_tables;
  pubname | schemaname | tablename

src/test/regress/sql/citus_sameer.sql

@@ -49,14 +49,12 @@ SELECT citus_split_shard_by_split_points(
 	1,
 	ARRAY['-1073741826'],
 	ARRAY[:worker_2_node, :worker_2_node],
-	'non_blocking');
+	'force_logical');
 -- On worker2, we want child shard 2 and dummy shard 1  --
 -- on worker1, we want child shard 3 and 1 and dummy shard 2  --
 
 \c - - - :worker_2_port
 SET search_path TO citus_split_shard_by_split_points_negative;
-SELECT * FROM pg_stat_subscription;
-SELECT * FROM pg_subscription_rel;
 SELECT * FROM show_catalog;
 
 \c - - - :worker_1_port