Merged PR 1539952: allChangesForCluster option for change streams

### Does this PR have any customer impact?
Yes. This PR provides the option of "allChangesForCluster" to be able to track all the changes happening in the cluster with a change stream cursor.

### Type (Feature, Refactoring, Bugfix, DevOps, Testing, Perf, etc)
Feature

### Does it involve schema level changes? (Table, Column, Index, UDF, etc level changes)
No

### Are you introducing any new config? If yes, do you have tests with and without them being set?
No

### Release Note (Refer [Template](../docs/release_notes/pgmongo_template.md))

### Description

first version of allChangesForCluster implementation

----
#### AI description  (iteration 1)
#### PR Classification
New feature

#### PR Summary
Implemented the first version of the `allChangesForCluster` feature in the change stream.
- `change_stream_wal_utils.c`: Added logic to track changes for the entire cluster and refactored relation ID retrieval.
- `change_stream_aggregation_stage.c`: Updated context initialization to handle `allChangesForCluster` and added error handling for missing collections.
- `change_stream_decoding_utils.c`: Enhanced event building to support cluster-wide change tracking and added a function to get collection by relation ID.
- `change_stream.h`: Extended `ChangeStreamEvent` and `ChangeStreamSpec` structures to include cluster-wide tracking.
- Updated tests in `pgmongo_change_stream_crud_tests.sql` and `pgmongo_change_stream_crud_tests.out` to reflect the new feature.

Author: rajeshkt78

pg_documentdb/include/metadata/collection.h

@@ -207,6 +207,9 @@ MongoCollection * CopyMongoCollection(const MongoCollection *collection);
 /* get Mongo collection metadata by collection id */
 MongoCollection * GetMongoCollectionByColId(uint64 collectionId, LOCKMODE lockMode);
 
+/* get Mongo collection metadata by realtion ID of a collection's shard */
+MongoCollection * GetMongoCollectionByRelationShardId(Oid relationId);
+
 /* get OID of Mongo documents table by collection id */
 Oid GetRelationIdForCollectionId(uint64 collectionId, LOCKMODE lockMode);
 

pg_documentdb/src/metadata/collection.c

@@ -128,6 +128,7 @@ static MongoCollection * GetMongoCollectionByNameDatumCore(Datum databaseNameDat
 														   Datum collectionNameDatum,
 														   LOCKMODE lockMode);
 static Datum GetCollectionOrViewCore(PG_FUNCTION_ARGS, bool allowViews);
+static uint64 GetCollectionIdFromShardName(const char *shardName);
 
 /*
  * CollectionCacheIsValid determines whether the collections hashes are
@@ -416,6 +417,50 @@ GetMongoCollectionByColId(uint64 collectionId, LOCKMODE lockMode)
 }
 
 
+/*
+ * GetMongoCollectionByRelationShardId gets the MongoCollection metadata by
+ * relation Id of one of the collection's shard tables. This is used by
+ * change stream aggregation to lookup a collection for a given change's
+ * relation id parsed from the WAL records. It finds the collection Id
+ * from the relation ID using the naming convention of the shard table
+ * name format documents_<collection ID>_<shard ID> and just calls the
+ * GetMongoCollectionByColId to lookup the collection's meta data from
+ * the hash table.
+ */
+MongoCollection *
+GetMongoCollectionByRelationShardId(Oid relationId)
+{
+	/* Get the relation's name using the relationId. */
+	const char *shardName = get_rel_name(relationId);
+
+	/* If the shardName lookup failed, return false. */
+	if (shardName == NULL)
+	{
+		return NULL;
+	}
+
+	/* Get the collection ID from the shard table name. */
+	uint64 collectionId = GetCollectionIdFromShardName(shardName);
+
+	/*
+	 * If the collection Id can't be parsed from relation's name,
+	 * then this is not a valid shard of a collection.
+	 */
+	if (collectionId == 0)
+	{
+		return NULL;
+	}
+
+	/*
+	 * Look up the collection using it's ID and return it. we can directly
+	 * lookup from the RelationIdToCollectionHash here, but this collection may
+	 * have been evicted from cache, so an invalidation may be needed. So
+	 * we call the GetMongoCollectionByColId which handles this already.
+	 */
+	return GetMongoCollectionByColId(collectionId, NoLock);
+}
+
+
 /*
  * GetMongoCollectionOrViewByNameDatum returns collection metadata by database and
  * collection name or NULL if the collection does not exist. Also returns views
@@ -1957,3 +2002,23 @@ UpdateMongoCollectionUsingIds(MongoCollection *mongoCollection, uint64 collectio
 	snprintf(mongoCollection->tableName, NAMEDATALEN, DOCUMENT_DATA_TABLE_NAME_FORMAT,
 			 collectionId);
 }
+
+
+static uint64
+GetCollectionIdFromShardName(const char *shardName)
+{
+	uint64 collectionId = 0, shardId = 0;
+
+	/* Use sscanf to parse the string */
+	int scanned = sscanf(shardName, "documents_%lu_%lu", &collectionId, &shardId);
+
+	/* Check if sscanf successfully scanned 2 numbers and the prefix is correct */
+	if (scanned == 2 && strncmp(shardName, "documents_", 10) == 0)
+	{
+		return collectionId;
+	}
+	else
+	{
+		return -1;
+	}
+}