Function MergeWorkerBsons

pg_documentdb/src/commands/current_op.c:348–403

* Logic that builds the currentOp responses (only runs on query coordinator) */

 * Logic that builds the currentOp responses (only runs on query coordinator)
 */
static void
MergeWorkerBsons(List *workerBsons, TupleDesc descriptor, Tuplestorestate *tupleStore)
{
	ListCell *workerCell;
	foreach(workerCell, workerBsons)
	{
		pgbson *workerBson = lfirst(workerCell);
		bson_iter_t workerIter;
		PgbsonInitIterator(workerBson, &workerIter);

		int errorCode = 0;
		const char *errorMessage = NULL;
		while (bson_iter_next(&workerIter))
		{
			const char *key = bson_iter_key(&workerIter);
			if (strcmp(key, ErrCodeKey) == 0)
			{
				errorCode = BsonValueAsInt32(bson_iter_value(&workerIter));
			}
			else if (strcmp(key, ErrMsgKey) == 0)
			{
				const char *string = bson_iter_utf8(&workerIter, NULL);
				errorMessage = pstrdup(string);
			}
			else if (strcmp(key, "activities") == 0)
			{
				bson_iter_t activityIter;
				if (bson_iter_recurse(&workerIter, &activityIter))
				{
					while (bson_iter_next(&activityIter))
					{
						pgbson *docBson = PgbsonInitFromDocumentBsonValue(bson_iter_value(
																			  &
																			  activityIter));

						Datum tupleValue[1] = { PointerGetDatum(docBson) };
						bool nulls[1] = { false };
						tuplestore_putvalues(tupleStore, descriptor, tupleValue, nulls);
					}
				}
			}
			else
			{
				ereport(ERROR, (errmsg("unknown field received from currentOp worker %s",
									   key)));
			}
		}

		if (errorMessage != NULL)
		{
			errorCode = errorCode == 0 ? ERRCODE_DOCUMENTDB_INTERNALERROR : errorCode;
			ereport(ERROR, (errcode(errorCode), errmsg("Error running currentOp: %s",
													   errorMessage)));
		}
	}
}