Use SPI parameter binding for composite-type catalog lookup

GetOrCreatePGStructType was psprintf-ing composite field names directly
into a '{...}'::name[] literal.  Hostile shapes (commas, colons, quotes,
spaces, backslashes, parens in field names) produced malformed array
literals and a confusing SPI error on auto-detect.

Switch to SPI_execute_with_args with the names and oids passed as
properly constructed name[]/oid[] arrays.  Adds a regression test
covering an auto-detect path through CREATE FOREIGN TABLE () against a
parquet file whose struct fields contain each hostile character class.

Addresses review feedback on PR #297.

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

Author: sfc-gh-dachristensen

pg_lake_engine/src/pgduck/parse_struct.c

@@ -40,6 +40,7 @@
 #include "commands/typecmds.h"
 #include "executor/spi.h"
 #include "nodes/makefuncs.h"
+#include "utils/array.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
@@ -734,51 +735,58 @@ FindOrCreatePGCompositeType(CompositeType * type)
 	 */
 	do
 	{
-		StringInfo	oidArray = makeStringInfo();
-		StringInfo	nameArray = makeStringInfo();
+		int			ncols = list_length(type->cols);
+		Datum	   *nameDatums = palloc(sizeof(Datum) * ncols);
+		Datum	   *oidDatums = palloc(sizeof(Datum) * ncols);
 		ListCell   *lc;
+		int			i = 0;
 
 		foreach(lc, type->cols)
 		{
 			CompositeCol *col = (CompositeCol *) lfirst(lc);
 
 			Assert(col->colType);
 
-			/* if we are the first time through, skip this */
-			if (oidArray->len)
-			{
-				appendStringInfoChar(oidArray, ',');
-				appendStringInfoChar(nameArray, ',');
-			}
-
-			appendStringInfo(oidArray, "%d", col->colType);
-			appendStringInfo(nameArray, "%s", col->colName);
+			/*
+			 * Build the arrays as proper Datums rather than string-formatting
+			 * into a name[] literal: field names may contain commas, quotes,
+			 * backslashes, or whitespace that would break array literal
+			 * parsing.
+			 */
+			nameDatums[i] = DirectFunctionCall1(namein,
+												CStringGetDatum(col->colName));
+			oidDatums[i] = ObjectIdGetDatum(col->colType);
+			i++;
 		}
 
-		char	   *findTypeQuery = psprintf(
-											 "SELECT oid, typname, typarray FROM pg_type JOIN "
-											 "(SELECT attrelid,"
-											 "    array_agg(atttypid ORDER BY attnum) AS types,"
-											 "    array_agg(attname ORDER BY attnum) AS names "
-											 "    FROM pg_attribute "
-											 "    WHERE attnum > 0 AND NOT attisdropped "
-											 "    GROUP BY attrelid) atts "
-											 "ON pg_type.typrelid = atts.attrelid AND "
-											 "    atts.names = '{%s}'::name[] and atts.types='{%s}'::oid[] "
-											 "WHERE typnamespace :: regnamespace in ('pg_catalog','"
-											 STRUCT_TYPES_SCHEMA "') LIMIT 1",
-											 nameArray->data,
-											 oidArray->data);
-
-		elog(DEBUG1, "checking for matching type oid via query: %s", findTypeQuery);
+		ArrayType  *nameArr = construct_array_builtin(nameDatums, ncols, NAMEOID);
+		ArrayType  *oidArr = construct_array_builtin(oidDatums, ncols, OIDOID);
+
+		const char *findTypeQuery =
+			"SELECT oid, typname, typarray FROM pg_type JOIN "
+			"(SELECT attrelid,"
+			"    array_agg(atttypid ORDER BY attnum) AS types,"
+			"    array_agg(attname ORDER BY attnum) AS names "
+			"    FROM pg_attribute "
+			"    WHERE attnum > 0 AND NOT attisdropped "
+			"    GROUP BY attrelid) atts "
+			"ON pg_type.typrelid = atts.attrelid AND "
+			"    atts.names = $1 AND atts.types = $2 "
+			"WHERE typnamespace :: regnamespace in "
+			"    ('pg_catalog','" STRUCT_TYPES_SCHEMA "') LIMIT 1";
+
+		Oid			argTypes[2] = {NAMEARRAYOID, OIDARRAYOID};
+		Datum		argValues[2] = {PointerGetDatum(nameArr),
+			PointerGetDatum(oidArr)};
 
 		Oid			foundOid = InvalidOid;
 		Oid			foundArrayOid = InvalidOid;
 		char	   *foundName = NULL;
 
 		SPI_START();
 
-		int			ret = SPI_execute(findTypeQuery, false, 1);
+		int			ret = SPI_execute_with_args(findTypeQuery, 2, argTypes,
+												argValues, NULL, false, 1);
 
 		if (ret == SPI_OK_SELECT && SPI_processed >= 1)
 		{

pg_lake_table/tests/pytests/test_duckdb_reserved_keywords.py

@@ -497,6 +497,91 @@ def test_each_reserved_keyword_csv(pg_conn, s3, extension, keyword):
 # ---------------------------------------------------------------------------
 
 
+def test_autodetect_struct_field_names_with_hostile_characters(pg_conn, s3, extension):
+    """
+    Auto-detect (CREATE FOREIGN TABLE () ... OPTIONS (path ...)) against a
+    parquet file whose struct field names contain characters that would break
+    a raw-formatted name[] array literal: quotes, commas, colons, spaces,
+    parens, backslashes.
+
+    Regression test for review feedback on PR #297 — after fixing
+    ParseDuckDBFieldName to handle SQL-standard "" doubling, the next layer
+    (GetOrCreatePGStructType's catalog-lookup SPI query) was still formatting
+    field names directly into a '{...}'::name[] literal, producing
+    "malformed array literal" for any hostile shape.
+    """
+    schema = "test_duckdb_kw_autodetect_hostile"
+    url = f"s3://{TEST_BUCKET}/{schema}/data.parquet"
+
+    run_command(f"CREATE SCHEMA {schema}", pg_conn)
+    pg_conn.commit()
+
+    run_command(
+        f"""
+        CREATE TYPE {schema}.edge_t AS (
+            plain          int,
+            "with space"   int,
+            U&"has\\0022one"   int,
+            "comma,here"   int,
+            "colon:here"   int,
+            "with(parens)" int,
+            U&"back\\005Cslash" int
+        )
+        """,
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    run_command(
+        f"""
+        CREATE TABLE {schema}.src (id int, e {schema}.edge_t)
+        USING iceberg
+        WITH (location = 's3://{TEST_BUCKET}/{schema}_src/')
+        """,
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    run_command(
+        f"INSERT INTO {schema}.src VALUES (1, ROW(1, 2, 3, 4, 5, 6, 7))",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    run_command(
+        f"COPY (SELECT * FROM {schema}.src) TO '{url}' WITH (format 'parquet')",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    # Drop the catalog-side composite type so the next CREATE FOREIGN TABLE is
+    # forced down the auto-detect path through GetOrCreatePGStructType.
+    run_command(f"DROP TABLE {schema}.src", pg_conn)
+    run_command(f"DROP TYPE {schema}.edge_t", pg_conn)
+    pg_conn.commit()
+
+    run_command(
+        f"""
+        CREATE FOREIGN TABLE {schema}.edge_pq ()
+        SERVER pg_lake
+        OPTIONS (path '{url}')
+        """,
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    result = run_query(
+        f'SELECT id, (e).plain, (e)."with space", (e).U&"has\\0022one", '
+        f'(e)."comma,here", (e)."colon:here", (e)."with(parens)", '
+        f'(e).U&"back\\005Cslash" FROM {schema}.edge_pq',
+        pg_conn,
+    )
+    assert result == [[1, 1, 2, 3, 4, 5, 6, 7]]
+
+    run_command(f"DROP SCHEMA {schema} CASCADE", pg_conn)
+    pg_conn.commit()
+
+
 def test_iceberg_composite_field_with_embedded_double_quote(pg_conn, s3, extension):
     """
     Composite type field names containing double-quote characters must be