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 parallel execution is enabled in PostgreSQL, JoinScan exclusively relies on Massively Parallel Processing (MPP) via datafusion-distributed to parallelize queries. DataFusion's in-process multithreading is completely bypassed because PostgreSQL has already launched independent parallel worker processes. Instead, the above physical plan is intercepted by the DistributedPlanner and sliced into network stages (DistributedExec), mapping distributed tasks 1:1 with PostgreSQL workers.

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. LateMaterializationRule — injects TantivyLookupExec to defer string materialization
2. SegmentedTopKRule — injects SegmentedTopKExec for Top K on deferred columns, removes the now-redundant SortExec(TopK), wraps blocking nodes with FilterPassthroughExec
3. FilterPushdown (Post) — pushes SegmentedTopKExec's DynamicFilterPhysicalExpr down to the scan

If max_parallel_workers_per_gather > 0 and PostgreSQL has planned parallel execution, DistributedPlanner converts the finalized physical plan into an MPP execution tree (DistributedExec), slicing it into isolated tasks.

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. We map PostgreSQL parallel workers to distributed tasks 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).
2. Task Estimation: During MPP planning, PgSearchScanTaskEstimator intercepts the leaf nodes and requests exactly this partition_count number of tasks.
3. Execution Routing: This routes large multi-segment tables to scale out across all available PostgreSQL parallel workers, where each parallel worker uses ParallelScanState to lazily claim segments. Conversely, tables with a single segment evaluate to exactly 1 task; datafusion-distributed detects the absence of parallel work, avoids MPP planning overhead entirely, and falls back to running the query via local serial execution on a single worker.

Key Files

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

GUCs