JoinScan

JoinScan intercepts PostgreSQL join planning and replaces the standard executor with a DataFusion-based pipeline that operates entirely on columnar index data. 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
  TantivyDecodeExec                   ← decodes term ordinals to strings for final K rows only
    TantivyFetchExec                  ← resolves doc addresses to term ordinals for those rows
      SegmentedTopKExec               ← global threshold pruning + final sort + LIMIT K
        HashJoinExec                  ← join on columnar 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 the lookup above it only fetches and 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 columnar indexed, all tables have ParadeDB 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.

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 TantivyDecodeExec over TantivyFetchExec to defer string materialization. With paradedb.defer_column_fetch = off the scan resolves term ordinals itself and only the decode node is injected
  3. RangeCoPartitionedJoinRule — flips a CollectLeft hash join to Partitioned mode when both sides declare compatible Partitioning::Range layouts (any join type except a null-aware NOT IN anti join), so MPP joins partition pairs task-locally instead of broadcasting the build side
  4. DeferredPlacementRule — decides per column where the fetch and the decode of a deferred string column run: a build side or a fan-out join moves the fetch into the scan, and a fan-out with no bounded consumer above moves the decode there too. paradedb.defer_column_fetch and paradedb.defer_string_decode override it
  5. DeferredAggregateRule — splits an aggregate that groups on a deferred string column into a partial aggregate on the term ordinals, a decode of one row per group, and a final aggregate on the strings
  6. 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 packed UInt64 (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().

The lookup has two halves with different access patterns, so they are separate nodes. TantivyFetchExec reads the columnar field (doc_address → term_ordinal); it wants doc order, which a join above the scan no longer keeps. TantivyDecodeExec reads the segment dictionary (term_ordinal → string); it is random access either way, and ordinals are much narrower than strings, so it can move above joins and shuffles at the same cost per row. The planner places the two next to each other and DeferredPlacementRule then moves either half into the scan when the path above would run it out of doc order or on multiplied rows. paradedb.defer_column_fetch and paradedb.defer_string_decode pin each half instead.

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 (Columnar 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. The lookup above it fetches and 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

File Purpose
mod.rs Lifecycle, activation checks, parallel support
build.rs RelNode, JoinCSClause, JoinSource
scan_state.rs DataFusion plan building, optimizer registration, result streaming
planning.rs Cost estimation, field validation, ORDER BY extraction
predicate.rs Postgres expression → JoinLevelExpr
range_partitioning_rule.rs Rules that synchronize Join-side MPP partition boundaries and co-partition the join
visibility_filter.rs Deferred MVCC visibility checking: VisibilityFilterNode, VisibilityFilterExec
translator.rs Postgres ↔ DataFusion expression mapping
explain.rs EXPLAIN output formatting

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

File Purpose
segmented_topk_exec.rs SegmentedTopKExec — per-segment buffers, global threshold publishing
segmented_topk_rule.rs Optimizer rule, wrap_blocking_nodes
visibility_ctid_resolver_rule.rs Wires scan-side FFHelper into VisibilityFilterExec for ctid resolution
tantivy_fetch_exec.rs Columnar field fetch: doc address → term ordinal
tantivy_decode_exec.rs Dictionary decode: term ordinal → string/bytes
filter_passthrough_exec.rs Transparent wrapper enabling filter pushdown through blocking nodes
batch_scanner.rs Scanner::next() — batch iteration, pre-filter, visibility
execution_plan.rs PgSearchScanPlan — dynamic filter integration
pre_filter.rs try_rewrite_binary, collect_filters
deferred_encode.rs Packed deferred column construction and unpacking

GUCs

GUC Default Effect
paradedb.enable_join_custom_scan on Master switch
paradedb.enable_range_partitioned_join true Range co-partitioned joins
paradedb.enable_segmented_topk true SegmentedTopKExec injection
paradedb.defer_column_fetch auto Fetch term ordinals with the decode (on), in the scan (off), or per the placement rule (auto)
paradedb.defer_string_decode auto Decode strings at the consumer (on), in the scan (off), or per the placement rule (auto)