Skip to content

Commit 8c2e5c1

Browse files
coderfendercoderfender
authored
Merge branch 'main' into experiment_roaring_bitmap_for_int32_anti_semi_joins
2 parents 630337e + b382ecd commit 8c2e5c1

22 files changed

Lines changed: 1411 additions & 401 deletions

File tree

‎.ai/skills/datafusion-ffi/SKILL.md‎

Lines changed: 360 additions & 0 deletions
Large diffs are not rendered by default.

‎AGENTS.md‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,3 +39,10 @@ When creating a PR, you MUST follow the [PR template](.github/pull_request_templ
3939

4040
See the [Testing Quick Start](docs/source/contributor-guide/testing.md#testing-quick-start)
4141
for the recommended pre-PR test commands.
42+
43+
## Agent Skills
44+
45+
Repository-specific agent skills live under `.ai/skills/`. Each subdirectory is
46+
a single skill with a `SKILL.md` (YAML frontmatter + body). Check that
47+
directory for applicable skills before working on a task; new skills go in
48+
`.ai/skills/<skill-name>/SKILL.md`.

‎benchmarks/src/hj.rs‎

Lines changed: 67 additions & 155 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,6 @@ use std::path::PathBuf;
2525

2626
use futures::StreamExt;
2727

28-
// TODO: Add existence joins
29-
3028
/// Run the Hash Join benchmark
3129
///
3230
/// This micro-benchmark focuses on the performance characteristics of Hash Joins.
@@ -304,193 +302,105 @@ const HASH_QUERIES: &[HashJoinQuery] = &[
304302
probe_size: "60M",
305303
},
306304
// RightSemi Join benchmarks with Int32 keys
307-
// Q16: RightSemi, 100% Density, 100% Hit rate
305+
//
306+
// Fanout (average build rows matched per probe row, as measured by running
307+
// the equivalent INNER JOIN under `EXPLAIN ANALYZE` and reading the
308+
// `HashJoinExec` metrics): 1 for Q16-Q18. Build keys here are primary
309+
// keys (`n_nationkey`, `s_suppkey`), so each probe row matches at most
310+
// one build row. `prob_hit` controls what fraction of probe rows find
311+
// that one match.
312+
//
313+
// Fanout still matters because semi joins short-circuit after the first
314+
// match. Coverage of fanout > 1 (build-side duplicates) is left for a
315+
// follow-up.
316+
//
317+
// Q16: RightSemi, Small build (25 rows), 100% Hit rate
318+
// Build Side: nation (25 rows) | Probe Side: customer (1.5M rows)
308319
HashJoinQuery {
309-
sql: r###"SELECT l.k
310-
FROM (
311-
SELECT CAST(l_suppkey AS INT) as k FROM lineitem
312-
) l
313-
WHERE EXISTS (
314-
SELECT 1 FROM (SELECT CAST(s_suppkey AS INT) as k FROM supplier) s WHERE s.k = l.k
315-
)"###,
320+
sql: r###"SELECT c.k
321+
FROM (SELECT CAST(n_nationkey AS INT) as k FROM nation) n
322+
RIGHT SEMI JOIN (SELECT CAST(c_nationkey AS INT) as k FROM customer) c
323+
ON n.k = c.k"###,
316324
density: 1.0,
317325
prob_hit: 1.0,
318-
build_size: "100K",
319-
probe_size: "60M_RightSemi",
326+
build_size: "25",
327+
probe_size: "1.5M_RightSemi",
320328
},
321-
// Q17: RightSemi, 100% Density, 10% Hit rate
329+
// Q17: RightSemi, Medium build (100K rows), 100% Hit rate
330+
// Build Side: supplier (100K rows) | Probe Side: lineitem (60M rows)
322331
HashJoinQuery {
323332
sql: r###"SELECT l.k
324-
FROM (
325-
SELECT CAST(CASE WHEN l_suppkey % 10 = 0 THEN l_suppkey ELSE l_suppkey + 1000000 END AS INT) as k
326-
FROM lineitem
327-
) l
328-
WHERE EXISTS (
329-
SELECT 1 FROM (SELECT CAST(s_suppkey AS INT) as k FROM supplier) s WHERE s.k = l.k
330-
)"###,
333+
FROM (SELECT CAST(s_suppkey AS INT) as k FROM supplier) s
334+
RIGHT SEMI JOIN (SELECT CAST(l_suppkey AS INT) as k FROM lineitem) l
335+
ON s.k = l.k"###,
331336
density: 1.0,
332-
prob_hit: 0.1,
333-
build_size: "100K",
334-
probe_size: "60M_RightSemi",
335-
},
336-
// Q18: RightSemi, 50% Density, 100% Hit rate
337-
HashJoinQuery {
338-
sql: r###"SELECT l.k
339-
FROM (
340-
SELECT CAST(l_suppkey * 2 AS INT) as k FROM lineitem
341-
) l
342-
WHERE EXISTS (
343-
SELECT 1 FROM (SELECT CAST(s_suppkey * 2 AS INT) as k FROM supplier) s WHERE s.k = l.k
344-
)"###,
345-
density: 0.5,
346-
prob_hit: 1.0,
347-
build_size: "100K",
348-
probe_size: "60M_RightSemi",
349-
},
350-
// Q19: RightSemi, 50% Density, 10% Hit rate
351-
HashJoinQuery {
352-
sql: r###"SELECT l.k
353-
FROM (
354-
SELECT CAST(CASE
355-
WHEN l_suppkey % 10 = 0 THEN l_suppkey * 2
356-
WHEN l_suppkey % 10 < 9 THEN l_suppkey * 2 + 1
357-
ELSE l_suppkey * 2 + 1000000
358-
END AS INT) as k
359-
FROM lineitem
360-
) l
361-
WHERE EXISTS (
362-
SELECT 1 FROM (SELECT CAST(s_suppkey * 2 AS INT) as k FROM supplier) s WHERE s.k = l.k
363-
)"###,
364-
density: 0.5,
365-
prob_hit: 0.1,
366-
build_size: "100K",
367-
probe_size: "60M_RightSemi",
368-
},
369-
// Q20: RightSemi, 10% Density, 100% Hit rate
370-
HashJoinQuery {
371-
sql: r###"SELECT l.k
372-
FROM (
373-
SELECT CAST(l_suppkey * 10 AS INT) as k FROM lineitem
374-
) l
375-
WHERE EXISTS (
376-
SELECT 1 FROM (SELECT CAST(s_suppkey * 10 AS INT) as k FROM supplier) s WHERE s.k = l.k
377-
)"###,
378-
density: 0.1,
379337
prob_hit: 1.0,
380338
build_size: "100K",
381339
probe_size: "60M_RightSemi",
382340
},
383-
// Q21: RightSemi, 10% Density, 10% Hit rate
341+
// Q18: RightSemi, Medium build (100K rows), 10% Hit rate
342+
// Build Side: supplier (100K rows) | Probe Side: lineitem (60M rows)
384343
HashJoinQuery {
385344
sql: r###"SELECT l.k
386-
FROM (
387-
SELECT CAST(CASE
388-
WHEN l_suppkey % 10 = 0 THEN l_suppkey * 10
389-
WHEN l_suppkey % 10 < 9 THEN l_suppkey * 10 + 1
390-
ELSE l_suppkey * 10 + 1000000
391-
END AS INT) as k
345+
FROM (SELECT CAST(s_suppkey AS INT) as k FROM supplier) s
346+
RIGHT SEMI JOIN (
347+
SELECT CAST(CASE WHEN l_suppkey % 10 = 0 THEN l_suppkey ELSE l_suppkey + 1000000 END AS INT) as k
392348
FROM lineitem
393349
) l
394-
WHERE EXISTS (
395-
SELECT 1 FROM (SELECT CAST(s_suppkey * 10 AS INT) as k FROM supplier) s WHERE s.k = l.k
396-
)"###,
397-
density: 0.1,
350+
ON s.k = l.k"###,
351+
density: 1.0,
398352
prob_hit: 0.1,
399353
build_size: "100K",
400354
probe_size: "60M_RightSemi",
401355
},
402356
// RightAnti Join benchmarks with Int32 keys
403-
// Q22: RightAnti, 100% Density, 100% Hit rate (no output)
357+
//
358+
// Fanout (average build rows matched per probe row, as measured by running
359+
// the equivalent INNER JOIN under `EXPLAIN ANALYZE` and reading the
360+
// `HashJoinExec` metrics): 1 for Q19-Q21. Build keys here are primary
361+
// keys (`n_nationkey`, `s_suppkey`), so each probe row matches at most
362+
// one build row. `prob_hit` controls what fraction of probe rows find
363+
// that one match (and are therefore filtered *out* by anti).
364+
//
365+
// Fanout still matters because anti joins short-circuit after the first
366+
// match. Coverage of fanout > 1 (build-side duplicates) is left for a
367+
// follow-up.
368+
//
369+
// Q19: RightAnti, Small build (25 rows), 100% Hit rate (no output)
370+
// Build Side: nation (25 rows) | Probe Side: customer (1.5M rows)
404371
HashJoinQuery {
405-
sql: r###"SELECT l.k
406-
FROM (
407-
SELECT CAST(l_suppkey AS INT) as k FROM lineitem
408-
) l
409-
WHERE NOT EXISTS (
410-
SELECT 1 FROM (SELECT CAST(s_suppkey AS INT) as k FROM supplier) s WHERE s.k = l.k
411-
)"###,
372+
sql: r###"SELECT c.k
373+
FROM (SELECT CAST(n_nationkey AS INT) as k FROM nation) n
374+
RIGHT ANTI JOIN (SELECT CAST(c_nationkey AS INT) as k FROM customer) c
375+
ON n.k = c.k"###,
412376
density: 1.0,
413377
prob_hit: 1.0,
414-
build_size: "100K",
415-
probe_size: "60M_RightAnti",
378+
build_size: "25",
379+
probe_size: "1.5M_RightAnti",
416380
},
417-
// Q23: RightAnti, 100% Density, 10% Hit rate (90% output)
381+
// Q20: RightAnti, Medium build (100K rows), 100% Hit rate (no output)
382+
// Build Side: supplier (100K rows) | Probe Side: lineitem (60M rows)
418383
HashJoinQuery {
419384
sql: r###"SELECT l.k
420-
FROM (
421-
SELECT CAST(CASE WHEN l_suppkey % 10 = 0 THEN l_suppkey ELSE l_suppkey + 1000000 END AS INT) as k
422-
FROM lineitem
423-
) l
424-
WHERE NOT EXISTS (
425-
SELECT 1 FROM (SELECT CAST(s_suppkey AS INT) as k FROM supplier) s WHERE s.k = l.k
426-
)"###,
385+
FROM (SELECT CAST(s_suppkey AS INT) as k FROM supplier) s
386+
RIGHT ANTI JOIN (SELECT CAST(l_suppkey AS INT) as k FROM lineitem) l
387+
ON s.k = l.k"###,
427388
density: 1.0,
428-
prob_hit: 0.1,
429-
build_size: "100K",
430-
probe_size: "60M_RightAnti",
431-
},
432-
// Q24: RightAnti, 50% Density, 100% Hit rate (no output)
433-
HashJoinQuery {
434-
sql: r###"SELECT l.k
435-
FROM (
436-
SELECT CAST(l_suppkey * 2 AS INT) as k FROM lineitem
437-
) l
438-
WHERE NOT EXISTS (
439-
SELECT 1 FROM (SELECT CAST(s_suppkey * 2 AS INT) as k FROM supplier) s WHERE s.k = l.k
440-
)"###,
441-
density: 0.5,
442389
prob_hit: 1.0,
443390
build_size: "100K",
444391
probe_size: "60M_RightAnti",
445392
},
446-
// Q25: RightAnti, 50% Density, 10% Hit rate (90% output)
393+
// Q21: RightAnti, Medium build (100K rows), 10% Hit rate (90% output)
394+
// Build Side: supplier (100K rows) | Probe Side: lineitem (60M rows)
447395
HashJoinQuery {
448396
sql: r###"SELECT l.k
449-
FROM (
450-
SELECT CAST(CASE
451-
WHEN l_suppkey % 10 = 0 THEN l_suppkey * 2
452-
WHEN l_suppkey % 10 < 9 THEN l_suppkey * 2 + 1
453-
ELSE l_suppkey * 2 + 1000000
454-
END AS INT) as k
455-
FROM lineitem
456-
) l
457-
WHERE NOT EXISTS (
458-
SELECT 1 FROM (SELECT CAST(s_suppkey * 2 AS INT) as k FROM supplier) s WHERE s.k = l.k
459-
)"###,
460-
density: 0.5,
461-
prob_hit: 0.1,
462-
build_size: "100K",
463-
probe_size: "60M_RightAnti",
464-
},
465-
// Q26: RightAnti, 10% Density, 100% Hit rate (no output)
466-
HashJoinQuery {
467-
sql: r###"SELECT l.k
468-
FROM (
469-
SELECT CAST(l_suppkey * 10 AS INT) as k FROM lineitem
470-
) l
471-
WHERE NOT EXISTS (
472-
SELECT 1 FROM (SELECT CAST(s_suppkey * 10 AS INT) as k FROM supplier) s WHERE s.k = l.k
473-
)"###,
474-
density: 0.1,
475-
prob_hit: 1.0,
476-
build_size: "100K",
477-
probe_size: "60M_RightAnti",
478-
},
479-
// Q27: RightAnti, 10% Density, 10% Hit rate (90% output)
480-
HashJoinQuery {
481-
sql: r###"SELECT l.k
482-
FROM (
483-
SELECT CAST(CASE
484-
WHEN l_suppkey % 10 = 0 THEN l_suppkey * 10
485-
WHEN l_suppkey % 10 < 9 THEN l_suppkey * 10 + 1
486-
ELSE l_suppkey * 10 + 1000000
487-
END AS INT) as k
397+
FROM (SELECT CAST(s_suppkey AS INT) as k FROM supplier) s
398+
RIGHT ANTI JOIN (
399+
SELECT CAST(CASE WHEN l_suppkey % 10 = 0 THEN l_suppkey ELSE l_suppkey + 1000000 END AS INT) as k
488400
FROM lineitem
489401
) l
490-
WHERE NOT EXISTS (
491-
SELECT 1 FROM (SELECT CAST(s_suppkey * 10 AS INT) as k FROM supplier) s WHERE s.k = l.k
492-
)"###,
493-
density: 0.1,
402+
ON s.k = l.k"###,
403+
density: 1.0,
494404
prob_hit: 0.1,
495405
build_size: "100K",
496406
probe_size: "60M_RightAnti",
@@ -515,7 +425,9 @@ impl RunOpt {
515425
None => 1..=HASH_QUERIES.len(),
516426
};
517427

518-
let config = self.common.config()?;
428+
let mut config = self.common.config()?;
429+
// Disable join reordering to ensure the optimizer doesn't swap join sides
430+
config.options_mut().optimizer.join_reordering = false;
519431
let rt = self.common.build_runtime()?;
520432
let ctx = SessionContext::new_with_config_rt(config, rt);
521433

‎clippy.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
disallowed-methods = [
22
{ path = "tokio::task::spawn", reason = "To provide cancel-safety, use `SpawnedTask::spawn` instead (https://github.com/apache/datafusion/issues/6513)" },
33
{ path = "tokio::task::spawn_blocking", reason = "To provide cancel-safety, use `SpawnedTask::spawn_blocking` instead (https://github.com/apache/datafusion/issues/6513)" },
4+
{ path = "std::vec::Vec::reserve", reason = "Use `Vec::try_reserve` so allocation failures can be reported instead of panicking", replacement = "try_reserve" },
45
]
56

67
disallowed-types = [

0 commit comments

Comments
 (0)