fix(sql): apply join LIMIT after post-join WHERE filters

A hash join with a post-join WHERE predicate truncated its output to the
user LIMIT during the probe, before `filter_and_project` evaluated the
predicate. The first `n` ON-matched rows were emitted and then filtered,
so rows satisfying the WHERE clause that sorted after the first `n`
probe matches were lost — frequently emptying the result entirely.

Real-collection joins push their WHERE into the scanned side, leaving
`post_filters` empty, so the ordering was harmless there. Virtual
catalog joins keep both sides as inline ProviderScans and carry the
WHERE as a join post-filter, which is where the truncation bit: e.g.
`SELECT t.typname FROM pg_type t LEFT JOIN pg_range r ON t.oid =
r.rngtypid WHERE t.typname = 'int4' LIMIT 5` returned 0 rows while the
same query without LIMIT returned the row.

The probe now runs under the byte-budget ceiling whenever post-filters
are present, and the user LIMIT is applied after filtering — so LIMIT
still truncates (narrow LIMIT) but never under-fills.

Tests: pg_catalog_reflection gains catalog join + WHERE + wide/narrow
LIMIT cases; sql_join_correctness gains the real-collection analogues.

Author: mkhairi

nodedb/src/data/executor/handlers/join/hash_handlers.rs

@@ -333,7 +333,17 @@ impl CoreLoop {
         // budget) and, if the probe fills it, surface a deterministic
         // `ResourcesExhausted` rather than dropping the excess rows. A budget
         // of 0 means "unlimited" → truly unbounded output.
-        let (probe_limit, enforce_output_budget) = if join.limit != usize::MAX {
+        //
+        // A user LIMIT may only cap the probe when there are no post-join
+        // WHERE filters: `filter_and_project` runs AFTER the probe, so
+        // truncating to `n` first would emit the first `n` ON-matched rows and
+        // then discard those failing the WHERE clause — under-filling (or
+        // emptying) the result even though later probe rows match. With
+        // post-filters present the probe runs under the budget ceiling and the
+        // user LIMIT is applied after filtering, below.
+        let has_post_filters = !join.post_filter_bytes.is_empty();
+        let (probe_limit, enforce_output_budget) = if join.limit != usize::MAX && !has_post_filters
+        {
             (join.limit, false)
         } else if budget == 0 {
             (usize::MAX, false)
@@ -374,6 +384,13 @@ impl CoreLoop {
             );
         }
 
+        // Deferred user LIMIT: when post-join WHERE filters exist the probe
+        // ran unbounded (see above) so the LIMIT must be applied here, after
+        // the filters have retained the matching rows.
+        if has_post_filters && join.limit != usize::MAX {
+            results.truncate(join.limit);
+        }
+
         let payload = super::super::super::response_codec::encode_binary_rows(&results);
         self.response_with_payload(join.task, payload)
     }

nodedb/tests/pg_catalog_reflection.rs

@@ -548,6 +548,86 @@ async fn cross_vtable_join_filters_on_joined_column() {
     );
 }
 
+/// A JOIN carrying BOTH a WHERE predicate and a LIMIT must still return the
+/// matching rows. The WHERE clause on a catalog join is evaluated after the
+/// join (there is no scan to push it into), so a LIMIT applied during the
+/// probe would cap the output before filtering and drop — often empty — the
+/// rows that actually match.
+#[tokio::test]
+async fn cross_vtable_join_with_where_and_limit_returns_matching_rows() {
+    let srv = TestServer::start().await;
+    srv.exec("CREATE COLLECTION reflect_join_where_limit (id INTEGER PRIMARY KEY)")
+        .await
+        .expect("create collection");
+
+    let unlimited = srv
+        .query_text(
+            "SELECT t.typname FROM pg_type t \
+             LEFT JOIN pg_range r ON t.oid = r.rngtypid \
+             WHERE t.typname = 'int4'",
+        )
+        .await
+        .expect("join with WHERE evaluates");
+    assert_eq!(
+        unlimited,
+        vec!["int4".to_string()],
+        "baseline: the unlimited join must return the matching type"
+    );
+
+    let limited = srv
+        .query_text(
+            "SELECT t.typname FROM pg_type t \
+             LEFT JOIN pg_range r ON t.oid = r.rngtypid \
+             WHERE t.typname = 'int4' LIMIT 5",
+        )
+        .await
+        .expect("join with WHERE and LIMIT evaluates");
+    assert_eq!(
+        limited, unlimited,
+        "a LIMIT wider than the result must not drop rows that satisfy the WHERE clause"
+    );
+}
+
+/// A LIMIT narrower than the filtered result set still truncates — the LIMIT
+/// moves after the post-join filter, it is not discarded.
+#[tokio::test]
+async fn cross_vtable_join_with_where_honors_narrow_limit() {
+    let srv = TestServer::start().await;
+    srv.exec("CREATE COLLECTION reflect_narrow_limit_a (id INTEGER PRIMARY KEY)")
+        .await
+        .expect("create collection");
+    srv.exec("CREATE COLLECTION reflect_narrow_limit_b (id INTEGER PRIMARY KEY)")
+        .await
+        .expect("create collection");
+
+    let unlimited = srv
+        .query_text(
+            "SELECT c.relname FROM pg_class c \
+             JOIN pg_namespace n ON n.oid = c.relnamespace \
+             WHERE n.nspname = 'public'",
+        )
+        .await
+        .expect("join with WHERE evaluates");
+    assert!(
+        unlimited.len() > 1,
+        "test needs a multi-row baseline to check truncation, got {unlimited:?}"
+    );
+
+    let limited = srv
+        .query_text(
+            "SELECT c.relname FROM pg_class c \
+             JOIN pg_namespace n ON n.oid = c.relnamespace \
+             WHERE n.nspname = 'public' LIMIT 1",
+        )
+        .await
+        .expect("join with WHERE and LIMIT evaluates");
+    assert_eq!(
+        limited.len(),
+        1,
+        "LIMIT 1 must truncate the filtered rows to one: {limited:?}"
+    );
+}
+
 /// The three-way `pg_class ⋈ pg_attribute ⋈ pg_type` join is the literal
 /// shape `\d <table>` emits to describe a relation's columns and their types.
 #[tokio::test]

nodedb/tests/sql_join_correctness.rs

@@ -338,3 +338,60 @@ async fn natural_join_is_not_cartesian() {
         }
     }
 }
+
+// ---------------------------------------------------------------------------
+// LIMIT with post-join filters
+// ---------------------------------------------------------------------------
+
+/// A join carrying both a post-join WHERE predicate and a LIMIT must return
+/// the matching rows: the LIMIT applies to the filtered result, never to the
+/// pre-filter probe output.
+#[tokio::test]
+async fn join_with_post_filter_and_wide_limit_keeps_matching_rows() {
+    let server = TestServer::start().await;
+    setup_join_tables(&server).await;
+
+    let unlimited = server
+        .query_text(
+            "SELECT j_t2.id FROM j_t1 \
+             JOIN j_t2 ON j_t2.t1_id = j_t1.id \
+             WHERE j_t2.y = 200",
+        )
+        .await
+        .expect("join with WHERE evaluates");
+    assert_eq!(unlimited, vec!["q".to_string()]);
+
+    let limited = server
+        .query_text(
+            "SELECT j_t2.id FROM j_t1 \
+             JOIN j_t2 ON j_t2.t1_id = j_t1.id \
+             WHERE j_t2.y = 200 LIMIT 5",
+        )
+        .await
+        .expect("join with WHERE and LIMIT evaluates");
+    assert_eq!(
+        limited, unlimited,
+        "a LIMIT wider than the filtered result must not drop matching rows"
+    );
+}
+
+/// A LIMIT narrower than the filtered result still truncates.
+#[tokio::test]
+async fn join_with_post_filter_honors_narrow_limit() {
+    let server = TestServer::start().await;
+    setup_join_tables(&server).await;
+
+    let rows = server
+        .query_text(
+            "SELECT j_t2.id FROM j_t1 \
+             JOIN j_t2 ON j_t2.t1_id = j_t1.id \
+             WHERE j_t2.y > 0 LIMIT 1",
+        )
+        .await
+        .expect("join with WHERE and LIMIT evaluates");
+    assert_eq!(
+        rows.len(),
+        1,
+        "LIMIT 1 must truncate the filtered rows to one: {rows:?}"
+    );
+}