Function InsertOrReplaceDocument

pg_documentdb/src/commands/insert.c:1347–1439

* Inserts a document with the given shardKeyValue and object_id, and on conflict * with the _id index, reapplies the update on the conflicting document (Similar to upsert behavior) */

 * with the _id index, reapplies the update on the conflicting document (Similar to upsert behavior)
 */
bool
InsertOrReplaceDocument(uint64 collectionId, const char *shardTableName, int64
						shardKeyValue,
						pgbson *objectId, pgbson *document,
						const bson_value_t *updateSpecValue)
{
	StringInfoData query;
	const int argCount = 4;
	Oid argTypes[4];
	Datum argValues[4];
	int spiStatus PG_USED_FOR_ASSERTS_ONLY = 0;
	uint64 planId;

	SPI_connect();

	pgbson *updateSpecDoc = BsonValueToDocumentPgbson(updateSpecValue);

	initStringInfo(&query);
	appendStringInfo(&query, "INSERT INTO %s.", ApiDataSchemaName);

	if (shardTableName != NULL && shardTableName[0] != '\0')
	{
		appendStringInfoString(&query, shardTableName);
	}
	else
	{
		appendStringInfo(&query, "documents_" UINT64_FORMAT, collectionId);
	}

	appendStringInfo(&query, " (shard_key_value, object_id, document) "
							 " VALUES ($1, %s.bson_from_bytea($2), "
							 "%s.bson_from_bytea($3))",
					 CoreSchemaName, CoreSchemaName);

	planId = QUERY_ID_INSERT_OR_REPLACE;

	const char *additionalArgs = "";
	if (IsClusterVersionAtleast(DocDB_V0, 111, 0))
	{
		additionalArgs = ", ctid, tableoid";
	}

	if (shardTableName != NULL && shardTableName[0] != '\0')
	{
		/* Direct shard - we need to extract tableId_shardId as a suffix */
		/* Prefix length is the length of documents_ */
		const int prefixLength = 10;
		const char *shardSuffix = shardTableName + prefixLength;
		appendStringInfo(&query, " ON CONFLICT ON CONSTRAINT collection_pk_%s"
								 " DO UPDATE SET document ="
								 " COALESCE(%s.update_bson_document("
								 " %s.documents_%s.document, %s.bson_from_bytea($4), '{}'::%s.bson, NULL::%s.bson, NULL::%s.bson, NULL::TEXT%s),"
								 " %s.documents_%s.document)",
						 shardSuffix, ApiInternalSchemaNameV2, ApiDataSchemaName,
						 shardSuffix, CoreSchemaName, CoreSchemaName, CoreSchemaName,
						 CoreSchemaName, additionalArgs,
						 ApiDataSchemaName, shardSuffix);
	}