diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index ffbd47f9a..bed4cafb0 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -6573,9 +6573,26 @@ The rest of the shell suite was ported for #1341, every check by its name: the ` | 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 | +| `test_the_timestamptz_table_holds_every_row` | **premise**: the table holds every row inserted | +| `test_max_ts_plans_the_metadata_agg_node` | `max(ts)` over timestamptz plans the metadata node | +| `test_decoding_ts_reads_the_tables_blocks` | **control**: a decode of `ts` reads at least 10 blocks | +| `test_min_max_ts_read_under_a_tenth_of_a_decode` | `max(ts)` and `min(ts)` each read under a tenth of the decode's blocks (parametrized) | +| `test_timestamptz_parity` | `max` and `min` equal heap's, before and after deleting rows at both ends (parametrized) | +| `test_the_deletes_removed_the_same_rows_at_both_ends` | the deletes removed rows, the same rows from each table | +| `test_count_min_max_with_no_where_are_the_vectorized_aggregate` | **premise**: with no WHERE, the aggregate is the vectorized node | +| `test_a_where_with_no_column_in_it_gives_heaps_answer` | `(SELECT false)`, `now() < ...` and `current_user = ...` each give heap's answer (#1435) (parametrized) | +| `test_heaps_answer_to_a_false_gate_is_no_rows` | **premise**: heap's answer to a false gate is no rows | +| `test_the_aggregate_uses_the_metadata_node` | `count(*)` and sum/min/max plan the metadata node (parametrized) | +| `test_the_aggregate_answers_as_heap` | count, sum, avg, min and max over int4 and text, the multi-aggregate, the filtered aggregate and the `sum(bigint)` fallback each equal heap (parametrized, 13 names) | +| `test_the_metadata_node_beats_a_parallel_plan` | the node wins against a parallel plan, for `count(*)` and the multi-aggregate (#133) (parametrized) | +| `test_the_metadata_plan_has_no_gather` | neither plan has a Gather (parametrized) | +| `test_count_reads_no_zone_maps` | `count(*)` reads no zone maps | +| `test_an_aggregate_that_needs_zone_maps_reads_them` | `count(*), sum(id)` reads at least one: the measured delta, so the zero above cannot pass on a counter that never moves | +| `test_a_filtered_aggregate_is_not_the_metadata_node` | a filtered aggregate falls back | +| `test_an_empty_table_counts_zero` | an empty table counts 0 | +| `test_an_empty_table_aggregates_to_null` | an empty table's `sum` and `min` are NULL (parametrized) | + +ONE CHECK PER TEST (#1339). Three module fixtures take every reading in order, and each test asserts one, so a red cannot stop the next check from running. Measured with #133 restored (the metadata path priced at the scan's cost) on PG17: the shell reddens 4 arms, the parallel-plan and no-Gather pairs. The earlier twin, three tests holding 40 checks, reported 1. This one reports the same 4. ## 86. test_format_downgrade.py: an alpha4 binary refuses what this build wrote @@ -6606,7 +6623,7 @@ A ZERO IS NOT A MEASUREMENT. On the shell twin, the red run before the section e THE FIXTURE NEEDS MANY ROWS PER DEVICE. The `LATERAL` form costs about a block per device whatever the table's size, so at 5,000 rows per device it read 45 blocks of a 177-page table and "a tenth" failed. Here it is 36,000 rows per device. -On a private database, because the document names its tables, `readings` and `devices`. +On a private database, built once by a module fixture, because the document names its tables, `readings` and `devices`. Removal proof: the document's `LATERAL` edited to `ORDER BY ts` (ascending) returns each device's first reading, and `the LATERAL form returns what DISTINCT ON returns` fails. On the shell twin that is the only red, on every major. @@ -6614,7 +6631,17 @@ Removal proof: the document's `LATERAL` edited to `ORDER BY ts` (ascending) retu | test | what it holds | | --- | --- | -| `test_the_latest_reading_recipe_does_what_the_document_says` | the document's three blocks are found; the table holds its rows; without the btree the `LATERAL` form reads the table about once per device; the `GROUP BY` form and, with the document's index, the `LATERAL` form return the rows `DISTINCT ON` returns; the `GROUP BY` form reads at least one block and no more than the table, as the document says (#1339); the indexed `LATERAL` reads at least one block and under a tenth of the table | +| `test_the_documents_three_recipe_blocks_were_found` | **premise**: the document's three recipe blocks are found | +| `test_the_table_holds_every_row` | **premise**: the table holds every row | +| `test_without_the_btree_the_lateral_form_reads_the_table_once_per_device` | the trap the document warns about: without the btree, the `LATERAL` form reads the table about once per device | +| `test_the_group_by_form_returns_what_distinct_on_returns` | the `GROUP BY` form returns the rows `DISTINCT ON` returns | +| `test_the_group_by_form_reads_the_table_once` | the `GROUP BY` form reads at least one block and no more than the table, as the document says (#1339) | +| `test_the_documents_index_exists` | **premise**: the document's index exists | +| `test_the_lateral_form_returns_what_distinct_on_returns` | with the index, the `LATERAL` form returns the rows `DISTINCT ON` returns | +| `test_the_indexed_lateral_form_read_the_table_at_all` | **premise**: the indexed `LATERAL` form read at least one block | +| `test_with_the_btree_the_lateral_form_reads_under_a_tenth_of_the_table` | with the index, the `LATERAL` form reads under a tenth of the table | + +ONE CHECK PER TEST (#1339). A module fixture runs every step in the shell's order, the trap before the document's index exists, and records each reading; each test asserts one. Measured with the document's index statement replaced by `SELECT 1` on PG17: the shell reddens "the document's index exists" and "reads under a tenth of the table". The earlier one-test twin reported only the first. This one reports both. ## 88. test_storage_row_repair.py: the alpha5 release notes' repair procedure works as written diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt index b14ab1a9a..8c755df65 100644 --- a/test/pytest/expected_tests.txt +++ b/test/pytest/expected_tests.txt @@ -833,4 +833,6 @@ guard_tests 440 # eight tests, one per check, by collection over the cluster files on this tree. # cluster_tests 787 -> 792 for #1341's completion of test_native_skip.py, the twin of native_skip.sh: # five tests, by collection over the cluster files on this tree. -cluster_tests 792 +# cluster_tests 792 -> 837 for #1339, one check per test in test_native_agg.py (3 -> 40 +# tests) and test_latest_reading.py (1 -> 9 tests), by collection over the cluster files. +cluster_tests 837 diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py index de1d8befd..49584dc3a 100644 --- a/test/pytest/test_compare_to_bash.py +++ b/test/pytest/test_compare_to_bash.py @@ -416,7 +416,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_agg, native_groupagg_batch, " + "build_refusal, differential, native_groupagg_batch, " "native_index_projection, native_parquet_pushdown, " "native_reclaim, native_repack, " "native_saop_pushdown, native_vacuum_race, native_vecskip, sorted_pathkeys, " diff --git a/test/pytest/test_latest_reading.py b/test/pytest/test_latest_reading.py index 9b072a783..6b87bb4f1 100644 --- a/test/pytest/test_latest_reading.py +++ b/test/pytest/test_latest_reading.py @@ -11,8 +11,10 @@ under a tenth of the table, and at least one block (a query that reads nothing would otherwise pass); without it, it reads the table about once per device. -ON A PRIVATE DATABASE (`pgc_own_db`): the document names its tables, `readings` -and `devices`, so they cannot carry a per-test prefix. Independent of +ON A PRIVATE DATABASE, built once by a module fixture: the document names its +tables, `readings` and `devices`, so they cannot carry a per-test prefix. The fixture +runs every step in the shell's order and records each reading; each test asserts one, +so one red cannot stop the next from running (#1339, as #1517-#1519 do). Independent of `test/latest_reading.sh`: its own extraction of the document (Python, not awk), its own data, device count and row count. """ @@ -20,6 +22,9 @@ import pathlib import re +import psycopg +import pytest + DOC = pathlib.Path(__file__).resolve().parent.parent.parent / "docs" / "how-to.md" SECTION = "## Find the latest reading per device" DEVICES = 25 @@ -63,55 +68,114 @@ def _blocks(conn, sql): return int(row[0]) if row else None -def test_the_latest_reading_recipe_does_what_the_document_says(pgc_own_db, expect): - conn = pgc_own_db - blocks = _recipe_blocks() - expect.num(len(blocks), 3, "premise: the document's three recipe blocks were found") - if len(blocks) != 3: - return - index_sql, group_sql, lateral_sql = blocks - - with conn.cursor() as cur: - cur.execute("CREATE TABLE readings (device int, ts timestamptz, v float8) USING pgcolumnar") - cur.execute("CREATE TABLE devices (device int PRIMARY KEY)") - cur.execute(f"INSERT INTO devices SELECT g FROM generate_series(1, {DEVICES}) g") - cur.execute( - f"INSERT INTO readings SELECT 1 + (h % {DEVICES}), '2025-12-01'::timestamptz" - f" + (h % {ROWS}) * interval '30 s', (h % 613)::float8" - f" FROM (SELECT (hashint8(g) & 9223372036854775807) AS h" - f" FROM generate_series(1::bigint, {ROWS}) g) x" - ) - expect.num(_one(conn, "SELECT count(*) FROM readings"), ROWS, - "premise: the table holds every row") - - oracle = _rows(conn, "SELECT DISTINCT ON (device) device, ts FROM readings" - " ORDER BY device, ts DESC") - pages = _one(conn, "SELECT pg_relation_size('readings') / 8192") - - trap = _blocks(conn, lateral_sql) - print(f"-- table {pages} pages; LATERAL without the btree read {trap} blocks") - expect.at_least(trap, pages * DEVICES // 2, +def _try(fn, *args): + """A reading, or the error that replaced it: a failed step reddens its own test + rather than erroring the fixture and every test with it.""" + try: + return fn(*args) + except psycopg.Error as exc: + return f"error: {exc.diag.message_primary}" + + +@pytest.fixture(scope="module") +def recipe(pgc_cluster): + """-> the readings, taken in the shell's order: the trap before the document's + index exists, the indexed form after.""" + db = "pgc_latest_reading" + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {db} WITH (FORCE)") + a.execute(f"CREATE DATABASE {db}") + got = {"blocks": _recipe_blocks()} + try: + with psycopg.connect(pgc_cluster.dsn(dbname=db), autocommit=True) as conn: + conn.execute("CREATE EXTENSION pgcolumnar") + conn.execute("CREATE TABLE readings (device int, ts timestamptz, v float8) USING pgcolumnar") + conn.execute("CREATE TABLE devices (device int PRIMARY KEY)") + conn.execute(f"INSERT INTO devices SELECT g FROM generate_series(1, {DEVICES}) g") + _try(conn.execute, + f"INSERT INTO readings SELECT 1 + (h % {DEVICES}), '2025-12-01'::timestamptz" + f" + (h % {ROWS}) * interval '30 s', (h % 613)::float8" + f" FROM (SELECT (hashint8(g) & 9223372036854775807) AS h" + f" FROM generate_series(1::bigint, {ROWS}) g) x") + got["count"] = _try(_one, conn, "SELECT count(*) FROM readings") + got["oracle"] = str(_try(_rows, conn, "SELECT DISTINCT ON (device) device, ts FROM readings" + " ORDER BY device, ts DESC")) + got["pages"] = _try(_one, conn, "SELECT pg_relation_size('readings') / 8192") + if len(got["blocks"]) == 3: + index_sql, group_sql, lateral_sql = got["blocks"] + got["trap"] = _try(_blocks, conn, lateral_sql) + got["group rows"] = str(_try(_rows, conn, group_sql)) + got["group blocks"] = _try(_blocks, conn, group_sql) + _try(conn.execute, index_sql) + got["indexes"] = _try(_one, conn, "SELECT count(*) FROM pg_indexes WHERE tablename = 'readings'") + got["lateral rows"] = str(_try(_rows, conn, lateral_sql)) + got["lateral blocks"] = _try(_blocks, conn, lateral_sql) + print(f"-- table {got['pages']} pages; readings: " + + ", ".join(f"{k}={got.get(k)}" for k in ("trap", "group blocks", "lateral blocks"))) + yield got + finally: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {db} WITH (FORCE)") + + +def _int(v, what=""): + """A block count, or 0 with the error printed, so an error or an absent step + fails the bound it was meant to meet by name, with its reason in the output.""" + if isinstance(v, int): + return v + print(f"-- {what or 'reading'}: not a count: {v}") + return 0 + + +def _num(v, what=""): + """A count for expect.num, or -1 with the error printed: expect.num refuses a + non-number with a VacuityError rather than a named red (#1522 review).""" + return v if isinstance(v, int) else _int(v, what) - 1 + + +def test_the_documents_three_recipe_blocks_were_found(recipe, expect): + expect.num(len(recipe["blocks"]), 3, "premise: the document's three recipe blocks were found") + + +def test_the_table_holds_every_row(recipe, expect): + expect.num(_num(recipe["count"], "count"), ROWS, "premise: the table holds every row") + + +def test_without_the_btree_the_lateral_form_reads_the_table_once_per_device(recipe, expect): + expect.at_least(_int(recipe.get("trap")), _int(recipe["pages"], "pages") * DEVICES // 2, "without the btree the LATERAL form reads the table once per device") - expect.text(str(_rows(conn, group_sql)), str(oracle), + +def test_the_group_by_form_returns_what_distinct_on_returns(recipe, expect): + expect.text(recipe.get("group rows", "absent"), recipe["oracle"], "the GROUP BY form returns what DISTINCT ON returns") + + +def test_the_group_by_form_reads_the_table_once(recipe, expect): # The document says this form reads the table once (#1339). Rows alone cannot tell # it from a per-device subquery, which returns the same rows and reads the table # once per device. Reading nothing also reads under the table, so a zero is refused. - grp = _blocks(conn, group_sql) - print(f"-- table {pages} pages; the GROUP BY form read {grp} blocks") - expect.at_least(pages + 1 - grp if grp else 0, 1, + grp = _int(recipe.get("group blocks")) + expect.at_least(_int(recipe["pages"], "pages") + 1 - grp if grp else 0, 1, "the GROUP BY form reads the table once") - with conn.cursor() as cur: - cur.execute(index_sql) - expect.num(_one(conn, "SELECT count(*) FROM pg_indexes WHERE tablename = 'readings'"), 1, - "premise: the document's index exists") - expect.text(str(_rows(conn, lateral_sql)), str(oracle), + +def test_the_documents_index_exists(recipe, expect): + expect.num(_num(recipe.get("indexes", 0), "indexes"), 1, "premise: the document's index exists") + + +def test_the_lateral_form_returns_what_distinct_on_returns(recipe, expect): + expect.text(recipe.get("lateral rows", "absent"), recipe["oracle"], "the LATERAL form returns what DISTINCT ON returns") - lat = _blocks(conn, lateral_sql) - print(f"-- with the btree the LATERAL form read {lat} blocks") + + +def test_the_indexed_lateral_form_read_the_table_at_all(recipe, expect): # Reading nothing also reads under a tenth of the table, so a zero is refused. - expect.at_least(lat, 1, "premise: the indexed LATERAL form read the table at all") - expect.at_least(pages - 10 * lat, 1, + expect.at_least(_int(recipe.get("lateral blocks")), 1, + "premise: the indexed LATERAL form read the table at all") + + +def test_with_the_btree_the_lateral_form_reads_under_a_tenth_of_the_table(recipe, expect): + lat = _int(recipe.get("lateral blocks")) + expect.at_least(_int(recipe["pages"], "pages") - 10 * lat if lat else 0, 1, "with the btree the LATERAL form reads under a tenth of the table") diff --git a/test/pytest/test_native_agg.py b/test/pytest/test_native_agg.py index 0f93126d2..920a1ab32 100644 --- a/test/pytest/test_native_agg.py +++ b/test/pytest/test_native_agg.py @@ -16,10 +16,19 @@ 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. +ONE CHECK PER TEST (#1339). Three module fixtures, one per data set, take every +reading in the order the arms need it -- the deletes, the zone-map counter deltas and +the parallel plans happen in the fixture, never inside a test -- and each test asserts +one reading. A red then cannot stop the next check from running, so the twin reports +the same red set as the shell (#1514's review, item 4). + `pg_stat_reset()` is database-wide; this file must not run concurrently with another that reads statistics. """ +import psycopg +import pytest + ROWS = 300000 @@ -30,6 +39,18 @@ def _one(conn, sql): return row[0] if row else None +def _try(fn, *args, **kw): + """A reading, or the error that replaced it, so a failed step reddens its own + check rather than erroring the fixture and every check with it. EVERY reading in + a fixture goes through this (#1522 review): one bare call that raises takes down + every test on that fixture as an unnamed ERROR, which is the hiding this file's + one-check-per-test shape exists to prevent.""" + try: + return fn(*args, **kw) + except psycopg.Error as exc: + return f"error: {exc.diag.message_primary}" + + def _blocks(conn, table, sql): """Blocks of `table` read while `sql` ran. @@ -52,10 +73,51 @@ def _blocks(conn, table, sql): return int(row[0]) if row else None -def test_min_max_over_timestamptz_read_no_data(pgc_conn, expect): - with pgc_conn.cursor() as cur: - cur.execute("CREATE TABLE readings (sensor bigint, at timestamptz) USING pgcolumnar") - cur.execute("SELECT pgcolumnar.set_options('readings', stripe_row_limit => 2000)") +def _int(v, what=""): + """A count, or 0 with the error printed, so a bound it was meant to meet fails by + name and the reason is in the output rather than lost (#1522 review).""" + if isinstance(v, int): + return v + print(f"-- {what or 'reading'}: not a count: {v}") + return 0 + + +def _num(v, what=""): + """A count for expect.num, or -1 with the error printed. expect.num refuses a + non-number with a VacuityError rather than a named red (#1522 review).""" + return v if isinstance(v, int) else _int(v, what) - 1 + + +def _explain(c, sql): + return "\n".join(r[0] for r in c.execute(sql)) + + +def _schema_conn(pgc_cluster, schema): + """A connection whose private schema is first on the search_path, as pgc_conn + gives a test, but for a module fixture.""" + conn = psycopg.connect(pgc_cluster.dsn(), autocommit=True) + conn.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE") + conn.execute(f"CREATE SCHEMA {schema}") + conn.execute(f"SET search_path = {schema}, public") + return conn + + +def _drop(conn, schema): + try: + conn.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE") + finally: + conn.close() + + +# ---- min/max over timestamptz read no data (#1293) --------------------------------- + +@pytest.fixture(scope="module") +def ts(pgc_cluster): + c = _schema_conn(pgc_cluster, "pgc_agg_ts") + got = {} + try: + c.execute("CREATE TABLE readings (sensor bigint, at timestamptz) USING pgcolumnar") + c.execute("SELECT pgcolumnar.set_options('readings', stripe_row_limit => 2000)") # bigint arithmetic: the shell twin's first draft overflowed int here, the # INSERT failed silently, and every arm measured an empty table. # THE HEAP IS WRITTEN FROM THE GENERATOR, NOT COPIED (#1339). It was a CTAS @@ -65,86 +127,123 @@ def test_min_max_over_timestamptz_read_no_data(pgc_conn, expect): gen = ("SELECT g % 37, '2025-11-15'::timestamptz" f" + ((g::bigint * 104729) % {ROWS}) * interval '2 min'" f" FROM generate_series(1, {ROWS}) g") - cur.execute(f"INSERT INTO readings {gen}") - cur.execute("CREATE TABLE readings_heap (sensor bigint, at timestamptz) USING heap") - cur.execute(f"INSERT INTO readings_heap {gen}") + # the INSERTs too: the shell's psql_run carries on after a failed write and names + # its reds, so a write-path mutant must not error the fixture here + _try(c.execute, f"INSERT INTO readings {gen}") + c.execute("CREATE TABLE readings_heap (sensor bigint, at timestamptz) USING heap") + _try(c.execute, f"INSERT INTO readings_heap {gen}") + got["count"] = _try(_one, c, "SELECT count(*) FROM readings") + got["plan"] = str(_try(_explain, c, "EXPLAIN SELECT max(at) FROM readings")) + got["decode"] = _try(_blocks, c, "readings", "SELECT max(at + interval '0 s') FROM readings") + got["max"] = _try(_blocks, c, "readings", "SELECT max(at) FROM readings") + got["min"] = _try(_blocks, c, "readings", "SELECT min(at) FROM readings") + print(f"-- readings blocks: decode={got['decode']} max={got['max']} min={got['min']}") + + def both(sql): + return (str(_try(_one, c, sql.format(t="readings"))), + str(_try(_one, c, sql.format(t="readings_heap")))) + got["max parity"] = both("SELECT max(at) FROM {t}") + got["min parity"] = both("SELECT min(at) FROM {t}") + # The newest and oldest readings deleted: a zone map still records them. + for t in ("readings", "readings_heap"): + _try(c.execute, f"DELETE FROM {t} WHERE at > (SELECT max(at) FROM readings_heap) - interval '1 hour'" + f" OR at < (SELECT min(at) FROM readings_heap) + interval '45 min'") + got["after delete"] = (_try(_one, c, "SELECT count(*) FROM readings"), + _try(_one, c, "SELECT count(*) FROM readings_heap")) + got["max parity after"] = both("SELECT max(at) FROM {t}") + got["min parity after"] = both("SELECT min(at) FROM {t}") + yield got + finally: + _drop(c, "pgc_agg_ts") + - expect.num(_one(pgc_conn, "SELECT count(*) FROM readings"), ROWS, - "premise: the timestamptz table holds every row inserted") +def test_the_timestamptz_table_holds_every_row(ts, expect): + expect.num(_num(ts["count"], "count"), ROWS, "premise: the timestamptz table holds every row inserted") - with pgc_conn.cursor() as cur: - cur.execute("EXPLAIN SELECT max(at) FROM readings") - plan = "\n".join(r[0] for r in cur.fetchall()) - expect.text("yes" if "Columnar Vectorized Aggregates" in plan else "no", "yes", + +def test_max_ts_plans_the_metadata_agg_node(ts, expect): + expect.text("yes" if "Columnar Vectorized Aggregates" in ts["plan"] else f"no ({ts['plan']})", "yes", "max(ts) over timestamptz plans the metadata agg node") - decode = _blocks(pgc_conn, "readings", "SELECT max(at + interval '0 s') FROM readings") - mx = _blocks(pgc_conn, "readings", "SELECT max(at) FROM readings") - mn = _blocks(pgc_conn, "readings", "SELECT min(at) FROM readings") - print(f"-- readings blocks: decode={decode} max={mx} min={mn}") - expect.at_least(decode, 10, "control: decoding ts reads the table's blocks") +def test_decoding_ts_reads_the_tables_blocks(ts, expect): + expect.at_least(_int(ts["decode"], "decode"), 10, "control: decoding ts reads the table's blocks") + + +@pytest.mark.parametrize("key, name", [ + ("max", "max(ts) reads under a tenth of the blocks a decode of ts reads"), + ("min", "min(ts) reads under a tenth of the blocks a decode of ts reads"), +]) +def test_min_max_ts_read_under_a_tenth_of_a_decode(ts, expect, key, name): # The margin is decode - 10 * reading: positive passes, and a failure # reports by how much it fell short. - expect.at_least(decode - 10 * mx, 1, - "max(ts) reads under a tenth of the blocks a decode of ts reads") - expect.at_least(decode - 10 * mn, 1, - "min(ts) reads under a tenth of the blocks a decode of ts reads") - - def both(sql): - return (str(_one(pgc_conn, sql.format(t="readings"))), - str(_one(pgc_conn, sql.format(t="readings_heap")))) - - got, want = both("SELECT max(at) FROM {t}") - expect.text(got, want, "max(timestamptz) parity") - got, want = both("SELECT min(at) FROM {t}") - expect.text(got, want, "min(timestamptz) parity") - - # The newest and oldest readings deleted: a zone map still records them. - with pgc_conn.cursor() as cur: - for t in ("readings", "readings_heap"): - cur.execute( - f"DELETE FROM {t} WHERE at > (SELECT max(at) FROM readings_heap) - interval '1 hour'" - f" OR at < (SELECT min(at) FROM readings_heap) + interval '45 min'" - ) - n, nh = (_one(pgc_conn, "SELECT count(*) FROM readings"), - _one(pgc_conn, "SELECT count(*) FROM readings_heap")) - expect.text(f"{n < ROWS}|{n}", f"True|{nh}", + reading = _int(ts[key], key) + expect.at_least(_int(ts["decode"], "decode") - 10 * reading if reading else 0, 1, name) + + +@pytest.mark.parametrize("key, name", [ + ("max parity", "max(timestamptz) parity"), + ("min parity", "min(timestamptz) parity"), + ("max parity after", "max(timestamptz) parity after deleting the newest rows"), + ("min parity after", "min(timestamptz) parity after deleting the oldest rows"), +]) +def test_timestamptz_parity(ts, expect, key, name): + expect.text(*ts[key], name) + + +def test_the_deletes_removed_the_same_rows_at_both_ends(ts, expect): + n, nh = ts["after delete"] + # isinstance first, so an error string reds the check instead of raising TypeError + expect.text(f"{isinstance(n, int) and n < ROWS}|{n}", f"True|{nh}", "the deletes removed rows at both ends, the same rows from each") - got, want = both("SELECT max(at) FROM {t}") - expect.text(got, want, "max(timestamptz) parity after deleting the newest rows") - got, want = both("SELECT min(at) FROM {t}") - expect.text(got, want, "min(timestamptz) parity after deleting the oldest rows") +# ---- a WHERE with no column in it is applied (#1435) ------------------------------- + GATED = "SELECT count(*) || '/' || coalesce(min(id)::text, '-') || '/' || coalesce(max(id)::text, '-') FROM" +# a literal list of pairs, not a dict: the twin grader reads a parametrize table, and +# `dict.items()` hides the names from it +GATES = [ + ("(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"), +] -def test_a_where_with_no_column_in_it_is_applied(pgc_conn, expect): +@pytest.fixture(scope="module") +def gate(pgc_cluster): """A WHERE that references no column is a gating qual: planned as a one-time filter, not part of the per-row quals. The ungrouped vectorized aggregate answered from the zone maps alone and ignored it, so a false gate returned every row's aggregates (#1435). Each arm is compared with heap.""" - with pgc_conn.cursor() as cur: - cur.execute("CREATE TABLE gn (id int) USING pgcolumnar") - cur.execute("CREATE TABLE gh (id int) USING heap") - cur.execute("SELECT pgcolumnar.set_options('gn', stripe_row_limit => 1000)") - cur.execute("INSERT INTO gn SELECT g FROM generate_series(1, 5000) g") - cur.execute("INSERT INTO gh SELECT g FROM generate_series(1, 5000) g") - cur.execute("EXPLAIN (COSTS OFF) SELECT count(*), min(id), max(id) FROM gn") - 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") - # 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") + c = _schema_conn(pgc_cluster, "pgc_agg_gate") + try: + c.execute("CREATE TABLE gn (id int) USING pgcolumnar") + c.execute("CREATE TABLE gh (id int) USING heap") + c.execute("SELECT pgcolumnar.set_options('gn', stripe_row_limit => 1000)") + _try(c.execute, "INSERT INTO gn SELECT g FROM generate_series(1, 5000) g") + _try(c.execute, "INSERT INTO gh SELECT g FROM generate_series(1, 5000) g") + got = {"plan": str(_try(_explain, c, "EXPLAIN (COSTS OFF) SELECT count(*), min(id), max(id) FROM gn"))} + for where, _ in GATES: + got[where] = (_try(_one, c, f"{GATED} gn WHERE {where}"), + _try(_one, c, f"{GATED} gh WHERE {where}")) + yield got + finally: + _drop(c, "pgc_agg_gate") + + +def test_count_min_max_with_no_where_are_the_vectorized_aggregate(gate, expect): + expect.text("yes" if "Columnar Vectorized Aggregates" in gate["plan"] else f"no ({gate['plan']})", + "yes", "premise: count, min and max with no WHERE are the vectorized aggregate") + + +@pytest.mark.parametrize("where, name", GATES) +def test_a_where_with_no_column_in_it_gives_heaps_answer(gate, expect, where, name): + expect.text(*gate[where], name) + + +def test_heaps_answer_to_a_false_gate_is_no_rows(gate, expect): + expect.text(gate["(SELECT false)"][1], "0/-/-", "premise: heap's answer to a false gate is no rows") # ---- the int4, text, parallel, zone-map and fallback arms (#1341) ------------------ @@ -158,12 +257,15 @@ def _plan(conn, sql, parallel=False): 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}") + try: + cur.execute("EXPLAIN (COSTS OFF) " + sql) + plan = "\n".join(r[0] for r in cur.fetchall()) + finally: + # in a finally, so a failed EXPLAIN cannot leave the settings on (#1514 review) + 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 @@ -183,80 +285,121 @@ def _zm_scans(conn): 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)") +def _zm_delta(conn, sql): + """Zone-map scans `sql` made: the counter after it, less the counter before. + + The query's own error is recorded and the counter still read, as the shell's + psql_run does, so the twin reports the shell's red set (#1522 review). Both twins + therefore pass "reads no zone maps" on a count(*) that errors; the count(*) + parity arm is what reds then.""" + before = _zm_scans(conn) + _try(_one, conn, sql) + return _zm_scans(conn) - before + + +PARITY = [ + ("count(*)", "SELECT count(*) FROM {t}"), + ("count(col,nulls)", "SELECT count(s) FROM {t}"), + ("sum(int4)", "SELECT sum(id) FROM {t}"), + ("sum(int4,nulls)", "SELECT sum(s) FROM {t}"), + ("avg(int4,nulls)", "SELECT avg(s) FROM {t}"), + ("min(int4)", "SELECT min(id) FROM {t}"), + ("max(int4)", "SELECT max(id) FROM {t}"), + ("min(int4,nulls)", "SELECT min(s) FROM {t}"), + ("min(text)", "SELECT min(label) FROM {t}"), + ("max(text)", "SELECT max(label) FROM {t}"), + ("multi-agg", "SELECT (count(*), sum(id), min(id), max(id))::text FROM {t}"), + ("filtered aggregate parity", "SELECT (count(*), sum(id))::text FROM {t} WHERE id > 6000"), + ("sum(bigint) fallback parity", "SELECT sum(k) FROM {t}"), +] + + +@pytest.fixture(scope="module") +def meta(pgc_cluster): + c = _schema_conn(pgc_cluster, "pgc_agg_meta") + got = {} + try: + c.execute("CREATE TABLE ah (id int, s int, k bigint, label text)") + c.execute("CREATE TABLE an (id int, s int, k bigint, label text) USING pgcolumnar") + c.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") + _try(c.execute, f"INSERT INTO ah {gen}") + _try(c.execute, f"INSERT INTO an {gen}") + # the node first: every parity arm also holds for a correct fallback + got["count node"] = _aggnode(str(_try(_plan, c, "SELECT count(*) FROM an"))) + got["sum node"] = _aggnode(str(_try(_plan, c, "SELECT sum(id), min(id), max(id) FROM an"))) + for name, sql in PARITY: + got[name] = (str(_try(_one, c, sql.format(t="an"))), str(_try(_one, c, sql.format(t="ah")))) + # it has to win on cost, not only exist (#133) + got["par count"] = str(_try(_plan, c, "SELECT count(*) FROM an", parallel=True)) + got["par multi"] = str(_try(_plan, c, "SELECT count(*), sum(id), min(id), max(id) FROM an", + parallel=True)) + # count(*) needs no zone map; an aggregate that needs one must be seen reading it, + # so the zero cannot pass because the counter never moves + got["zm count"] = _try(_zm_delta, c, "SELECT count(*) FROM an") + got["zm sum"] = _try(_zm_delta, c, "SELECT count(*), sum(id) FROM an") + got["filtered node"] = _aggnode(str(_try(_plan, c, "SELECT count(*), sum(id) FROM an WHERE id > 6000"))) + _try(c.execute, "CREATE TABLE ae (id int, label text) USING pgcolumnar") + got["empty count"] = _try(_one, c, "SELECT count(*) FROM ae") + got["empty sum"] = str(_try(_one, c, "SELECT sum(id) IS NULL FROM ae")) + got["empty min"] = str(_try(_one, c, "SELECT min(label) IS NULL FROM ae")) + yield got + finally: + _drop(c, "pgc_agg_meta") + + +@pytest.mark.parametrize("key, name", [ + ("count node", "count(*) uses the metadata agg node"), + ("sum node", "sum/min/max uses the metadata agg node"), +]) +def test_the_aggregate_uses_the_metadata_node(meta, expect, key, name): + expect.text(meta[key], "yes", name) + + +@pytest.mark.parametrize("name, sql", PARITY) +def test_the_aggregate_answers_as_heap(meta, expect, name, sql): + # sql is the fixture's query; read here only so the table names its own rows + expect.text(*meta[name], name) + + +@pytest.mark.parametrize("key, name", [ + ("par count", "count(*) beats a parallel plan"), + ("par multi", "multi-agg beats a parallel plan"), +]) +def test_the_metadata_node_beats_a_parallel_plan(meta, expect, key, name): + expect.text(_aggnode(meta[key]), "yes", name) + + +@pytest.mark.parametrize("key, name", [ + ("par count", "count(*) plan has no Gather"), + ("par multi", "multi-agg plan has no Gather"), +]) +def test_the_metadata_plan_has_no_gather(meta, expect, key, name): + expect.text(_gather(meta[key]), "no", name) + + +def test_count_reads_no_zone_maps(meta, expect): + expect.num(_num(meta["zm count"], "zm count"), 0, "count(*) reads no zone maps") + + +def test_an_aggregate_that_needs_zone_maps_reads_them(meta, expect): + # the measured delta, not a yes/no (#1514 review, item 1) + expect.at_least(_int(meta["zm sum"], "zm sum"), 1, "an aggregate that needs zone maps reads them") + + +def test_a_filtered_aggregate_is_not_the_metadata_node(meta, expect): + expect.text(meta["filtered node"], "no", "filtered aggregate not metadata node") + + +def test_an_empty_table_counts_zero(meta, expect): + expect.num(_num(meta["empty count"], "empty count"), 0, "empty count(*)") + + +@pytest.mark.parametrize("key, name", [ + ("empty sum", "empty sum NULL"), + ("empty min", "empty min NULL"), +]) +def test_an_empty_table_aggregates_to_null(meta, expect, key, name): + expect.text(meta[key], "True", name)