Add pg_lake_table.enable_partitioned_write_pushdown GUC

Adds a setting to control whether partitioned writes use DuckDB
PARTITION_BY pushdown (default: on). When off, partitioned writes
fall back to row-by-row processing which supports file_size_bytes
splitting. Applies to both INSERT..SELECT and COPY FROM paths.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: sfc-gh-mslot

pg_lake_copy/src/copy/copy.c

@@ -659,6 +659,9 @@ IsCopyFromPushdownable(Relation relation, List *columnNameList,
 	 */
 	const char *partitionBy = GetIcebergTablePartitionByOption(relationId);
 
+	if (partitionBy != NULL && !EnablePartitionedWritePushdown)
+		return false;
+
 	if (partitionBy != NULL &&
 		GetPartitionByExpressionsForRelation(relationId) == NIL)
 		return false;

pg_lake_table/include/pg_lake/planner/insert_select.h

@@ -23,6 +23,9 @@
 /* pg_lake_table.enable_insert_select_pushdown setting */
 extern bool EnableInsertSelectPushdown;
 
+/* pg_lake_table.enable_partitioned_write_pushdown setting */
+extern PGDLLEXPORT bool EnablePartitionedWritePushdown;
+
 bool		IsPushdownableInsertSelectQuery(Query *query);
 bool		IsInsertSelectQuery(Query *query);
 Oid			GetInsertRelidFromInsertSelect(Query *query);

pg_lake_table/src/init.c

@@ -129,6 +129,20 @@ _PG_init(void)
 							 NULL,
 							 NULL);
 
+	DefineCustomBoolVariable("pg_lake_table.enable_partitioned_write_pushdown",
+							 "Enables pushdown of partitioned writes to DuckDB "
+							 "using PARTITION_BY. When off, partitioned writes "
+							 "use row-by-row processing which supports "
+							 "file_size_bytes splitting.",
+							 NULL,
+							 &EnablePartitionedWritePushdown,
+							 true,
+							 PGC_USERSET,
+							 GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE,
+							 NULL,
+							 NULL,
+							 NULL);
+
 	DefineCustomBoolVariable("pg_lake_table.enable_data_file_pruning",
 							 "Enables data file pruning based on the metadata statistics "
 							 "for iceberg tables.",

pg_lake_table/src/planner/insert_select.c

@@ -44,6 +44,9 @@ static bool TypeContainsUnsuitableForPushdown(Oid typeId, int32 typmod, CopyData
 /* pg_lake_table.enable_insert_select_pushdown setting */
 bool		EnableInsertSelectPushdown = true;
 
+/* pg_lake_table.enable_partitioned_write_pushdown setting */
+bool		EnablePartitionedWritePushdown = true;
+
 /*
  * IsPushdownableInsertSelectQuery checks whether the given query is an INSERT..SELECT
  * that can be delegated.
@@ -126,6 +129,15 @@ IsPushdownableInsertSelectQuery(Query *query)
 	 */
 	const char *partitionBy = GetIcebergTablePartitionByOption(insertIntoRelid);
 
+	if (partitionBy != NULL && !EnablePartitionedWritePushdown)
+	{
+		ereport(DEBUG4,
+				(errmsg("INSERT..SELECT into partitioned table is not "
+						"pushdownable (enable_partitioned_write_pushdown is off)")));
+		RelationClose(insertRelation);
+		return false;
+	}
+
 	if (partitionBy != NULL &&
 		GetPartitionByExpressionsForRelation(insertIntoRelid) == NIL)
 	{