test: the runtime-filter pytest twin is paired with its shell suite by name and covers every arm (#1341)

test/check_ledger.tsv

@@ -2825,9 +2825,9 @@ native_join_runtime_filter	native_join_runtime_filter	plan has build tap	15;16;1
 native_join_runtime_filter	native_join_runtime_filter	plan has runtime coordinator	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	plan retains core Hash Join	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	premise: and its build is empty	15;16;17;18;19	never	-
-native_join_runtime_filter	native_join_runtime_filter	premise: heap's answer under a true gate over a columnar dimension holds 200 rows	15;16;17;18;19	never	-
-native_join_runtime_filter	native_join_runtime_filter	premise: heap's answer under a true gate over a heap dimension holds 200 rows	15;16;17;18;19	never	-
-native_join_runtime_filter	native_join_runtime_filter	premise: heap's empty-then-full rescan answer holds the 200 keys	15;16;17;18;19	never	-
+native_join_runtime_filter	native_join_runtime_filter	premise: heap's answer under a true gate over a columnar dimension holds every joined row	15;16;17;18;19	never	-
+native_join_runtime_filter	native_join_runtime_filter	premise: heap's answer under a true gate over a heap dimension holds every joined row	15;16;17;18;19	never	-
+native_join_runtime_filter	native_join_runtime_filter	premise: heap's empty-then-full rescan answer holds every dimension key	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	premise: the empty-then-full rescan runs under the coordinator	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	premise: the fact qual removes matching rows on heap	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	premise: the large-dimension empty build runs under the coordinator	15;16;17;18;19	never	-

test/native_join_runtime_filter.sh

@@ -181,7 +181,7 @@ SQLRE="${SQLR//(VALUES (0),(8000))/(VALUES (99999),(0))}"
 SQLRHE="${SQLRH//(VALUES (0),(8000))/(VALUES (99999),(0))}"
 check "premise: the empty-then-full rescan runs under the coordinator" \
 	"$(pc "${rf_on}SET max_parallel_workers_per_gather=0;EXPLAIN (COSTS OFF) $SQLRE" | grep -c 'Columnar Runtime Filter Coordinator')" 1
-check "premise: heap's empty-then-full rescan answer holds the 200 keys" "$(q "$SQLRHE" | head -1)" "0|200"
+check "premise: heap's empty-then-full rescan answer holds every dimension key" "$(q "$SQLRHE" | head -1)" "0|200"
 check "an empty build on one rescan does not empty the next" \
 	"$(pc "${rf_on}SET max_parallel_workers_per_gather=0;$SQLRE" | grep -v '^SET$')" "$(q "$SQLRHE")"
 
@@ -377,7 +377,7 @@ for dim in dim_gh dim_gc; do
 		"$(grep -c 'Columnar Runtime Filter Coordinator' <<<"$ong")" 1
 	# the reference must hold rows, or a server that answered nothing on both
 	# sides would pass the comparisons below as equal
-	check "premise: heap's answer under a true gate over a $kind dimension holds 200 rows" \
+	check "premise: heap's answer under a true gate over a $kind dimension holds every joined row" \
 		"$(q "SELECT count(*) FROM heap_g JOIN $dim ON heap_g.k=$dim.k WHERE (SELECT true)")" 200
 	for gate in false true; do
 		check "a (SELECT $gate) gate over a $kind dimension equals heap" \

test/pytest/TESTS.md

@@ -70,7 +70,7 @@ behaviour, the source of that number is named.
 - [22. test_writes_wrote_rows.py: a write that wrote nothing](#22-test_writes_wrote_rowspy-a-write-that-wrote-nothing)
 - [23. test_mutation_ledger.py: which checks have ever been red](#23-test_mutation_ledgerpy-which-checks-have-ever-been-red)
 - [24. test_loop_coverage_premise.py: a loop that never ran asserted nothing](#24-test_loop_coverage_premisepy-a-loop-that-never-ran-asserted-nothing)
-- [25. test_join_runtime_filter.py: serial join runtime filter](#25-test_join_runtime_filterpy-serial-join-runtime-filter)
+- [25. test_native_join_runtime_filter.py: serial join runtime filter](#25-test_native_join_runtime_filterpy-serial-join-runtime-filter)
 - [26. test_check_records.py: every counted assertion is a record](#26-test_check_recordspy-every-counted-assertion-is-a-record)
 - [27. test_skip_loop_arms.py: a skipped arm records under its own name](#27-test_skip_loop_armspy-a-skipped-arm-records-under-its-own-name)
 - [28. test_docs_join_clustering.py: the runtime filter's layout precondition](#28-test_docs_join_clusteringpy-the-runtime-filters-layout-precondition)
@@ -3004,7 +3004,7 @@ is how a guard gets switched off. **So the residual is a loop whose premise boun
 wrong collection**, which a reviewer catches and a sweep does not. 3.5 names it.
 
 
-## 25. test_join_runtime_filter.py: serial join runtime filter
+## 25. test_native_join_runtime_filter.py: serial join runtime filter
 
 Pytest twin of `test/native_join_runtime_filter.sh`. The two files are independent:
 each builds its own fixtures and expected values. They share only the public
@@ -3091,6 +3091,11 @@ A LATERAL whose first outer row builds nothing and whose second builds every key
 execution would answer the second with no rows. Premises check the coordinator and
 that heap's answer holds the keys.
 
+### `test_a_fact_qual_conjunction_with_late_materialization_matches_heap`
+
+The shell's fact-qual arms on the default path, late materialization on: a conjunction that also names a non-key column (`k > 100 AND n % 3 <> 0`) must still return heap's rows under the runtime filter coordinator. A premise first shows the qual removes matching rows on heap (100 of 150), so an answer that ignored it would not equal heap's. The late-mat-off test above covers the same shape with late materialization off. Added when this file was renamed to pair with `native_join_runtime_filter.sh` and aligned to the shell's names (#1341).
+
+
 ## 26. test_check_records.py: every counted assertion is a record
 
 #937, first phase. The shell harness makes counting and recording the same call,

test/pytest/expected_tests.txt

@@ -815,4 +815,6 @@ guard_tests 439
 # three tests, by collection over the cluster files on this tree.
 # cluster_tests 747 -> 749 for #1341's test_native_ios.py, the twin of native_ios.sh:
 # two tests, by collection over the cluster files on this tree.
-cluster_tests 749
+# 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

test/pytest/test_compare_to_bash.py

@@ -140,6 +140,7 @@
             "native_index_fetch_stripe_cost",
             "native_index_projection",
             "native_ios",
+            "native_join_runtime_filter",
             "native_join_vector_agg",
             "native_ownership",
             "native_parquet_dict_oob",
@@ -421,8 +422,9 @@ 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, join_runtime_filter, native_groupagg_batch, "
-                "native_index_projection, native_parquet_pushdown, native_reclaim, native_repack, "
+                "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, "
                 "sorted_projection",
                 "and it is these files, so another appearing is a diff a reviewer "

test/pytest/test_join_runtime_filter.py → test/pytest/test_native_join_runtime_filter.py

@@ -106,16 +106,17 @@ def test_serial_join_runtime_filter(pgc_conn, expect):
         h = c.fetchone()
     expect.rows([a], [o], "runtime answer equals off")
     expect.rows([a], [h], "runtime answer equals heap")
-    for name, s in [
-        ("LEFT refusal", "SELECT count(*)FROM f LEFT JOIN d ON f.k=d.k"),
-        ("SEMI refusal", "SELECT count(*)FROM f WHERE EXISTS(SELECT 1 FROM d WHERE d.k=f.k)"),
-        ("ANTI refusal", "SELECT count(*)FROM f WHERE NOT EXISTS(SELECT 1 FROM d WHERE d.k=f.k)"),
-        ("CROSS refusal", "SELECT count(*)FROM f CROSS JOIN d"),
+    # the shell's "$n refusal", built the same way so the two templates meet
+    for n, s in [
+        ("LEFT", "SELECT count(*)FROM f LEFT JOIN d ON f.k=d.k"),
+        ("SEMI", "SELECT count(*)FROM f WHERE EXISTS(SELECT 1 FROM d WHERE d.k=f.k)"),
+        ("ANTI", "SELECT count(*)FROM f WHERE NOT EXISTS(SELECT 1 FROM d WHERE d.k=f.k)"),
+        ("CROSS", "SELECT count(*)FROM f CROSS JOIN d"),
     ]:
         expect.num(
             _plan(pgc_conn, s, True).count("Columnar Runtime Filter Coordinator"),
             0,
-            name,
+            f"{n} refusal",
         )
 
 
@@ -183,12 +184,12 @@ def test_scattered_join_runtime_bloom(pgc_conn, expect):
     expect.plan_node(
         plan,
         provider="Columnar Runtime Filter Coordinator",
-        name="scattered plan uses the runtime coordinator",
+        name="scattered plan has coordinator",
     )
     expect.num(
         _json_field(plan, "Runtime Filter Groups Removed"),
         0,
-        "scattered hull cannot drop a group",
+        "scattered interval cannot drop groups",
     )
     expect.num(
         _json_field(plan, "Columnar Chunk Groups Read"),
@@ -255,7 +256,7 @@ def test_cross_type_int4_int8_bloom(pgc_conn, expect):
     expect.plan_node(
         plan,
         provider="Columnar Runtime Filter Coordinator",
-        name="int4/int8 plan uses the runtime coordinator",
+        name="int4/int8 plan has coordinator",
     )
     expect.num(
         _removed_or_zero(plan),
@@ -325,7 +326,7 @@ def test_collation_mismatch_bloom_only(pgc_conn, expect):
     expect.plan_node(
         plan,
         provider="Columnar Runtime Filter Coordinator",
-        name="collation-mismatch plan uses the runtime coordinator",
+        name="collation-mismatch plan has coordinator",
     )
     expect.at_least(
         _removed_or_zero(matched_plan),
@@ -399,7 +400,7 @@ def test_runtime_filter_rebuilds_on_lateral_rescan(pgc_conn, expect):
         provider="Columnar Runtime Filter Coordinator",
         name="lateral rescan still uses the runtime coordinator",
     )
-    expect.rows(got, heap, "lateral rescan answers equal heap")
+    expect.rows(got, heap, "rescan answers equal heap")
 
 
 def test_saturated_build_disables_bloom(pgc_conn, expect):
@@ -443,7 +444,7 @@ def test_saturated_build_disables_bloom(pgc_conn, expect):
     expect.plan_node(
         plan,
         provider="Columnar Runtime Filter Coordinator",
-        name="saturated plan uses the runtime coordinator",
+        name="saturated plan has coordinator",
     )
     bloom = _json_field(plan, "Runtime Filter Bloom")
     expect.num(
@@ -504,14 +505,14 @@ def test_three_table_join_order_unchanged(pgc_conn, expect):
     expect.plan_node(
         on_plan,
         provider="Columnar Runtime Filter Coordinator",
-        name="3-table plan uses the runtime coordinator",
+        name="3-table plan has coordinator",
     )
     expect.rows(
         _relation_names(on_plan),
         _relation_names(off_plan),
-        "3-table relation order matches filter-off",
+        "3-table join order matches filter-off",
     )
-    expect.rows([on_rows], [off_rows], "3-table answer matches filter-off")
+    expect.rows([on_rows], [off_rows], "3-table answer equals filter-off")
 
 
 def test_empty_build_runtime_filter(pgc_conn, expect):
@@ -555,7 +556,7 @@ def test_empty_build_runtime_filter(pgc_conn, expect):
     expect.plan_node(
         plan,
         provider="Columnar Runtime Filter Coordinator",
-        name="empty-build plan uses the runtime coordinator",
+        name="empty-build plan has coordinator",
     )
     expect.rows([got], [heap], "empty-build answer equals heap")
 
@@ -667,7 +668,7 @@ def test_early_limit_matches_heap(pgc_conn, expect):
     expect.plan_node(
         plan,
         provider="Columnar Runtime Filter Coordinator",
-        name="early limit still uses the runtime coordinator",
+        name="early limit has coordinator",
     )
     expect.rows(got, heap, "early limit equals heap")
 
@@ -728,6 +729,68 @@ def test_fact_qual_with_late_mat_off_matches_heap(pgc_conn, expect):
     expect.rows([got], [heap], "late-mat-off fact qual equals heap")
 
 
+def test_a_fact_qual_conjunction_with_late_materialization_matches_heap(pgc_conn, expect):
+    """The shell's fact-qual arms, on the default path: late materialization ON.
+
+    The coordinator used to mark only the join key, so a conjunction that also named a
+    non-key column dropped every surviving row. The arm above covers it with late
+    materialization off; this is the default configuration the shell suite runs.
+    """
+    with pgc_conn.cursor() as cur:
+        cur.execute(
+            """
+            CREATE TABLE dimc(k int);
+            INSERT INTO dimc SELECT g FROM generate_series(7300, 7449) g;
+            CREATE TABLE factc(k int, n int, payload text) USING pgcolumnar;
+            SELECT pgcolumnar.set_options($t$factc$t$, stripe_row_limit => 1000);
+            INSERT INTO factc SELECT g, g, repeat(md5(g::text), 3)
+            FROM generate_series(1, 16000) g;
+            CREATE TABLE heapc AS SELECT * FROM factc;
+            ANALYZE dimc;
+            ANALYZE factc
+            """
+        )
+        if _has_runtime_filter(pgc_conn):
+            cur.execute("SET pgcolumnar.enable_join_runtime_filter=on")
+        cur.execute("SET max_parallel_workers_per_gather=0")
+        cur.execute("SET enable_nestloop=off")
+        cur.execute("SET enable_mergejoin=off")
+        sql = """
+            SELECT count(*), sum(factc.k), sum(factc.n)
+            FROM factc JOIN dimc ON factc.k = dimc.k
+            WHERE factc.k > 100 AND factc.n % 3 <> 0
+        """
+        cur.execute(
+            "EXPLAIN (ANALYZE, FORMAT JSON, COSTS OFF, TIMING OFF, SUMMARY OFF) " + sql
+        )
+        plan = cur.fetchone()[0]
+        cur.execute(sql)
+        got = cur.fetchone()
+        cur.execute(
+            """
+            SELECT count(*), sum(heapc.k), sum(heapc.n)
+            FROM heapc JOIN dimc ON heapc.k = dimc.k
+            WHERE heapc.k > 100 AND heapc.n % 3 <> 0
+            """
+        )
+        heap = cur.fetchone()
+        cur.execute("SELECT count(*) FROM heapc JOIN dimc ON heapc.k = dimc.k "
+                    "WHERE heapc.k > 100 AND heapc.n % 3 <> 0")
+        kept = cur.fetchone()[0]
+        cur.execute("SELECT count(*) FROM heapc JOIN dimc ON heapc.k = dimc.k")
+        matched = cur.fetchone()[0]
+
+    # the conjunction must remove some matching rows, or ignoring it equals heap's answer
+    expect.text(f"{kept}|{matched}", "100|150",
+                "premise: with late materialization on, the fact qual removes matching rows")
+    expect.plan_node(
+        plan,
+        provider="Columnar Runtime Filter Coordinator",
+        name="fact-qual plan has coordinator",
+    )
+    expect.rows([got], [heap], "fact-qual conjunction equals heap")
+
+
 def test_a_gating_qual_over_the_join_matches_heap(pgc_conn, expect):
     """A WHERE that names no column puts a gating Result over the join (#1437).
 
@@ -765,7 +828,7 @@ def test_a_gating_qual_over_the_join_matches_heap(pgc_conn, expect):
         with pgc_conn.cursor() as cur:
             cur.execute(f"SELECT count(*) FROM hg JOIN {dim} ON hg.k = {dim}.k WHERE (SELECT true)")
             expect.num(cur.fetchone()[0], 150,
-                       f"premise: heap's answer under a true gate over a {kind} dimension holds 150 rows")
+                       f"premise: heap's answer under a true gate over a {kind} dimension holds every joined row")
         for gate in ("false", "true"):
             with pgc_conn.cursor() as cur:
                 cur.execute(f"{sql} WHERE (SELECT {gate})")
@@ -855,5 +918,5 @@ def test_an_empty_build_on_one_rescan_does_not_empty_the_next(pgc_conn, expect):
     expect.num(plan.count("Columnar Runtime Filter Coordinator"), 1,
                "premise: the empty-then-full rescan runs under the coordinator")
     expect.text(f"{heap[0][0]}|{heap[0][1]}", "0|150",
-                "premise: heap's empty-then-full rescan answer holds the 150 keys")
+                "premise: heap's empty-then-full rescan answer holds every dimension key")
     expect.rows(got, heap, "an empty build on one rescan does not empty the next")

test/pytest/unported_suites.txt

@@ -90,7 +90,6 @@ native_fold_deferral
 native_fold_skipguard
 native_gap
 native_groupagg_wide_cost
-native_join_runtime_filter
 native_late_materialization
 native_lazy_slot
 native_metadata_flush