Use RelidByRelfilenumberWithTempTables instead of RelidByRelfilenumber

PG18 and PG latest minors ignore temporary relations in
RelidByRelfilenumber (RelidByRelfilenode in PG15)
Relevant PG commit:
https://github.com/postgres/postgres/commit/86831952

We write a new function, RelidByRelfilenumberWithTempTables,
which does an extra check in case RelidByRelfilenumber returned
InvalidOid (i.e. 0).
This extra check is essentially a copy paste of previous
RelidByRelfilenumber behavior in case of a temp table.

src/backend/columnar/columnar_metadata.c

@@ -1993,7 +1993,8 @@ ColumnarNamespaceId(void)
 static uint64
 LookupStorageId(RelFileLocator relfilelocator)
 {
-	Oid relationId = RelidByRelfilenumber(RelationTablespace_compat(relfilelocator),
+	Oid relationId = RelidByRelfilenumberWithTempTables(RelationTablespace_compat(
+															relfilelocator),
 														RelationPhysicalIdentifierNumber_compat(
 															relfilelocator));
 
@@ -2123,3 +2124,85 @@ GetFirstRowNumberAttrIndexInColumnarStripe(TupleDesc tupleDesc)
 		   ? (Anum_columnar_stripe_first_row_number - 1)
 		   : tupleDesc->natts - 1;
 }
+
+
+/*
+ * Map a relation's (tablespace, relfilenumber) to a relation's oid and cache
+ * the result.
+ * PG18 and latest PG minors have excluded temporary tables from RelidByRelfilenumber,
+ * so we need to handle them here.
+ * Relevant PG commit: https://github.com/postgres/postgres/commit/dcdc95cb4
+ * This function does an extra search if relid is InvalidOid, for temp tables only.
+ * Code is mostly copy-paste from PG's RelidByRelfilenumber.
+ * Returns InvalidOid if no relation matching the criteria could be found.
+ */
+Oid
+RelidByRelfilenumberWithTempTables(Oid reltablespace, RelFileNumber relfilenumber)
+{
+	Oid relid;
+
+#if PG_VERSION_NUM >= PG_VERSION_16
+	relid = RelidByRelfilenumber(reltablespace, relfilenumber);
+#else
+	relid = RelidByRelfilenode(reltablespace, relfilenumber);
+#endif
+
+	if (relid == InvalidOid)
+	{
+		ScanKeyData skey[2];
+		HeapTuple ntp;
+
+		/* pg_class will show 0 when the value is actually MyDatabaseTableSpace */
+		if (reltablespace == MyDatabaseTableSpace)
+		{
+			reltablespace = 0;
+		}
+
+		/* check for plain relations by looking in pg_class */
+		Relation relation = table_open(RelationRelationId, AccessShareLock);
+
+		ScanKeyInit(&skey[0],
+					Anum_pg_class_reltablespace,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(reltablespace));
+		ScanKeyInit(&skey[1],
+					Anum_pg_class_relfilenode,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(relfilenumber));
+
+		SysScanDesc scandesc = systable_beginscan(relation,
+												  ClassTblspcRelfilenodeIndexId,
+												  true,
+												  NULL,
+												  2,
+												  skey);
+
+		bool found = false;
+
+		while (HeapTupleIsValid(ntp = systable_getnext(scandesc)))
+		{
+			Form_pg_class classform = (Form_pg_class) GETSTRUCT(ntp);
+
+			if (found)
+			{
+				elog(ERROR,
+					 "unexpected duplicate for tablespace %u, relfilenumber %u",
+					 reltablespace, relfilenumber);
+			}
+			found = true;
+
+			Assert(classform->reltablespace == reltablespace);
+			Assert(classform->relfilenode == relfilenumber);
+
+			if (classform->relpersistence == RELPERSISTENCE_TEMP)
+			{
+				relid = classform->oid;
+			}
+		}
+
+		systable_endscan(scandesc);
+		table_close(relation, AccessShareLock);
+	}
+
+	return relid;
+}

src/backend/columnar/columnar_writer.c

@@ -183,7 +183,7 @@ ColumnarWriteRow(ColumnarWriteState *writeState, Datum *columnValues, bool *colu
 		writeState->stripeSkipList = stripeSkipList;
 		writeState->compressionBuffer = makeStringInfo();
 
-		Oid relationId = RelidByRelfilenumber(RelationTablespace_compat(
+		Oid relationId = RelidByRelfilenumberWithTempTables(RelationTablespace_compat(
 																writeState->relfilelocator),
 															RelationPhysicalIdentifierNumber_compat(
 																writeState->relfilelocator));
@@ -404,7 +404,7 @@ FlushStripe(ColumnarWriteState *writeState)
 
 	elog(DEBUG1, "Flushing Stripe of size %d", stripeBuffers->rowCount);
 
-	Oid relationId = RelidByRelfilenumber(RelationTablespace_compat(
+	Oid relationId = RelidByRelfilenumberWithTempTables(RelationTablespace_compat(
 															writeState->relfilelocator),
 														RelationPhysicalIdentifierNumber_compat(
 															writeState->relfilelocator));

src/include/columnar/columnar_metadata.h

@@ -65,5 +65,7 @@ extern List * StripesForRelfilelocator(RelFileLocator relfilelocator);
 extern void ColumnarStorageUpdateIfNeeded(Relation rel, bool isUpgrade);
 extern List * ExtractColumnarRelOptions(List *inOptions, List **outColumnarOptions);
 extern void SetColumnarRelOptions(RangeVar *rv, List *reloptions);
+extern Oid RelidByRelfilenumberWithTempTables(Oid reltablespace, RelFileNumber
+											  relfilenumber);
 
 #endif /* COLUMNAR_METADATA_H */

src/include/pg_version_compat.h

@@ -521,7 +521,6 @@ get_guc_variables_compat(int *gucCount)
 #define RelationPhysicalIdentifierBackend_compat(a) (a->smgr_rnode.node)
 typedef RelFileNode RelFileLocator;
 typedef Oid RelFileNumber;
-#define RelidByRelfilenumber(a, b) RelidByRelfilenode(a, b)
 
 #define float_abs(a) Abs(a)