JoinScan

JoinScan intercepts PostgreSQL join planning and replaces the standard executor with a DataFusion-based pipeline that operates entirely on Tantivy's columnar fast fields. The core strategy is late materialization: execute the join using only index data, apply sorting and limits, then access the PostgreSQL heap only for the final K result rows.

Physical Plan

For a typical SELECT ... FROM files JOIN documents ... ORDER BY title LIMIT K:

ProjectionExec
  TantivyLookupExec                   ← materializes deferred strings for final K rows only
    SegmentedTopKExec                 ← global threshold pruning + final sort + LIMIT K
      HashJoinExec                    ← join on fast fields
        PgSearchScan (documents)      ← BM25 search
        PgSearchScan (files)          ← lazy scan, deferred columns, receives dynamic filters

When a viable PostgreSQL parallel launch is available, JoinScan uses Massively Parallel Processing (MPP) via datafusion-distributed to parallelize queries. DataFusion's in-process multithreading is bypassed because PostgreSQL has already launched independent parallel worker processes. For a viable MPP launch, DistributedPlanner slices the physical plan into network stages (DistributedExec); logical tasks are assigned round-robin across the workers that attached, so one worker may host multiple tasks.

SegmentedTopKExec publishes dynamic filter thresholds that are pushed down through the join to the probe-side scan, pruning rows at the scanner level. It also performs the final materialized sort and LIMIT, so TantivyLookupExec only decodes K rows (not K×segments).

How It Works

1. Activation

JoinScan fires when all conditions are met: LIMIT present, equi-join keys exist, all columns are fast fields, all tables have BM25 indexes, and at least one @@@ predicate. See create_custom_path() for the full checklist.

2. Planning

The planner hook builds a JoinCSClause — a serializable IR capturing the RelNode join tree, predicates, ORDER BY, and LIMIT. This is stored in CustomScan.custom_private and deserialized at execution time.

• build.rs — RelNode, JoinCSClause, JoinSource
• planning.rs — cost estimation, field validation
• predicate.rs — Postgres expression translation
• privdat.rs — serialization

3. Physical Plan Construction

scan_state.rs builds a DataFusion logical plan from the JoinCSClause, then runs physical optimization:

1. RangePartitioningRule — coordinates split points across joins for MPP range partitioning, sampling both sides of the join and injecting the merged sample into both PgSearchTableProviders
2. LateMaterializationRule — injects TantivyLookupExec to defer string materialization
3. RangeCoPartitionedJoinRule — flips a CollectLeft inner hash join to Partitioned mode when both sides declare compatible Partitioning::Range layouts, so MPP joins partition pairs task-locally instead of broadcasting the build side
4. SegmentedTopKRule — injects SegmentedTopKExec for Top K on deferred columns, removes the now-redundant SortExec(TopK) and transfers ownership of its already pushed-down DynamicFilterPhysicalExpr into the injected node, wraps blocking nodes with FilterPassthroughExec

When MPP is eligible, DistributedPlanner builds an MPP execution tree (DistributedExec), slicing it into isolated tasks. Plans with fewer than two producer tasks, or launches with fewer than two attached workers, run serially.

4. Deferred Columns

String columns are emitted as a 2-way UnionArray (doc_address | term_ordinal) so intermediate nodes work with cheap integer ordinals instead of decoded strings. The decision to defer is made in configure_deferred_outputs().

5. Pruning Path

There are two primary pruning mechanisms for dynamic filters that are pushed down to the scan:

1. Query-Time Pushdown (Inverted Index): Filters that are static and known at the start of the scan (such as InList predicates generated from a HashJoin build side) are intercepted during the first poll_next of the scan stream. They are converted into native Tantivy queries (e.g., TermSetQuery) and ANDed into the main search query via try_dynamic_filter_pushdown. This allows Tantivy to use its inverted index to filter documents while executing the search, providing the highest possible pruning performance. The DataFusion expressions are then rewritten to lit(true) so they are not evaluated again.

2. Pre-Filter Pushdown (Fast Fields): For evolving thresholds, such as the global threshold from SegmentedTopKExec, the threshold is pushed down to the scan via filter pushdown. This works because SegmentedTopKExec and PgSearchScan share an Arc<DynamicFilterPhysicalExpr>. The scanner reads current() on every batch and applies the filter after the search but before Arrow column materialization. For strings, it translates literals to per-segment ordinal bounds via try_rewrite_binary and filters directly against the fetched term ordinals.

6. Execution Result

After all input is consumed, SegmentedTopKExec materializes sort column values, performs the final sort, and emits exactly K rows. TantivyLookupExec decodes deferred strings for those K rows only. JoinScanState extracts CTIDs and fetches heap tuples — the only point where the PostgreSQL heap is accessed.

7. MPP Execution and Parallelism

JoinScan does not use DataFusion's standard in-process multithreading. Since PostgreSQL already coordinates execution across independent backend processes via the Gather node, relying on thread-level parallelism inside a Postgres worker would result in Workers * Threads explosions, and Postgres does not support interacting with its APIs anywhere but on the main thread.

Instead, MPP via datafusion-distributed is our only mechanism for parallelizing joins. It assigns logical tasks across PostgreSQL parallel workers based on segment count:

1. Partition Output Definition: Because index segments are checked out atomically from shared memory, PgSearchScanPlan natively partitions its output by the number of segments. In table_provider.rs, we formally expose the scan's output partition count as min(segment_count, target_partitions). When the RangePartitioningRule has injected a range sample, the scan instead declares Partitioning::Range with the sample's split points, which lets DataFusion treat the two sides of a join as co-partitioned.
2. Task Estimation: During MPP planning, PgSearchScanTaskEstimator intercepts the leaf nodes and requests exactly this partition_count number of tasks.
3. Execution Routing: For a viable MPP launch, tasks are assigned round-robin across the workers PostgreSQL attached, and each worker uses ParallelScanState to lazily claim segments. A one-task plan does not launch MPP workers and runs serially.

Key Files

Execution-layer files under pg_search/src/scan/:

GUCs