CUDA: promote opt-in exact bitmap join plans

docs/getting_started.md

@@ -102,6 +102,41 @@ Relation pragmas you'll actually use:
 Rule builders: `atom <= body1 & body2 & ~neg & Filter((v,), "...")`.
 See the {py:mod}`srdatalog.dsl` API reference for every operator.
 
+## Opt-in exact bitmap join plans
+
+`with_plan(dedup_bitmap=True)` selects an exact bitmap implementation for a
+binary set-valued join projection. It changes the physical plan, not the rule's
+logical tuple set, and is not enabled automatically:
+
+```python
+from srdatalog import Program, Relation, Var
+
+join, value, destination = Var("join"), Var("value"), Var("destination")
+Assign = Relation("Assign", 2, input_file="Assign.csv")
+Points = Relation("Points", 2, input_file="Points.csv")
+Output = Relation("Output", 2)
+prog = Program([
+    (Output(value, destination) <= Assign(join, destination) & Points(join, value))
+    .with_plan(dedup_bitmap=True)
+])
+```
+
+For recursive rules, select the intended semi-naive variants explicitly with
+`with_plan(delta=..., dedup_bitmap=True)`. The compiler maintains the required
+dictionary indexes across producer strata and recursive iterations.
+
+This specialization supports unconstrained binary `NoProvenance` projections:
+the shared join variable is eliminated and one distinct variable from each
+source is retained. Filters, negation, constants, incompatible execution
+strategies, count-only rules, unsupported index types, and plans matching no
+recursive variant are rejected rather than silently approximated. Runtime
+shared-memory capacity limits also fail explicitly.
+
+Bitmap plans require a complete runner (`build_project` or `compile_runner`),
+not standalone `compile_pipeline` kernel emission. Keep the ordinary plan as a
+correctness/performance control; enabling a bitmap plan is not by itself evidence
+of an end-to-end speedup.
+
 ## Translating from Nim
 
 The upstream Nim reference has a few dozen benchmark programs under

src/srdatalog/dsl.py

@@ -342,6 +342,7 @@ class PlanEntry:
     - work_stealing   -> mid-level work-stealing (task queue + steal loop)
     - block_group     -> block-group work partitioning
     - dedup_hash      -> GPU hash table for in-kernel existential dedup
+    - dedup_bitmap    -> exact bitmap dedup for binary join projections
   `balanced_root` / `balanced_sources` drive balanced partitioning for
   skewed joins (not yet lowered in Python).
   '''
@@ -355,6 +356,7 @@ class PlanEntry:
   dedup_hash: bool = False
   balanced_root: tuple[str, ...] = ()
   balanced_sources: tuple[str, ...] = ()
+  dedup_bitmap: bool = False
 
 
 @dataclass(frozen=True)
@@ -406,6 +408,7 @@ def with_plan(
     work_stealing: bool = False,
     block_group: bool = False,
     dedup_hash: bool = False,
+    dedup_bitmap: bool = False,
     balanced_root: tuple[str, ...] | list[str] | None = None,
     balanced_sources: tuple[str, ...] | list[str] | None = None,
   ) -> Rule:
@@ -420,6 +423,7 @@ def with_plan(
       work_stealing=work_stealing,
       block_group=block_group,
       dedup_hash=dedup_hash,
+      dedup_bitmap=dedup_bitmap,
       balanced_root=tuple(balanced_root) if balanced_root is not None else (),
       balanced_sources=tuple(balanced_sources) if balanced_sources is not None else (),
     )

src/srdatalog/ir/codegen/cuda/api.py

@@ -23,6 +23,10 @@
     spec for count-phase shape, since the runner files contain
     count bodies but no isolated count-only goldens exist).
 
+Plans requiring preprocessing or multiple coordinated kernel phases must use
+`compile_runner`. Exact bitmap plans cannot be represented by standalone
+kernel entry points and are rejected there.
+
 See:
   - docs/stage2_emitter_audit.md — the per-milestone migration plan.
   - docs/ir_lowering_semantics.md — the formal lowering rules.
@@ -86,6 +90,9 @@ def compile_runner(
   The byte-equivalence gate (`tests/test_runner_byte_equivalence.py`)
   anchors this entry point to the upstream goldens throughout the
   migration.
+
+  Exact bitmap plans use this entry point to emit their execute-only runner,
+  including preprocessing and exact count/materialization.
   '''
   from srdatalog.ir.codegen.cuda.runner import emit_runner_full
 
@@ -134,6 +141,8 @@ def compile_kernel_body(
       slots advance by 2 per FULL_VER D2L source — matching legacy
       `compute_view_slot_offsets`. Pass {} or None for plain DSAI.
   '''
+  if ep.bitmap_join is not None:
+    raise ValueError('dedup_bitmap requires a complete runner; use compile_runner')
   from srdatalog.ir.codegen.cuda.emit import EmitCtx, emit
   from srdatalog.ir.codegen.cuda.envelope import (
     assign_handle_positions,

src/srdatalog/ir/codegen/cuda/bitmap.py

@@ -0,0 +1,113 @@
+'''Execute-only CUDA runner for exact binary projection bitmap plans.'''
+
+from __future__ import annotations
+
+import srdatalog.ir.mir.types as m
+from srdatalog.ir.hir.types import Version
+
+
+def _bind_relation(source: m.ColumnSource, name: str) -> str:
+  full = f'get_relation_by_schema<{source.rel_name}, FULL_VER>(db)'
+  if source.version is Version.DELTA:
+    relation = (
+      f'(iteration == 0) ? {full} : get_relation_by_schema<{source.rel_name}, DELTA_VER>(db)'
+    )
+  elif source.version is Version.FULL:
+    relation = full
+  else:
+    raise ValueError('dedup_bitmap sources must use FULL or DELTA')
+  cols = ', '.join(map(str, source.index))
+  return (
+    f'  auto& {name}_relation = {relation};\n'
+    f'  const auto& {name}_index = {name}_relation.get_index(SRDatalog::IndexSpec{{{{{cols}}}}});\n'
+  )
+
+
+def gen_bitmap_runner(
+  node: m.ExecutePipeline,
+  db_type_name: str,
+  rel_index_types: dict[str, str],
+) -> tuple[str, str]:
+  '''Emit an exact set-projection runner, without hash or ordinary join kernels.'''
+  plan = node.bitmap_join
+  if plan is None or len(node.dest_specs) != 1:
+    raise ValueError('dedup_bitmap requires one binary destination')
+  if node.count or node.dedup_hash or node.work_stealing or node.block_group or node.use_fan_out:
+    raise ValueError('dedup_bitmap cannot be combined with other execution strategies')
+  dest = node.dest_specs[0]
+  if dest.version is not Version.NEW or sorted(dest.index) != [0, 1] or len(dest.vars) != 2:
+    raise ValueError('dedup_bitmap requires a binary NEW destination')
+  for source in (plan.assign, plan.points):
+    if sorted(source.index) != [0, 1] or source.prefix_vars:
+      raise ValueError('dedup_bitmap requires unconstrained binary source indexes')
+    index_type = rel_index_types.get(source.rel_name, '')
+    if index_type and not any(
+      t in index_type for t in ('DeviceSortedArrayIndex', 'Device2LevelIndex')
+    ):
+      raise ValueError(f'dedup_bitmap does not support index type {index_type!r}')
+
+  runner = f'JitRunner_{node.rule_name}'
+  declaration = (
+    f'struct {runner} {{\n'
+    f'  using DB = {db_type_name};\n'
+    '  static void execute(DB& db, uint32_t iteration);\n'
+    '};\n\n'
+  )
+  dictionary = ', '.join(map(str, reversed(plan.points.index)))
+  body = f'''void {runner}::execute(DB& db, uint32_t iteration) {{
+  nvtxRangePushA("{node.rule_name}");
+  struct RangeEnd {{ ~RangeEnd() {{ nvtxRangePop(); }} }} range_end;
+  namespace bitmap = SRDatalog::GPU::bitmap;
+  bitmap::Input input{{}};
+  auto columns = [](const auto& index) -> bitmap::Columns {{
+    using Index = std::remove_cvref_t<decltype(index)>;
+    static_assert(Index::arity == 2 && std::is_same_v<typename Index::ValueType, uint32_t>);
+    if (index.size() == 0) return {{}};
+    return {{index.size(), index.data().template column_ptr<0>(),
+            index.data().template column_ptr<1>()}};
+  }};
+  auto segments = [&](const auto& index, auto& full, auto& head) {{
+    if constexpr (requires {{ index.full(); index.head(); }}) {{
+      full = columns(index.full());
+      head = columns(index.head());
+    }} else {{
+      full = columns(index);
+    }}
+  }};
+  auto keys = [](const auto& index) -> bitmap::Keys {{
+    if (index.size() == 0) return {{}};
+    if (index.num_unique_root_values() == 0)
+      throw std::runtime_error("dedup_bitmap: nonempty index has no value-key cache");
+    return {{index.num_unique_root_values(), index.root_unique_values().data()}};
+  }};
+'''
+  body += _bind_relation(plan.assign, 'assign')
+  body += _bind_relation(plan.points, 'points')
+  body += f'''  segments(assign_index, input.assign, input.assign_head);
+  segments(points_index, input.points_full, input.points_head);
+  const auto& value_index = points_relation.get_index(SRDatalog::IndexSpec{{{{{dictionary}}}}});
+  auto collect_keys = [&](const auto& index) {{
+    if constexpr (requires {{ index.full(); index.head(); }}) {{
+      input.heaps_full = keys(index.full());
+      input.heaps_head = keys(index.head());
+    }} else {{
+      input.heaps_full = keys(index);
+    }}
+  }};
+  collect_keys(value_index);
+  auto& destination = get_relation_by_schema<{dest.rel_name}, NEW_VER>(db);
+  using Destination = std::remove_reference_t<decltype(destination)>;
+  static_assert(!has_provenance_v<typename Destination::semiring_type>);
+  bitmap::execute(input, [&](uint64_t rows) -> bitmap::Output {{
+    const uint64_t old_rows = destination.size();
+    constexpr uint64_t limit = std::numeric_limits<uint32_t>::max();
+    if (old_rows > limit || rows > limit - old_rows)
+      throw std::overflow_error("dedup_bitmap: NEW relation exceeds uint32 row limit");
+    destination.resize_interned_columns(static_cast<std::size_t>(old_rows + rows), 0);
+    return {{destination.template interned_column<0>() + old_rows,
+            destination.template interned_column<1>() + old_rows}};
+  }});
+}}
+
+'''
+  return declaration, '#include "gpu/runtime/jit/bitmap_join.h"\n\n' + declaration + body

src/srdatalog/ir/codegen/cuda/complete_runner.py

@@ -36,6 +36,7 @@
 # NumSources and diverging from upstream.
 import srdatalog.ir.dialects.relation.d2l  # noqa: F401
 import srdatalog.ir.mir.types as m
+from srdatalog.ir.codegen.cuda.bitmap import gen_bitmap_runner
 from srdatalog.ir.codegen.cuda.materialized import is_materialized_pipeline
 from srdatalog.ir.codegen.cuda.pipeline_utils import (
   assign_handle_positions,
@@ -910,6 +911,8 @@ def gen_complete_runner(
   orchestrator can call `JitRunner_X::execute()`).
   '''
   assert isinstance(node, m.ExecutePipeline)
+  if node.bitmap_join is not None:
+    return gen_bitmap_runner(node, db_type_name, rel_index_types or {})
   if rel_index_types is None:
     rel_index_types = {}
 

src/srdatalog/ir/codegen/cuda/orchestrator.py

@@ -167,13 +167,33 @@ def collect_canonical_specs(
 # -----------------------------------------------------------------------------
 
 
+def _required_index_sources(ep: m.ExecutePipeline) -> list[m.MirNode]:
+  if ep.bitmap_join is None:
+    return ep.source_specs
+  plan = ep.bitmap_join
+  return [
+    *ep.source_specs,
+    plan.assign,
+    plan.points,
+    m.ColumnSource(
+      rel_name=plan.points.rel_name,
+      version=plan.points.version,
+      index=list(reversed(plan.points.index)),
+    ),
+  ]
+
+
 def _gen_execute_pipeline(
   instr: m.ExecutePipeline,
   indent: str,
   iter_var: str,
   count_only_rels: set[str],
 ) -> str:
   runner_name = f"JitRunner_{instr.rule_name}"
+  if instr.bitmap_join is not None:
+    if instr.count or is_count_only_pipeline(instr, count_only_rels):
+      raise ValueError('dedup_bitmap does not support count-only execution')
+    return indent + f"{runner_name}::execute(db, {iter_var});\n"
   if instr.count or is_count_only_pipeline(instr, count_only_rels):
     out = indent + "// Count-only query mode\n"
     out += indent + "{\n"
@@ -226,7 +246,7 @@ def _gen_parallel_group(
     else:
       other_ops.append(op)
 
-  has_dedup = any(op.dedup_hash for op in exec_ops)
+  has_dedup = any(op.dedup_hash or op.bitmap_join is not None for op in exec_ops)
 
   if len(exec_ops) <= 1:
     out = indent + "// === ParallelGroup (single rule, sequential) ===\n"
@@ -241,7 +261,10 @@ def _gen_parallel_group(
     return out
 
   if has_dedup:
-    out = indent + "// === ParallelGroup (sequential, dedup_hash present) ===\n"
+    strategy = (
+      "dedup_bitmap" if any(op.bitmap_join is not None for op in exec_ops) else "dedup_hash"
+    )
+    out = indent + f"// === ParallelGroup (sequential, {strategy} present) ===\n"
     for op in exec_ops:
       if op.count or is_count_only_pipeline(op, count_only_rels):
         out += _gen_execute_pipeline(op, indent, iter_var, count_only_rels)
@@ -663,7 +686,7 @@ def gen_fixpoint_body(
           exec_pipelines.append(op)
 
     for ep in exec_pipelines:
-      for src_spec in ep.source_specs:
+      for src_spec in _required_index_sources(ep):
         if isinstance(src_spec, m.ColumnSource):
           ver = version_string(src_spec.version.code)
           spec_type = gen_index_spec_type(src_spec.rel_name, ver, list(src_spec.index))
@@ -851,7 +874,7 @@ def gen_non_recursive_block(
             if isinstance(op, m.ExecutePipeline):
               exec_pipelines.append(op)
         for ep in exec_pipelines:
-          for src_spec in ep.source_specs:
+          for src_spec in _required_index_sources(ep):
             rel_name, raw_ver, idx = extract_source_info(src_spec)
             if rel_name:
               ver = version_string(raw_ver)

src/srdatalog/ir/hir/index.py

@@ -26,6 +26,7 @@
 from __future__ import annotations
 
 from srdatalog.ir.hir.pass_ import IRLevel, PassInfo, PassLevel
+from srdatalog.ir.hir.plan import bitmap_join_patterns
 from srdatalog.ir.hir.types import (
   HirProgram,
   HirRuleVariant,
@@ -101,9 +102,28 @@ def _append_unique_indices(
       dest.append(cidx)
 
 
+def bitmap_index_requirements(hir: HirProgram) -> dict[str, set[tuple[int, ...]]]:
+  '''Include the value-first dictionary index even when no logical source uses it.
+
+  These indices must also survive producer strata: DELTA uses FULL on iteration zero.
+  '''
+  out: dict[str, set[tuple[int, ...]]] = {}
+  for stratum in hir.strata:
+    for variant in (*stratum.base_variants, *stratum.recursive_variants):
+      patterns = bitmap_join_patterns(variant)
+      if patterns is None:
+        continue
+      assign, points = patterns
+      for pattern in patterns:
+        out.setdefault(pattern.rel_name, set()).add(tuple(pattern.index_cols))
+      out[points.rel_name].add(tuple(reversed(points.index_cols)))
+  return out
+
+
 def select_indices(hir: HirProgram) -> HirProgram:
   '''Pass 5 entry. Mutates and returns the HirProgram.'''
   decls = hir.relation_decls
+  bitmap_indices = bitmap_index_requirements(hir)
 
   # ----- First pass: global_index_map over all strata.
   for stratum in hir.strata:
@@ -117,6 +137,12 @@ def select_indices(hir: HirProgram) -> HirProgram:
         if idx_list not in hir.global_index_map[rel_name]:
           hir.global_index_map[rel_name].append(idx_list)
 
+  for rel_name, indices in bitmap_indices.items():
+    registered = hir.global_index_map.setdefault(rel_name, [])
+    for index in sorted(indices):
+      if list(index) not in registered:
+        registered.append(list(index))
+
   # ----- Second pass: per-stratum required + canonical.
   for stratum in hir.strata:
     if stratum.is_recursive:
@@ -141,6 +167,10 @@ def select_indices(hir: HirProgram) -> HirProgram:
         if rel_name in all_idx:
           _append_unique_indices(indices, all_idx[rel_name], arity)
 
+      # Bitmap sources and their value dictionary are explicit physical
+      # requirements, including when a producer already has other local indices.
+      _append_unique_indices(indices, bitmap_indices.get(rel_name, set()), arity)
+
       # Fallback 1: global index map (indices used by OTHER strata).
       if not indices and rel_name in hir.global_index_map:
         for idx in hir.global_index_map[rel_name]: