Function ShardCollectionCore

pg_documentdb/src/sharding/sharding.c:1384–1825

static void
ShardCollectionCore(ShardCollectionArgs *args)
{
	Datum databaseDatum = CStringGetTextDatum(args->databaseName);
	Datum collectionDatum = CStringGetTextDatum(args->collectionName);

	/* Allow for checking if reference metadata tables are correctly handled */
	EnsureMetadataTableReplicated("collections");

	MongoCollection *collection = GetMongoCollectionByNameDatum(
		databaseDatum, collectionDatum, AccessShareLock);

	if (collection == NULL)
	{
		if (args->shardingMode != ShardCollectionMode_Shard)
		{
			ereport(ERROR, (errcode(ERRCODE_DOCUMENTDB_NAMESPACENOTSHARDED),
							errmsg("Collection %s.%s is not sharded",
								   args->databaseName, args->collectionName),
							errdetail_log(
								"Unable to perform re/unshard operation on a collection that is missing: %s.%s",
								args->databaseName, args->collectionName)));
		}

		CreateCollection(databaseDatum, collectionDatum);
		collection = GetMongoCollectionByNameDatum(
			databaseDatum, collectionDatum, AccessShareLock);

		if (collection == NULL)
		{
			ereport(ERROR, (errcode(ERRCODE_DOCUMENTDB_INTERNALERROR),
							errmsg("Failed to create collection %s.%s for sharding",
								   args->databaseName, args->collectionName),
							errdetail_log(
								"Collection was not found after implicit create for sharding for %s.%s",
								args->databaseName, args->collectionName)));
		}
	}

	if (collection->shardKey == NULL &&
		args->shardingMode != ShardCollectionMode_Shard)
	{
		ereport(ERROR, (errcode(ERRCODE_DOCUMENTDB_NAMESPACENOTSHARDED),
						errmsg("Collection %s.%s is not sharded",
							   args->databaseName, args->collectionName),
						errdetail_log(
							"Can not re/unshard collection that is not sharded: %s.%s",
							args->databaseName, args->collectionName)));
	}

	if (!args->forceRedistribution &&
		collection->shardKey != NULL && PgbsonEquals(collection->shardKey,
													 args->shardKeyDefinition))
	{
		ereport(NOTICE, (errmsg(
							 "Skipping Sharding for collection %s.%s as the same options were passed in.",
							 args->databaseName, args->collectionName)));
		return;