test: native_agg's pytest twin covers every shell arm (#1341)

test/pytest/TESTS.md

@@ -6554,14 +6554,15 @@ THE ROW COUNT IS A PREMISE because the shell twin's first draft overflowed `int`
 
 Removal proof, on this twin: forcing every group dirty reddens `max(ts) reads under a tenth of the blocks a decode of ts reads` (got -1487, wanted at least 1).
 
-Not ported, and declared INCOMPLETE in `test_compare_to_bash.py`: the `int4`/`text` parity, `sum`/`avg`, parallel-plan and fallback arms.
+The rest of the shell suite was ported for #1341, every check by its name: the `int4`, `text` and `sum`/`avg` parity arms with the metadata node asserted first, the node winning against a parallel plan with no Gather (#133), `count(*)` reading no zone maps against an aggregate that must read them, and the filtered, `sum(bigint)` and empty-table fallbacks. Removal proof for those: `count(col)` folded from the row count instead of the value count reddens `count(col,nulls)` in both twins.
 
 ### Every test
 
 | test | what it holds |
 | --- | --- |
 | `test_min_max_over_timestamptz_read_no_data` | the table holds every row; `max(ts)` plans the metadata node; a decode reads at least 10 blocks; `max` and `min` each read under a tenth of that; parity with heap before and after deleting rows at both ends |
 | `test_a_where_with_no_column_in_it_is_applied` | a WHERE that references no column, such as `(SELECT false)`, `now() < ...` or `current_user = ...`, gives heap's answer, with a premise that the same aggregate with no WHERE is the vectorized node (#1435) |
+| `test_the_metadata_aggregates_answer_as_heap_and_win_their_plans` | count, sum, avg, min and max over int4 and text equal heap with the metadata node asserted first; the node beats a parallel plan with no Gather; `count(*)` reads no zone maps while `count(*), sum(id)` does; filtered, `sum(bigint)` and empty-table answers are right |
 
 ## 86. test_format_downgrade.py: an alpha4 binary refuses what this build wrote
 

test/pytest/expected_tests.txt

@@ -817,4 +817,6 @@ guard_tests 439
 # two tests, by collection over the cluster files on this tree.
 # cluster_tests 749 -> 750 for #1341's test_a_fact_qual_conjunction_with_late_materialization_matches_heap,
 # when test_join_runtime_filter.py became test_native_join_runtime_filter.py: one test, by collection.
-cluster_tests 750
+# cluster_tests 750 -> 751 for #1341's completion of test_native_agg.py, the twin of native_agg.sh:
+# one test, by collection over the cluster files on this tree.
+cluster_tests 751

test/pytest/test_compare_to_bash.py

@@ -119,6 +119,7 @@
             "latest_reading",
             "maintenance_crash",
             "maintenance_rollback",
+            "native_agg",
             "native_backend_crash",
             "native_batch_fold_projection",
             "native_chunk_length_bound",
@@ -249,12 +250,6 @@
     # having no SKIP verdict for an arm a run chooses not to make, does not carry.
     "large_values": "the 1 MB to 200 MB round trips and the TOASTed-update arms are "
                     "ported; the opt-in chunk-over-1-GB refusal is not",
-    # #1293 ported the arms it added: min/max over timestamptz read no data, and
-    # stay right after deletes. The int4/text parity, sum/avg, parallel-plan and
-    # fallback arms of the shell suite predate it and are #432's to port.
-    "native_agg": "the timestamptz min/max arms (blocks read against a decode "
-                  "control, parity before and after deletes) are ported; the "
-                  "int4/text parity, sum/avg, parallel-plan and fallback arms are not",
     # The port asserts the engine counters, the heap oracle, the scattered zero
     # and the declared-collation property. The shell suite additionally pins the
     # three states of max_upper and its own fixture geometry by name, and those
@@ -422,7 +417,7 @@ def test_the_loop_reader_invents_nothing_in_this_corpus(expect):
                 "every name the loop reader returns appears verbatim in the file it "
                 "came from")
     expect.text(", ".join(files),
-                "build_refusal, differential, native_groupagg_batch, "
+                "build_refusal, differential, native_agg, native_groupagg_batch, "
                 "native_index_projection, native_parquet_pushdown, "
                 "native_reclaim, native_repack, "
                 "native_saop_pushdown, native_vacuum_race, native_vecskip, sorted_pathkeys, "

test/pytest/test_native_agg.py

@@ -11,11 +11,10 @@
 against a decode's 403, and every parity arm stayed green. So each arm reads the
 table's own blocks from pg_statio against a control that must decode the column.
 
-A partial port of `test/native_agg.sh`: the timestamptz arms only. The int4,
-text, sum/avg, parallel and fallback arms are not ported, and
-`test_compare_to_bash.py` declares the pair INCOMPLETE with that reason.
-Independent of the shell twin: its own table, row count, key spacing and
-delete windows.
+The pytest twin of `test/native_agg.sh`, every check by the shell's name. The
+timestamptz arms came first (#1293); the int4, text, sum/avg, parallel-plan,
+zone-map-read, fallback and empty-table arms were ported for #1341. Independent of
+the shell twin: its own tables, row counts, key spacing and delete windows.
 
 `pg_stat_reset()` is database-wide; this file must not run concurrently with
 another that reads statistics.
@@ -136,9 +135,128 @@ def test_a_where_with_no_column_in_it_is_applied(pgc_conn, expect):
         plan = "\n".join(r[0] for r in cur.fetchall())
     expect.text("yes" if "Columnar Vectorized Aggregates" in plan else f"no ({plan})", "yes",
                 "premise: count, min and max with no WHERE are the vectorized aggregate")
-    for name, where in (("(SELECT false)", "(SELECT false)"), ("now()", "now() < '2000-01-01'"),
-                        ("current_user", "current_user = 'pgc_nobody'")):
-        expect.text(_one(pgc_conn, f"{GATED} gn WHERE {where}"), _one(pgc_conn, f"{GATED} gh WHERE {where}"),
-                    f"a WHERE {'' if name.startswith('(') else 'on '}{name} with no column in it gives heap's answer")
+    # WRITTEN OUT, so each name is the shell's literal rather than a built string
+    for where, name in (
+            ("(SELECT false)", "a WHERE (SELECT false) with no column in it gives heap's answer"),
+            ("now() < '2000-01-01'", "a WHERE on now() with no column in it gives heap's answer"),
+            ("current_user = 'pgc_nobody'",
+             "a WHERE on current_user with no column in it gives heap's answer")):
+        expect.text(_one(pgc_conn, f"{GATED} gn WHERE {where}"),
+                    _one(pgc_conn, f"{GATED} gh WHERE {where}"), name)
     expect.text(_one(pgc_conn, f"{GATED} gh WHERE (SELECT false)"), "0/-/-",
                 "premise: heap's answer to a false gate is no rows")
+
+
+# ---- the int4, text, parallel, zone-map and fallback arms (#1341) ------------------
+
+def _plan(conn, sql, parallel=False):
+    with conn.cursor() as cur:
+        if parallel:
+            # parallel_tuple_cost keeps its default on purpose: zeroing it makes a
+            # Gather of raw rows look free, and the wrong costing then still picks
+            # the right plan (the shell's comment, measured there)
+            cur.execute("SET max_parallel_workers_per_gather = 4")
+            cur.execute("SET parallel_setup_cost = 0")
+            cur.execute("SET min_parallel_table_scan_size = 0")
+        cur.execute("EXPLAIN (COSTS OFF) " + sql)
+        plan = "\n".join(r[0] for r in cur.fetchall())
+        if parallel:
+            for g in ("max_parallel_workers_per_gather", "parallel_setup_cost",
+                      "min_parallel_table_scan_size"):
+                cur.execute(f"RESET {g}")
+    return plan
+
+
+def _aggnode(plan):
+    return "yes" if "Columnar Vectorized Aggregates" in plan else "no"
+
+
+def _gather(plan):
+    return "yes" if "Gather" in plan else "no"
+
+
+def _zm_scans(conn):
+    with conn.cursor() as cur:
+        cur.execute("SELECT pg_stat_force_next_flush()")
+        cur.execute("SELECT coalesce(idx_scan, 0) + coalesce(seq_scan, 0) FROM pg_stat_all_tables "
+                    "WHERE relid = 'pgcolumnar.zone_map'::regclass")
+        return int(cur.fetchone()[0])
+
+
+def test_the_metadata_aggregates_answer_as_heap_and_win_their_plans(pgc_conn, expect):
+    c = pgc_conn
+    with c.cursor() as cur:
+        cur.execute("CREATE TABLE ah (id int, s int, k bigint, label text)")
+        cur.execute("CREATE TABLE an (id int, s int, k bigint, label text) USING pgcolumnar")
+        cur.execute("SELECT pgcolumnar.set_options('an', stripe_row_limit => 2048, "
+                    "chunk_group_row_limit => 1024)")
+        gen = ("SELECT g, CASE WHEN g % 5 = 0 THEN NULL ELSE g % 997 END, (g * 11)::bigint, "
+               "'tag-' || (g % 37) FROM generate_series(1, 12000) g")
+        cur.execute(f"INSERT INTO ah {gen}")
+        cur.execute(f"INSERT INTO an {gen}")
+
+    # the node first: every parity arm below also holds for a correct fallback
+    expect.text(_aggnode(_plan(c, "SELECT count(*) FROM an")), "yes",
+                "count(*) uses the metadata agg node")
+    expect.text(_aggnode(_plan(c, "SELECT sum(id), min(id), max(id) FROM an")), "yes",
+                "sum/min/max uses the metadata agg node")
+
+    expect.text(str(_one(c, "SELECT count(*) FROM an")), str(_one(c, "SELECT count(*) FROM ah")), "count(*)")
+    expect.text(str(_one(c, "SELECT count(s) FROM an")), str(_one(c, "SELECT count(s) FROM ah")),
+                "count(col,nulls)")
+    expect.text(str(_one(c, "SELECT sum(id) FROM an")), str(_one(c, "SELECT sum(id) FROM ah")), "sum(int4)")
+    expect.text(str(_one(c, "SELECT sum(s) FROM an")), str(_one(c, "SELECT sum(s) FROM ah")),
+                "sum(int4,nulls)")
+    expect.text(str(_one(c, "SELECT avg(s) FROM an")), str(_one(c, "SELECT avg(s) FROM ah")),
+                "avg(int4,nulls)")
+    expect.text(str(_one(c, "SELECT min(id) FROM an")), str(_one(c, "SELECT min(id) FROM ah")), "min(int4)")
+    expect.text(str(_one(c, "SELECT max(id) FROM an")), str(_one(c, "SELECT max(id) FROM ah")), "max(int4)")
+    expect.text(str(_one(c, "SELECT min(s) FROM an")), str(_one(c, "SELECT min(s) FROM ah")),
+                "min(int4,nulls)")
+    expect.text(str(_one(c, "SELECT min(label) FROM an")), str(_one(c, "SELECT min(label) FROM ah")),
+                "min(text)")
+    expect.text(str(_one(c, "SELECT max(label) FROM an")), str(_one(c, "SELECT max(label) FROM ah")),
+                "max(text)")
+    with c.cursor() as cur:
+        cur.execute("SELECT count(*), sum(id), min(id), max(id) FROM an")
+        got = cur.fetchone()
+        cur.execute("SELECT count(*), sum(id), min(id), max(id) FROM ah")
+        want = cur.fetchone()
+    expect.text(str(got), str(want), "multi-agg")
+
+    # it has to win on cost, not only exist (#133)
+    par_count = _plan(c, "SELECT count(*) FROM an", parallel=True)
+    expect.text(_aggnode(par_count), "yes", "count(*) beats a parallel plan")
+    expect.text(_gather(par_count), "no", "count(*) plan has no Gather")
+    par_multi = _plan(c, "SELECT count(*), sum(id), min(id), max(id) FROM an", parallel=True)
+    expect.text(_aggnode(par_multi), "yes", "multi-agg beats a parallel plan")
+    expect.text(_gather(par_multi), "no", "multi-agg plan has no Gather")
+
+    # count(*) needs no zone map; an aggregate that needs one must be seen reading it,
+    # so the zero above cannot pass because the counter never moves
+    before = _zm_scans(c)
+    _one(c, "SELECT count(*) FROM an")
+    after = _zm_scans(c)
+    expect.num(after - before, 0, "count(*) reads no zone maps")
+    before = _zm_scans(c)
+    _one(c, "SELECT count(*), sum(id) FROM an")
+    after = _zm_scans(c)
+    expect.text("yes" if after - before > 0 else f"no ({after - before})", "yes",
+                "an aggregate that needs zone maps reads them")
+
+    with c.cursor() as cur:
+        cur.execute("SELECT count(*), sum(id) FROM an WHERE id > 6000")
+        got = cur.fetchone()
+        cur.execute("SELECT count(*), sum(id) FROM ah WHERE id > 6000")
+        want = cur.fetchone()
+    expect.text(str(got), str(want), "filtered aggregate parity")
+    expect.text(_aggnode(_plan(c, "SELECT count(*), sum(id) FROM an WHERE id > 6000")), "no",
+                "filtered aggregate not metadata node")
+    expect.text(str(_one(c, "SELECT sum(k) FROM an")), str(_one(c, "SELECT sum(k) FROM ah")),
+                "sum(bigint) fallback parity")
+
+    with c.cursor() as cur:
+        cur.execute("CREATE TABLE ae (id int, label text) USING pgcolumnar")
+    expect.num(_one(c, "SELECT count(*) FROM ae"), 0, "empty count(*)")
+    expect.text(str(_one(c, "SELECT sum(id) IS NULL FROM ae")), "True", "empty sum NULL")
+    expect.text(str(_one(c, "SELECT min(label) IS NULL FROM ae")), "True", "empty min NULL")