Add test for enable_partitioned_write_pushdown GUC

Verifies that when the GUC is off, partitioned INSERT..SELECT
falls back to row-by-row while non-partitioned pushdown remains
unaffected, and that the row-by-row path produces correct results.

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

Author: sfc-gh-mslot

pg_lake_table/tests/pytests/test_partitioned_pushdown.py

@@ -2113,3 +2113,108 @@ def test_partition_evolution_pushdown(
     assert src_data == tgt_data
 
     _drop_tables(pg_conn, "evo_src", "evo_tgt")
+
+
+def test_enable_partitioned_write_pushdown_guc(
+    extension,
+    s3,
+    with_default_location,
+    pg_conn,
+    pgduck_conn,
+    grant_access_to_data_file_partition,
+):
+    """
+    When pg_lake_table.enable_partitioned_write_pushdown is off,
+    partitioned INSERT..SELECT falls back to row-by-row processing
+    but still produces correct results. Non-partitioned pushdown
+    remains unaffected.
+    """
+    _setup_schema(pg_conn)
+    _drop_tables(pg_conn, "guc_src", "guc_tgt_part", "guc_tgt_nopart")
+    src = _table("guc_src")
+    tgt_part = _table("guc_tgt_part")
+    tgt_nopart = _table("guc_tgt_nopart")
+
+    # source table
+    run_command(
+        f"""
+        CREATE TABLE {src}(id INT, ts TIMESTAMP)
+        USING iceberg WITH (autovacuum_enabled = false);
+        INSERT INTO {src}
+        SELECT i, '2025-01-01'::timestamp + (i || ' days')::interval
+        FROM generate_series(1, 20) AS i;
+        """,
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    # partitioned target
+    run_command(
+        f"""
+        CREATE TABLE {tgt_part}(id INT, ts TIMESTAMP)
+        USING iceberg
+        WITH (partition_by = 'month(ts)', autovacuum_enabled = false);
+        """,
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    # non-partitioned target
+    run_command(
+        f"""
+        CREATE TABLE {tgt_nopart}(id INT, ts TIMESTAMP)
+        USING iceberg WITH (autovacuum_enabled = false);
+        """,
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    # verify pushdown is on by default for partitioned tables
+    assert_query_pushdownable(
+        f"INSERT INTO {tgt_part} SELECT * FROM {src}", pg_conn
+    )
+
+    # disable partitioned write pushdown
+    run_command(
+        "SET pg_lake_table.enable_partitioned_write_pushdown TO false;", pg_conn
+    )
+
+    # partitioned INSERT..SELECT should no longer be pushdownable
+    assert_query_not_pushdownable(
+        f"INSERT INTO {tgt_part} SELECT * FROM {src}", pg_conn
+    )
+
+    # non-partitioned INSERT..SELECT should still be pushdownable
+    assert_query_pushdownable(
+        f"INSERT INTO {tgt_nopart} SELECT * FROM {src}", pg_conn
+    )
+
+    # insert via row-by-row fallback should still produce correct results
+    run_command(f"INSERT INTO {tgt_part} SELECT * FROM {src};", pg_conn)
+    pg_conn.commit()
+
+    res = run_query(
+        f"""
+        SELECT count(*), count(DISTINCT id)
+        FROM {tgt_part};
+        """,
+        pg_conn,
+    )
+    assert res[0][0] == 20
+    assert res[0][1] == 20
+
+    # verify partition files are correct
+    file_cnt = run_query(
+        f"""
+        SELECT count(*)
+        FROM lake_table.files
+        WHERE table_name = '{tgt_part}'::regclass;
+        """,
+        pg_conn,
+    )[0][0]
+    assert file_cnt == 1  # all in January 2025
+
+    run_command(
+        "RESET pg_lake_table.enable_partitioned_write_pushdown;", pg_conn
+    )
+    _drop_tables(pg_conn, "guc_src", "guc_tgt_part", "guc_tgt_nopart")