Use GetLatestSnapshot() over GetActiveSnapshot()

We already do serialization properly via `LockTableForUpdate` on
modify vs vacuum, that part is fine.

The problem is that when the waiting transaction is unblocked, it
cannot really see the catalog changes done by the others, unless
the snapshot is `GetLatestSnapshot()`

And, in VACUUM & all other places, we already do that properly:

https://github.com/Snowflake-Labs/pg_lake/blob/bfa38cef35eea16aade99f2557fab7c923a9f9c9/pg_lake_table/src/fdw/writable_table.c#L692-L696

So, we should do that for table scan as well, and this PR is
about that.

Essentially, we get to Iceberg catalog operations follow a global order defined by
`LockTableForUpdate()` within that order, each 'catalog op' sees all committed
changes from earlier `catalog ops`.

We also use the same snapshot for all tables in a scan.

Author: sfc-gh-okalaci

pg_lake_table/src/fdw/snapshot.c

@@ -52,6 +52,7 @@
 #include "utils/typcache.h"
 
 static PgLakeTableScan * CreateTableScanForRelation(Oid relationId,
+													Snapshot snapshot,
 													int uniqueRelationIdentifier,
 													List *baseRestrictInfoList,
 													bool includeChildren,
@@ -81,6 +82,13 @@ CreatePgLakeScanSnapshot(List *rteList,
 						 bool includeChildren,
 						 Oid resultRelationId)
 {
+	/*
+	 * Make sure: a) We see the concurrent changes that might have happened on
+	 * the data_files catalog b) We have a consistent view of multiple tables
+	 * if the query involves multiple tables.
+	 */
+	Snapshot	postgresSnapshot = GetLatestSnapshot();
+
 	List	   *tableScans = NIL;
 
 	ListCell   *relationCell = NULL;
@@ -116,7 +124,8 @@ CreatePgLakeScanSnapshot(List *rteList,
 		}
 
 		PgLakeTableScan *tableScan =
-			CreateTableScanForRelation(relationId, uniqueRelationIdentifier,
+			CreateTableScanForRelation(relationId, postgresSnapshot,
+									   uniqueRelationIdentifier,
 									   baseRestrictInfoList,
 									   includeChildren && rte->inh,
 									   isResultRelation);
@@ -174,7 +183,7 @@ GetBaseRestrictInfoForRelation(List *relationRestrictionsList, int uniqueRelatio
  * CreateTableScanForRelation creates a table scan for the given relation.
  */
 static PgLakeTableScan *
-CreateTableScanForRelation(Oid relationId, int uniqueRelationIdentifier, List *baseRestrictInfoList,
+CreateTableScanForRelation(Oid relationId, Snapshot snapshot, int uniqueRelationIdentifier, List *baseRestrictInfoList,
 						   bool includeChildren, bool isResultRelation)
 {
 	List	   *fileScans = NIL;
@@ -187,7 +196,6 @@ CreateTableScanForRelation(Oid relationId, int uniqueRelationIdentifier, List *b
 		 * We'll calculate the deletion files based on the pruned data files
 		 * using the same snapshot.
 		 */
-		Snapshot	snapshot = GetActiveSnapshot();
 		bool		dataOnly = true;
 		bool		newFilesOnly = false;
 		List	   *dataFiles =
@@ -317,7 +325,8 @@ CreateTableScanForRelation(Oid relationId, int uniqueRelationIdentifier, List *b
 			 * hierarchy?
 			 */
 			PgLakeTableScan *childScan =
-				CreateTableScanForRelation(childId, childRelationIdentifier,
+				CreateTableScanForRelation(childId, snapshot,
+										   childRelationIdentifier,
 										   baseRestrictInfoList,
 										   includeChildren,
 										   isResultRelation);