Replace per-column TupleDesc array with HTAB cache in CopyToStateData

Previously the TupleDesc cache was a flat TupleDesc* array indexed by
attnum, pre-populated only for top-level composite columns.  Replace it
with an HTAB (Oid -> TupleDescCacheEntry) that:

- Is allocated in StartCopyTo and destroyed in EndCopyTo.
- Is pre-populated with the Oids of all top-level composite columns
  (same as before).
- Is passed recursively through PGDuckSerialize / StructOutForPGDuck /
  ArrayOutForPGDuck so that nested composite types discovered during
  serialization are looked up once and inserted into the cache on first
  encounter, avoiding repeated lookup_rowtype_tupdesc calls for every
  row.

TupleDescCacheEntry (typid key + tupdesc value) is defined in
serialize.h alongside the updated PGDuckSerialize signature.  All
existing NULL-passing call sites (rewrite_query, map_conversion, etc.)
are unaffected.

Signed-off-by: David Christensen <david.christensen@snowflake.com>

Author: sfc-gh-dachristensen

pg_lake_engine/include/pg_lake/pgduck/array_conversion.h

@@ -19,9 +19,10 @@
 
 #include "access/tupdesc.h"
 #include "utils/array.h"
+#include "utils/hsearch.h"
 #include "pg_lake/copy/copy_format.h"
 
 #define ARRAY_OUT_OID (751)
 
 extern PGDLLEXPORT char *ArrayOutForPGDuck(ArrayType *array, CopyDataFormat format,
-										   TupleDesc cachedElemTupleDesc);
+										   HTAB *tupdesc_cache);

pg_lake_engine/include/pg_lake/pgduck/serialize.h

@@ -23,11 +23,22 @@
 #include "access/tupdesc.h"
 #include "pg_lake/parquet/field.h"
 #include "pg_lake/pgduck/type.h"
+#include "utils/hsearch.h"
 
 #define BYTEA_OUT_OID 31
 
+/*
+ * Entry in the Oid -> TupleDesc hash table used to cache composite-type
+ * descriptors across rows.  typid is the hash key and must be first.
+ */
+typedef struct TupleDescCacheEntry
+{
+	Oid			typid;
+	TupleDesc	tupdesc;
+} TupleDescCacheEntry;
+
 extern PGDLLEXPORT char *PGDuckSerialize(FmgrInfo *flinfo, Oid typeOid, Datum value,
-										 CopyDataFormat format, TupleDesc cachedTupleDesc);
+										 CopyDataFormat format, HTAB *tupdesc_cache);
 extern PGDLLEXPORT char *PGDuckOnlySerialize(Oid typeOid, Datum value);
 extern PGDLLEXPORT bool IsPGDuckSerializeRequired(PGType postgresType);
 extern PGDLLEXPORT char *IntervalOutForPGDuck(Datum value);

pg_lake_engine/include/pg_lake/pgduck/struct_conversion.h

@@ -19,8 +19,9 @@
 
 #include "access/tupdesc.h"
 #include "pg_lake/copy/copy_format.h"
+#include "utils/hsearch.h"
 
 #define RECORD_OUT_OID 2291
 
 extern PGDLLEXPORT char *StructOutForPGDuck(Datum myStruct, CopyDataFormat format,
-											TupleDesc cachedTupleDesc);
+											HTAB *tupdesc_cache);

pg_lake_engine/src/csv/csv_writer.c

@@ -38,6 +38,7 @@
 #include "pg_lake/extensions/postgis.h"
 #include "pg_lake/pgduck/numeric.h"
 #include "pg_lake/pgduck/serialize.h"
+#include "pg_lake/pgduck/struct_conversion.h"
 #include "pg_lake/util/numeric.h"
 #include "executor/executor.h"
 #include "mb/pg_wchar.h"
@@ -106,11 +107,12 @@ typedef struct CopyToStateData
 	FmgrInfo   *out_functions;	/* lookup info for output functions */
 
 	/*
-	 * Per-column TupleDesc cache for composite-type columns.  NULL means the
-	 * column is not composite (or is an anonymous RECORD column).  A non-NULL
-	 * value is a pinned TupleDesc looked up once during StartCopyTo.
+	 * Oid -> TupleDesc hash table for composite-type columns.  Pre-populated
+	 * during StartCopyTo with the top-level composite column types; lazily
+	 * extended with nested composite types encountered during serialization.
+	 * Each entry holds a pinned TupleDesc released in EndCopy.
 	 */
-	TupleDesc  *composite_tupdescs;
+	HTAB	   *tupdesc_cache;
 	MemoryContext rowcontext;	/* per-row evaluation context */
 	uint64		bytes_processed;	/* number of bytes processed so far */
 
@@ -297,18 +299,15 @@ EndCopy(CopyToState cstate)
 		}
 	}
 
-	/* Release pinned TupleDescs from the composite type cache */
+	/* Release all pinned TupleDescs from the cache */
 	{
-		ListCell   *cur;
+		HASH_SEQ_STATUS seq;
+		TupleDescCacheEntry *entry;
 
-		foreach(cur, cstate->attnumlist)
-		{
-			int			attnum = lfirst_int(cur);
-			TupleDesc	cached = cstate->composite_tupdescs[attnum - 1];
-
-			if (cached != NULL)
-				ReleaseTupleDesc(cached);
-		}
+		hash_seq_init(&seq, cstate->tupdesc_cache);
+		while ((entry = (TupleDescCacheEntry *) hash_seq_search(&seq)) != NULL)
+			ReleaseTupleDesc(entry->tupdesc);
+		hash_destroy(cstate->tupdesc_cache);
 	}
 
 	MemoryContextDelete(cstate->rowcontext);
@@ -477,7 +476,19 @@ StartCopyTo(CopyToState cstate, TupleDesc tupDesc)
 
 	/* Get info about the columns we need to process. */
 	cstate->out_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo));
-	cstate->composite_tupdescs = (TupleDesc *) palloc0(num_phys_attrs * sizeof(TupleDesc));
+
+	/* Create the Oid -> TupleDesc cache for composite-type columns. */
+	{
+		HASHCTL		hctl;
+
+		memset(&hctl, 0, sizeof(hctl));
+		hctl.keysize = sizeof(Oid);
+		hctl.entrysize = sizeof(TupleDescCacheEntry);
+		hctl.hcxt = cstate->copycontext;
+		cstate->tupdesc_cache = hash_create("TupleDesc cache", 16, &hctl,
+											HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+	}
+
 	foreach(cur, cstate->attnumlist)
 	{
 		int			attnum = lfirst_int(cur);
@@ -496,9 +507,9 @@ StartCopyTo(CopyToState cstate, TupleDesc tupDesc)
 		fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
 
 		/*
-		 * Pre-populate the TupleDesc cache for composite columns and
-		 * arrays of composite elements.  Anonymous RECORD types are skipped
-		 * because their structure is not known until row data is inspected.
+		 * Pre-populate the TupleDesc cache for composite columns and arrays
+		 * of composite elements.  Anonymous RECORD types are skipped because
+		 * their structure is not known until row data is inspected.
 		 */
 		{
 			Oid			elemTypeOid = get_element_type(attr->atttypid);
@@ -507,8 +518,15 @@ StartCopyTo(CopyToState cstate, TupleDesc tupDesc)
 
 			if (get_typtype(baseTypeOid) == TYPTYPE_COMPOSITE &&
 				baseTypeOid != RECORDOID)
-				cstate->composite_tupdescs[attnum - 1] =
-					lookup_rowtype_tupdesc(baseTypeOid, baseTypmod);
+			{
+				bool		found;
+				TupleDescCacheEntry *entry = (TupleDescCacheEntry *)
+					hash_search(cstate->tupdesc_cache, &baseTypeOid,
+								HASH_ENTER, &found);
+
+				if (!found)
+					entry->tupdesc = lookup_rowtype_tupdesc(baseTypeOid, baseTypmod);
+			}
 		}
 	}
 
@@ -812,29 +830,11 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 
 				if (ShouldUseDuckSerialization(cstate->targetFormat, MakePGType(attr->atttypid, attr->atttypmod)))
 				{
-					/*
-					 * Since we are at the top-level when emitting an
-					 * attribute in CopyOneRowTo(), we are not inside a
-					 * composite type.
-					 *
-					 * For composite-type columns, cache the TupleDesc so
-					 * we avoid repeating the expensive lookup_rowtype_tupdesc
-					 * call on every row.
-					 */
-					/*
-					 * Pass the pre-populated TupleDesc (non-NULL for
-					 * composite columns and arrays of composite elements)
-					 * so PGDuckSerialize can skip the per-row
-					 * lookup_rowtype_tupdesc call.
-					 */
-					TupleDesc	cachedTupleDesc =
-						cstate->composite_tupdescs[attnum - 1];
-
 					string = PGDuckSerialize(&out_functions[attnum - 1],
 											 attr->atttypid,
 											 value,
 											 cstate->targetFormat,
-											 cachedTupleDesc);
+											 cstate->tupdesc_cache);
 				}
 				else
 					string = OutputFunctionCall(&out_functions[attnum - 1],

pg_lake_engine/src/pgduck/array_conversion.c

@@ -36,7 +36,7 @@
  */
 char *
 ArrayOutForPGDuck(ArrayType *array, CopyDataFormat format,
-				  TupleDesc cachedElemTupleDesc)
+				  HTAB *tupdesc_cache)
 {
 	Oid			element_type = ARR_ELEMTYPE(array);
 	char	   *p,
@@ -115,7 +115,7 @@ ArrayOutForPGDuck(ArrayType *array, CopyDataFormat format,
 		else
 		{
 			values[i] = PGDuckSerialize(&my_extra->proc, element_type, itemvalue,
-										format, cachedElemTupleDesc);
+										format, tupdesc_cache);
 
 			/* count data plus backslashes; detect chars needing quotes */
 			needquote = !IsSerializedAsContainer(element_type, format);

pg_lake_engine/src/pgduck/serialize.c

@@ -125,19 +125,19 @@ ConvertISOYearToBCIfNeeded(const char *dateTimestampString)
  */
 char *
 PGDuckSerialize(FmgrInfo *flinfo, Oid columnType, Datum value,
-				CopyDataFormat format, TupleDesc cachedTupleDesc)
+				CopyDataFormat format, HTAB *tupdesc_cache)
 {
 	if (flinfo->fn_oid == ARRAY_OUT_OID)
 	{
 		/* maps are a type of array */
 		if (IsMapTypeOid(columnType))
 			return MapOutForPGDuck(value, format);
 
-		return ArrayOutForPGDuck(DatumGetArrayTypeP(value), format, cachedTupleDesc);
+		return ArrayOutForPGDuck(DatumGetArrayTypeP(value), format, tupdesc_cache);
 	}
 
 	if (flinfo->fn_oid == RECORD_OUT_OID)
-		return StructOutForPGDuck(value, format, cachedTupleDesc);
+		return StructOutForPGDuck(value, format, tupdesc_cache);
 
 	/*
 	 * For Iceberg, intervals are serialized as struct(months, days,

pg_lake_engine/src/pgduck/struct_conversion.c

@@ -20,10 +20,11 @@
 
 #include "access/htup_details.h"
 #include "utils/builtins.h"
-#include "pg_lake/pgduck/serialize.h"
-#include "pg_lake/pgduck/struct_conversion.h"
+#include "utils/hsearch.h"
 #include "utils/lsyscache.h"
 #include "utils/typcache.h"
+#include "pg_lake/pgduck/serialize.h"
+#include "pg_lake/pgduck/struct_conversion.h"
 
 /*
  * structure to cache metadata needed for record I/O
@@ -52,12 +53,13 @@ typedef struct RecordIOData
  */
 
 char *
-StructOutForPGDuck(Datum myStruct, CopyDataFormat format, TupleDesc cachedTupleDesc)
+StructOutForPGDuck(Datum myStruct, CopyDataFormat format, HTAB *tupdesc_cache)
 {
 	HeapTupleHeader rec = DatumGetHeapTupleHeader(myStruct);
 	Oid			tupType;
 	int32		tupTypmod;
 	TupleDesc	tupdesc;
+	bool		tupdesc_from_cache = false;
 	HeapTupleData tuple;
 	RecordIOData *my_extra;
 	bool		needComma = false;
@@ -72,8 +74,28 @@ StructOutForPGDuck(Datum myStruct, CopyDataFormat format, TupleDesc cachedTupleD
 	/* Extract type info from the tuple itself */
 	tupType = HeapTupleHeaderGetTypeId(rec);
 	tupTypmod = HeapTupleHeaderGetTypMod(rec);
-	if (cachedTupleDesc != NULL)
-		tupdesc = cachedTupleDesc;
+
+	if (tupdesc_cache != NULL && tupType != RECORDOID)
+	{
+		bool		found;
+		TupleDescCacheEntry *entry = (TupleDescCacheEntry *)
+			hash_search(tupdesc_cache, &tupType, HASH_FIND, &found);
+
+		if (found)
+		{
+			tupdesc = entry->tupdesc;
+			tupdesc_from_cache = true;
+		}
+		else
+		{
+			/* New nested composite type; fetch, pin, and add to cache */
+			tupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
+			entry = (TupleDescCacheEntry *)
+				hash_search(tupdesc_cache, &tupType, HASH_ENTER, &found);
+			entry->tupdesc = tupdesc;
+			tupdesc_from_cache = true;
+		}
+	}
 	else
 		tupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
 	ncolumns = tupdesc->natts;
@@ -155,7 +177,7 @@ StructOutForPGDuck(Datum myStruct, CopyDataFormat format, TupleDesc cachedTupleD
 		}
 
 		attr = values[i];
-		value = PGDuckSerialize(&column_info->proc, column_type, attr, format, NULL);
+		value = PGDuckSerialize(&column_info->proc, column_type, attr, format, tupdesc_cache);
 
 		/* Detect whether we need double quotes for this value */
 		bool		needQuotes = !IsContainerType(column_type);
@@ -180,8 +202,8 @@ StructOutForPGDuck(Datum myStruct, CopyDataFormat format, TupleDesc cachedTupleD
 	pfree(values);
 	pfree(nulls);
 
-	/* Only release if we looked it up ourselves */
-	if (cachedTupleDesc == NULL)
+	/* Only release if we looked it up outside the cache */
+	if (!tupdesc_from_cache)
 		ReleaseTupleDesc(tupdesc);
 
 	return buf.data;