federated-planning

name

federated-planning

description

Cost-based federated SPARQL source selection + bind-vs-hash join planning over already-fetched source descriptors, plus an ANAPSID-style non-blocking streaming join with operator spill, via the opt-in sparq-fedplan crate. Use when planning a federated BGP across multiple SPARQL endpoints from their served statistics (VoID property/class partitions + mined scs: characteristic sets): deciding which sources can contribute to each triple pattern (HiBISCuS recall-safe pruning + CostFed skew-aware cardinality), choosing a join order with per-join bind-vs-hash-vs-streaming algorithm selection (characteristic-set star cardinality for intermediate sizes), and executing a memory-bounded non-blocking symmetric hash join over incrementally-arriving sub-results (StreamJoin, spill to a backing store, result multiset-equal to a blocking join). Pure + deterministic planning, no network I/O. Off by default; does not touch sparq-core/sparq-engine's lean build. Also covers live adaptive RE-planning at stage boundaries (mid-execution plan switching when observed cardinalities diverge from estimates, or when a source is observed to be slow — per-source EWMA-smoothed latency is folded into the cost model as a documented heuristic bias toward faster sources, smoothing out transient spikes) via the further opt-in adaptive-replan feature (AdaptiveExecutor) — sound because BGP join is order-independent (latency changes ordering/cost only, never results), with mid-operator swap + live source failover deferred.

sparq-fedplan — cost-based federated source selection + join planning

sparq-fedplan plans a federated SPARQL Basic Graph Pattern (BGP) across several remote endpoints from statistics already in hand — it never contacts the network. A caller fetches each source's descriptor once (the W3C VoID document a sparq-server serves at /.well-known/void, including the mined scs: characteristic sets), and the planner decides, deterministically:

1. which sources can contribute to each triple pattern, and
2. a join order + per-join bind-vs-hash algorithm over the selected sources.

It is the opt-in public surface for cost-based federation planning. Add it explicitly and enable the fedplan feature; it is not in sparq's default build (sparq-core/sparq-engine stay lean, the wasm artifact is unchanged unless you pull it in). There is no sparq-core/sparq-engine dependency — it plans over descriptors only.

Add the dependency

[dependencies]
sparq-fedplan = { path = "crates/sparq-fedplan", features = ["fedplan"] }
oxrdf = { version = "0.3", features = ["rdf-12"] }

The whole planner is behind the fedplan feature (off by default), so even a crate that depends on sparq-fedplan pays nothing for it unless the feature is enabled.

Build source descriptors

Either programmatically via the builder, or by parsing the served N-Triples document.

// `SourceDescriptorBuilder` is the (public, nameable) return type of `.builder(..)`; you only
// need to import it if you hold the builder in a `let` rather than chaining straight to `.build()`.
use sparq_fedplan::{SourceDescriptor, SourceDescriptorBuilder, SourceId, PredPartition, ClassPartition};

// Programmatic (VoID property/class partitions).
let src = SourceDescriptor::builder(SourceId::new("https://a.example/sparql"))
    .total_triples(10_000)
    .predicate(PredPartition { predicate: "http://xmlns.com/foaf/0.1/knows".into(),
        triples: 2000, distinct_subjects: 1000, distinct_objects: 1800 })
    .class(ClassPartition { class: "http://xmlns.com/foaf/0.1/Person".into(), entities: 1000 })
    .build();

// Or parse the served descriptor (the /.well-known/void N-Triples form, with scs: sets):
let parsed = SourceDescriptor::from_void_nt(SourceId::new("https://b.example/sparql"), nt)?;

A descriptor parsed from VoID partitions is authority-incomplete (it sees only predicate/class authorities, never subject/object instance authorities), so subject/object authority-pruning is disabled for it — recall-safe by construction. To enable authority pruning, build via .builder(..) and call .authorities_complete() only when you truly enumerate every authority the source mints (a HiBISCuS-style capability set / void:uriSpace declaration).

A descriptor may also carry an optional GenAI retrieval capability (RetrievalCapability — declared vector/text retrieval endpoints + a per-request cardinality hint), served under the sret: vocab (http://sparq.dev/ns/retrieval#, RETRIEVAL_NS) next to the VoID partitions and round-tripped via RetrievalCapability::to_void_nt / from_void_nt. It defaults to absent (descriptor.retrieval() == None) and is advisory planner metadata only: select_sources consumes it as a STATIC candidate-ordering hint (sources declaring a cardinalityHint order first, ascending by hint; undeclared keep index order after them; source index tie-breaks — so with no hints the ordering is the historical ascending-index order), but it can never change which triples answer a BGP: the retained-source SET and every estimate are hint-independent (differential-tested; the same answer-safe discipline as the cardinality estimators). Set it with .retrieval(..) on the builder. [FABLE-5] sq-3uijg

Select sources for a BGP (recall-safe)

use sparq_fedplan::{Bgp, TriplePattern, Term, Var, select_sources};

let bgp = Bgp::new(vec![
    TriplePattern::new(Term::Var(Var::new("s")),
        Term::Iri("http://xmlns.com/foaf/0.1/knows".into()), Term::Var(Var::new("o"))),
]);
let sources = [src];
let selection = select_sources(&bgp, &sources);
// selection[i].candidates: the sources retained for pattern i, with estimated_cardinality.

Recall-safety invariant: a source is pruned for a pattern only when the descriptor proves it holds no matching triple. A bound predicate absent from the source's (complete) predicate-partition set prunes; a bound class absent from a declared class section prunes; a bound subject/object whose authority is absent prunes only when the authority set is complete. On any uncertainty — open predicate, incomplete authority set, absent class section — the source is kept. The cardinality estimate never prunes (a source with a tiny or zero estimate is still retained). This is HiBISCuS's design goal: maximise pruning subject to never losing a result.

Optional live pattern probes in sparq-fedclient

sparq-fedplan itself remains pure and never contacts the network. When served VoID statistics are missing, the separate federation client can refine its PatternSources through the default-OFF sparq-fedclient/pattern_probe feature (which implies fedclient):

[dependencies]
sparq-fedclient = { path = "crates/sparq-fedclient", features = ["pattern_probe"] }

Create one PatternProbeSession per query, using the same SSRF-policy-controlled Fetcher as capability discovery, then pass endpoint/optional-descriptor pairs in planner source-index order:

use sparq_fedclient::{
    select_sources_with_pattern_probes, PatternProbeConfig, PatternProbeSession, ProbeSource,
};

let fetcher = sparq_fedclient::discovery::HttpFetcher::new();
let mut session = PatternProbeSession::new(&fetcher, PatternProbeConfig::default());
let selection = select_sources_with_pattern_probes(
    &bgp,
    &[ProbeSource {
        endpoint: "https://example.org/sparql",
        descriptor: discovered_descriptor.as_ref(),
    }],
    &mut session,
);

The request budget is per query and counts every ASK/SELECT HTTP request; repeated source-pattern pairs are cached. Served VoID cardinalities issue no probe. Recall safety is load-bearing: only an exact ASK false removes a source. Timeout, HTTP/parse error, inconsistent responses, or budget exhaustion retain it with the uniform fallback. A successful capped SELECT row count replaces that fallback and can change ranking/join order, never the answer multiset. [GPT-5.6] sq-fx5id.

Plan the join (bind vs hash)

use sparq_fedplan::{plan_bgp, PlanOptions, JoinAlgo, JoinNode};

let plan = plan_bgp(&bgp, &selection, &sources, &PlanOptions::default()).unwrap();
let order: Vec<usize> = plan.join_order(); // patterns in join order (left-deep)
let cost: f64 = plan.total_cost;

Each binary join is a bind join (cost ≈ L·(req + fan_out) — probe the right with the left's bindings; cheap when the left is small and the right selective) or a hash / symmetric join (cost ≈ R + L — scan both sides once; cheap when the left is large or the right unselective). The decision flips as the left intermediate grows past the point where per-row requests overtake a full scan; tune the round-trip penalty with PlanOptions::request_cost. Star arms (?s p₁ ?a . ?s p₂ ?b) use characteristic-set cardinality (Σ_{C⊇Q} count(C)·Π avg_mult) for intermediate sizes, capturing the predicate correlation an independence product loses.

Predicate-selectivity-aware non-star cardinality (opt-in, use_predicate_selectivity, sq-jsuzr)

By default a connected non-star join (e.g. a chain ?s :p ?o . ?o :q ?z) falls back to a coarse independence estimate out = |L|·|R| / max(|L|, |R|) that approximates the join-key distinct-value count by the larger leaf cardinality. Setting PlanOptions::use_predicate_selectivity = true instead folds the candidate's actual per-predicate distinct counts — distinct_subjects when the join variable sits in the candidate's subject position, distinct_objects when in its object position, summed (union) over the candidate's retained sources — into the ndv, giving a skew-aware estimate out = |L|·|R| / Σ distinct(join-key). These are the same VoID stats every SourceDescriptor already carries. When the relevant distinct count is absent (VoID leaves it 0 when unknown), or the join key is in neither position, it falls back to the identical max-leaf estimate, so it never fabricates an ndv. Cost-estimate-only: this changes the planner's cardinality (hence its join order and bind-vs-hash choice), but never the result multiset — BGP join is commutative/associative, so any order over the same patterns answers identically. The flag is OFF by default so the prior estimate is preserved byte-for-byte for A/B comparison; star arms keep using the characteristic-set estimate regardless. Whether the tighter estimate yields faster plans is a measurable hypothesis to confirm on the canonical perf host (sq-0g6g), not a claim made here.

Non-blocking streaming join + spill (StreamJoin, sq-vf7q)

The planner's JoinAlgo::Streaming choice corresponds to an execution-side operator: StreamJoin, an ANAPSID/XJoin-style symmetric hash join over two incrementally-arriving tuple streams. Feed tuples from either side as federated sub-results arrive; each push returns the results that arrival newly completes — it never blocks on either input finishing.

use sparq_fedplan::{StreamJoin, StreamJoinOptions, SpillStore, Tuple, Var, blocking_hash_join};

let opts = StreamJoinOptions { mem_budget_tuples: 100_000, spill_store: SpillStore::TempFile };
let mut join = StreamJoin::new([Var::new("s")], opts);
let _ = join.push_left(Tuple::new([(Var::new("s"), "a".into()), (Var::new("o"), "1".into())]));
let out = join.push_right(Tuple::new([(Var::new("s"), "a".into()), (Var::new("n"), "x".into())]));
assert_eq!(out.len(), 1); // emitted the moment both sides hold key s=a — non-blocking.

Bounded spill. Memory is capped at mem_budget_tuples; when an insert would exceed it, the largest in-memory join-key partition is spilled to a backing run (a temp file under std::env::temp_dir by default — std only, no new dependency; SpillStore::Memory is an in-process simulation for tests). Spilling only relocates tuples; the probe consults both the live bucket and every spilled run for the key.

Correctness invariant (load-bearing). The streamed + spilled result is multiset-equal to blocking_hash_join(left, right, join_vars) — same tuples, no loss, no duplication — for any stream interleaving and any budget (including one so low every partition spills). Each matching pair (l, r) is emitted exactly once, when the second of the two arrives and probes the other side (found in memory or a spill run). Proven by the streamed_equals_*, spill_path_equals_*, duplicate_keys_*, and emits_before_inputs_exhausted tests.

The planner picks Streaming over plain Hash when a hash-class join's combined estimated inputs L + R exceed PlanOptions::stream_threshold (default 100 000 rows; set to f64::INFINITY to always use plain hash) — large joins run non-blocking + spillable rather than materialising a side up front.

Live adaptive re-planning (opt-in adaptive-replan feature, sq-7s4z)

Behind the off-by-default adaptive-replan cargo feature (which implies fedplan), the crate adds the reactive half of ANAPSID adaptivity: mid-execution plan switching. A build that does not enable the feature compiles zero adaptive code (#[cfg]-gated out), so the lean default build and the fedplan-only build are byte-unchanged.

AdaptiveExecutor models execution as a sequence of stages (the left-deep join order) and holds, at all times, the patterns already joined (the prefix) and the patterns still to join (the suffix).

• Capture — RuntimeStats records the observed per-pattern leaf cardinality (real row counts the sources returned) and per-source latency, fed in as each stage completes. Latency is EWMA-smoothed per source (below) so the cost model + trigger track the trend, not the last raw sample.
• Trigger — at each stage boundary, maybe_replan(&stats) checks whether a not-yet-executed pattern's observed cardinality o diverges from its estimate e past ReplanPolicy::divergence_factor k either way (o > k·e or e > k·o; default k = 4), or (sq-b51o) a not-yet-executed pattern's slowest source is observed at more than k× the latency baseline. If so it re-invokes the cost model on the remaining patterns with the observed cardinalities substituted in (corrected_selection) and the join costs latency-weighted (below). Source membership is never re-pruned — only the order changes — so HiBISCuS recall-safety is preserved.
• Per-source latency weighting (sq-b51o) — a HEURISTIC, not optimal. A source can be slow (contended / far / rate-limited) even at exactly its predicted cardinality, so the re-planner also folds observed latency into the cost. Each candidate join's cost is scaled by factor = clamp(1 + latency_weight·(s − 1), latency_floor, latency_cap) where s = slowest_ewma_source_latency / latency_baseline over the pattern's retained sources. A source at baseline — or with no observation — yields factor = 1.0, so a re-plan with no latency data is byte-identical to the cardinality-only planner; a 2×-slow source costs 1.5× at the default latency_weight = 0.5, with latency_cap (4.0) bounding any one outlier. The constants (latency_weight 0.5, latency_baseline 100, latency_floor 0.5, latency_cap 4.0) are hand-tuned, not derived — a deliberately gentle bias toward faster sources / deferring a slow one, not a claim to compute the latency-optimal plan. Latency enters only the cost term (and the suffix-selection score), never the output cardinality, so results are unchanged. latency_weight = 0 disables it.
• Per-source vs slowest-arm aggregation (sq-s5kd) — opt-in, default unchanged. How a multi-source pattern's per-source latencies fold into its single cost factor is now a policy knob, ReplanPolicy::latency_aggregation: LatencyAggregation::SlowestArm (default — the sq-b51o bottleneck model above, bit-identical for existing callers) or LatencyAggregation::CardinalityWeighted — each retained source's latency runs through the same factor formula individually and the pattern is charged the cardinality-weighted mean of the per-source factors (an expected-work model: a tiny slow arm no longer dominates a huge fast arm's pattern; plain mean when every retained cardinality is 0). An unobserved source still contributes 1.0, so with no observations both modes are exactly neutral, and single-source patterns are mode-independent. The re-plan trigger stays slowest-arm under both modes (its job is to detect a pathologically slow arm; the weighted cost + hysteresis then decide adoption). Which model wins is workload-dependent (parallel-fetch unions really are bottlenecked; sequential/bind-join work is proportional) — the MECHANISM ships un-tuned for the federation bench harness to compare.
• Latency EWMA smoothing (sq-b51o follow-up) — a HEURISTIC α, not optimal. The cost factor and the trigger read a per-source exponentially-weighted moving average, not the single last sample: record_source_latency folds each sample in as ewma = α·observed + (1−α)·prev (first sample seeds it), with α = RuntimeStats::latency_alpha, default DEFAULT_LATENCY_ALPHA = 0.3 — a hand-picked factor (latest 30% / history 70%), not workload-derived. This is the cleaner anti-thrash than a bare last-sample-plus-clamp: a single transient spike does not move the average over the trigger band, but a sustained shift converges past it in a few samples. The latency_floor/latency_cap clamp is kept as a final guard. RuntimeStats::with_latency_alpha(α) overrides (α = 1.0 ⇒ un-smoothed last-sample behaviour); higher α = faster-tracking/twitchier, lower α = calmer/laggier.
• EWMA refinements (sq-3xkz) — per-source α, time-aware decay, eviction — all opt-in, default-off. Three knobs sharpen the EWMA when a RuntimeStats is reused across queries; each defaults to off so the default path is byte-identical to the single-global-α EWMA above.
  • Per-source adaptive α. RuntimeStats::set_source_alpha(source, α) / with_source_alpha gives a source its own smoothing rate; latency_alpha is the fallback and effective_alpha(source) resolves the override-else-global α. An empty override map ⇒ every source uses the global α ⇒ the prior behaviour reproduced. This ships the MECHANISM
    • a sensible default; choosing a good per-source α needs a real federated workload, so that tuning is deferred — no α value is claimed optimal.
  • Time-aware decay. The plain EWMA equal-weights samples regardless of the gap between them; record_source_latency_after(source, latency, elapsed) with a half-life set (set_decay_half_life / with_decay_half_life) inflates the effective α toward 1.0 as the elapsed gap grows past the half-life (α_eff = α₀ + (1−α₀)·(1 − 0.5^(Δt/half_life))), so a fresh sample after a long idle gap is trusted more and stale history decays toward the prior. The elapsed gap is passed in (the logical clock is injectable, never read internally), so the decay is deterministic + testable. No half-life ⇒ folds at the plain α (back-compat).
  • Staleness / eviction. evict_stale(max_age) drops every source whose source_age (clock − last_seen, on the same injectable logical clock) exceeds the threshold (strictly greater; equal is kept), returning the count evicted — so a stale latency stops biasing a long-lived store. advance_clock(elapsed) ages entries without recording a sample.
• Hysteresis — the re-planned suffix is adopted only if its estimated remaining cost (cardinality- and latency-weighted) beats the current suffix's by more than ReplanPolicy::improvement_margin (default 10%), with a hard max_replans budget (default 8). Stable-but-noisy stats — including jittery latency — never thrash; maybe_replan returns ReplanOutcome::{NoDivergence, KeptWithinHysteresis, Switched, BudgetExhausted}.

Soundness boundary (load-bearing). Re-planning reorders only the not-yet-started suffix — it is NOT a mid-operator swap (an in-flight join is never torn down). A BGP answer is the natural join of the per-pattern solution multisets, which is commutative and associative: any order over the same patterns yields the same result multiset, and the already-produced prefix is carried across the switch unchanged (no binding lost or duplicated). The latency weighting does not move this boundary — it touches only cost/ordering, never the output cardinality or the pattern set, so a latency-driven reorder is the same kind of pure suffix permutation; the EWMA smoothing changes only when the latency path fires, never this boundary — re-planning stays a sub-query / stage-boundary reorder, never mid-operator. Proven by adaptive::tests::replan_result_equals_static (cardinality-driven), latency_replan_result_equals_static (latency-driven) and ewma_replan_result_equals_static (EWMA-smoothed-latency-driven) — each genuinely flips the order yet yields the identical multiset to the static plan — plus an exhaustive all-permutations order-independence test.

The federation CLIENT that consumes this planner (sparq-fedclient, sq-dnko)

sparq-fedplan is the planning brain with no consumer — it plans, but nothing fetches descriptors or issues a query. The consumer is a separate opt-in crate, sparq-fedclient (epic sq-dnko / sq-3183, architecture research/federation-client-design.md): the streaming federation client that discovers each remote source's capability, lowers a query BGP into this crate's Bgp, calls select_sources + plan_bgp, interprets the resulting JoinTree into physical operators (Bind → VALUES bind-join, Hash/Streaming → the StreamJoin above, Local → sparq-engine eval), and streams results back. It REUSES this planner and the engine's service SRJ transport + SSRF guard; the dependency arrow points one-way into the engine.

sparq-fedclient started as the Phase-0 skeleton (sq-s1uy) — the public module layout (source / discovery / planner / pushdown / operators / stream) behind a default-OFF fedclient feature, plus the load-bearing dependency-boundary proof (sparq-core/sparq-engine have no edge to it, enforced by scripts/fedclient-boundary-guard.sh + crates/sparq-fedclient/tests/boundary.rs). Landed since:

• Phase 1 discovery (sq-nfxl) — Service-Description parser + VoID/scs: reuse + an SSRF-guarded fetch seam + ASK-probe fallback → a Capability (+ optional SourceDescriptor for this planner). The Capability also reads the SPARQL 1.2 SD sd:supportedVersion <sparql:version-*> advertisement (sq-ym6kf, consuming the server side sq-2msb emits) into sparql_versions: Vec<SparqlLanguageVersion>, so a 1.2-aware client can detect full-1.2 (triple-term / dir-lang) support via Capability::advertises_full_sparql_1_2() without probing. Honest boundary: an empty set means the source published no sd:supportedVersion ⇒ version posture UNKNOWN (older endpoints predate the term), not unsupported — distinguish the two with Capability::advertises_sparql_versions().
• Phase 2 source abstraction (sq-rsxf) — the Endpoint adapter over the engine's transport seam behind a default-deny SSRF guard.
  • Port-scoped allowlist entries (sq-vbnyc, follow-up to the engine's sq-a7jw4): source::EgressGuard's host allowlist now accepts a port-scoped entry (127.0.0.1:8053, [::1]:8080, .example.org:443) that re-opens a private host on THAT exact port only — strictly narrower than a bare host-level entry, which still re-opens every port (backward compatible). The dialled port (the authority's :port or the scheme default) is the port vetted, so a host:port entry never widens. The fedclient guard delegates the per-entry decision to the engine's shared sparq_engine::allowlist_entry_permits (and allowlist_entry_host_matches for the host-level "is this host on the list at all" query), so the fedclient guard and the engine SERVICE guard decide every host:port case identically — one source of truth, no divergent copy of the parsing (port-0/overflow/IPv6-bracket/ trailing-colon all fail-CLOSED). The same port-scoping flows through EgressGuard::check_addr / check_endpoint and all three native ureq SSRF resolvers (HttpTransport / HttpFragmentTransport / discovery HttpFetcher). There is no wildcard port and no global bypass; default-deny stays default-deny. Use EgressGuard::is_allowed_port(host, port) for the port-precise check.
• Phase 3 planner bridge + single-source interpreter (sq-j27p) — the consumer of THIS planner's JoinTree. The plan speaks pattern/source indices only (no endpoint-URL mapping — the Phase-0 finding); sparq_fedclient::SourceResolver is the index → adapter resolution layer that maps a plan pattern: usize → TriplePattern and a source: usize → a source adapter (the resolver requires the descriptors/adapters slices to be in the same order, and range-checks every lookup). lower_leaf lowers one BGP pattern to a single-pattern SELECT; materialize_single_source walks the JoinTree, fetches each leaf's SRJ through the Phase-2 adapter, parses it, and natural-joins in the plan's join order. The load-bearing property — the materialised federated result equals local sparq-engine evaluation of the same query (solutions_equal bag comparison) — is driven end-to-end in tests/planner_result_equals_local_eval.rs (in-memory Transport double) and, over a REAL in-process sparq-server loopback on 127.0.0.1:0, in tests/endpoint_loopback_result_equals_local.rs (sq-my8wd.2): the latter closes the honest gap that the equivalence tests otherwise only exercise the in-memory seam (the SRJ HTTP round-trip + the on-the-wire SSRF guard are never touched), and asserts the load-bearing egress invariant non-vacuously — a sibling default-deny Endpoint against the same live loopback host is FedError::EgressRefused while the per-endpoint-allowlisted one reaches the server, so the refusal cannot be a "server is down" artifact; discovery is driven the same way (allowlisted HttpFetcher reaches it via the ASK-probe fallback, a default-deny one is refused). The fedclient guard is owned per-endpoint (no process-global to widen), so the allowlist granted to one endpoint never leaks to another. materialize_single_source is single-source + blocking (its streaming counterpart is Phase 5 below); a leaf the planner retained >1 source for fails closed with InterpError::MultiSource. To fan such a leaf out as a per-source UNION use materialize_multi_source / stream_multi_source (bead sq-7yf0, below) — the opt-in multi-source entry points that never return MultiSource.
• Phase 4 capability-aware pushdown (sq-7byx) — the pushdown module decides the most precise sub-query each source is asked. exclusive_groups(selection, bgp) derives the FedX exclusive groups (maximal connected sub-patterns whose only retained source is one member — exactly-one-source, same-source, share-a-variable, via union-find); a cross-source or zero-source pattern is excluded. push_group(...) builds the maximal sub-algebra per group: projection trimmed to the join + output vars, the FILTER conjuncts the source's FilterClass covers AND that pass the common-variable check, ORDER/LIMIT when the capability allows — a full endpoint gets the whole group as one multi-pattern SELECT, a fragment source answers one pattern only (no collapse, no filter pushed — honest about a fragment server's access unit). common_variable_check(filter, group_vars) is the exact check Comunica omits (#834/#609): push a conjunct only when every variable it references is bound by the group. render_values_block / bind_block_size are the cross-group bind-join block primitive (VALUES for endpoints — DEFAULT_BIND_BLOCK; maxMpR for brTPF; none for plain TPF), mirroring sparq-engine's pub(crate) service.rs helpers. Pushdown only ever narrows a source's result, so it is correctness-preserving; the FILTER model is light (the parsed-query FILTER algebra wiring is Phase 5).
• Phase 5 streaming operators (sq-vtba) — the streaming counterpart of the Phase-3 interpreter, built ON THIS crate's non-blocking StreamJoin. sparq_fedclient::stream's SolutionStream is a bounded, backpressured Iterator over a std::sync::mpsc::sync_channel (the channel bound IS the backpressure); operators::ScatterPool is a bounded blocking thread-pool over the blocking transport — the ASYNC/RUNTIME decision (no async runtime is pulled in; all concurrency is std-only and confined to the opt-in crate). StreamingJoin drives THIS crate's StreamJoin over two SolutionStreams, bridging oxrdf::Term rows into the Tuple model losslessly via the term's canonical N-Triples form (Term::Display ↔ Term::from_str). stream_single_source walks the same JoinTree, fans each leaf's blocking fetch onto the pool, and chains the leaves through streaming joins so results EMIT before the inputs are exhausted. The load-bearing invariant — the streamed multiset is multiset-equal to the Phase-3 materialised result for any source-arrival interleaving (and both equal local eval) — is driven on the real engine path under injected per-leaf delays + a forced spill in tests/streaming_result_equals_phase3.rs. The pushed-down bind-join (VALUES/maxMpR) remains deferred; a bind-classified join runs as the same streaming symmetric hash join (identical result multiset).
• Multi-source UNION-per-leaf fan-out (sq-7yf0) — materialize_multi_source / stream_multi_source lift the single-source InterpError::MultiSource guard: a leaf the planner retained >1 source for is answered as the bag-union of every retained source's solutions for that pattern (SPARQL UNION's multiset semantics — concatenation, no de-dup, multiplicity preserved), resolving each candidate source index to its adapter through the SourceResolver. The streaming path fans each retained source's fetch onto the same ScatterPool, all feeding ONE per-leaf SolutionStream (a cloned SolutionSink per source job). Both fold the per-leaf unions through the unchanged left-deep natural_join / StreamingJoin. tests/multi_source_union_result_equals_local.rs proves the materialised AND streamed multi-source result equals local sparq-engine evaluation over the union of every source's graph. The single-source entry points keep the fail-closed MultiSource contract — multi-source is the opt-in entry point.
• Phase 6 the brTPF + TPF fragment adapters (sq-2qze): source::TpfSource (plain TPF — materialise a fragment to exhaustion, no bind-join) and source::BrTpfSource (bindings-restricted — push maxMpR-bounded binding blocks per request, the standardised brTPF bind-join), both over the FragmentTransport seam, with count-metadata (hydra:totalItems) cardinality surfaced as a one-pattern SourceDescriptor for this planner. The fragment adapters answer one triple pattern completely and return typed FragBindings via solutions(...) (a fragment server speaks triples, not SPARQL-Results-JSON, so their FederatedSource::execute is a deliberate Unsupported that points at solutions).
  • Native HTTP FragmentTransport + interpreter wiring (sq-yzca): source::HttpFragmentTransport (native-only — ureq behind the SAME default-deny SSRF resolver as the SRJ HttpTransport) is the production seam. It serialises a FragPattern into the Hydra TPF query string (?subject=&predicate=&object=, percent-encoded N-Triples terms), attaches a brTPF binding block as the values parameter (the server's text wire), follows the opaque hydra:next page URL to exhaustion, and parses the Turtle/TriG body (oxttl TriGParser, a Turtle superset) — splitting Hydra/VoID control triples (hydra:totalItems/void:triples → count; hydra:next → next link) from data triples (kept only when they match the requested pattern). The operators interpreter is wired: fetch_leaf_relation dispatches on source_type(), routing an endpoint/local leaf to the SRJ execute path and a TPF/brTPF leaf to the typed solutions path (lowered via planner::lower_leaf_fragment), converting FragBinding rows back to oxrdf::Term so a fragment leaf equi-joins with an endpoint leaf. A brTPF leaf currently runs as a complete unbound scan the interpreter hash-joins locally (the same discipline the Phase-3 interpreter applies to JoinAlgo::Bind); the streamed per-block bind-join feeder is a later phase.
  • brTPF binding-block wire codec (the wire module, sq-6ihg, follow-up to the server's sq-dxhb): the brTPF bind-join attaches a SET of upstream solution mappings (a &[FragBinding] block, at most maxMpR) to each fragment request, and that block is re-sent on every request of a bind nested-loop join. The sparq server parses it from a line-oriented TEXT wire — one mapping per line, space-separated position=term pairs, each term fully N-Triples-decorated — which is readable but verbose: it repeats the s=/p=/o= key and the <…>/"…"/^^<…> framing on every term. The wire module adds the compact, self-describing BINARY mapping wire the bead asks for (encode_bindings / decode_bindings) the client can emit instead, plus the text-wire writer (encode_bindings_text) so a client can speak EITHER form over the same FragBinding model (the server already parses the text one). The binary form's twofold compactness win: a 1-byte per-mapping header bitmask records which of s/p/o the mapping binds, so a position term carries no name bytes (the header bit IS the name), and a 1-byte kind tag distinguishes IRI / blank / literal so the bare lexical bytes follow length-prefixed with no <>/"" framing. A binding over an arbitrary (non-position) variable name still round-trips losslessly via an overflow EXTRA section, and the binary wire carries literals with embedded =, whitespace, or newlines that the one-mapping- per-line text wire cannot represent (the text writer drops a non-position variable — it has no brTPF slot). The container leads with a 4-byte magic (BINARY_MAGIC, ASCII bTPF)
    • a 1-byte BINARY_VERSION so a future revision is detectable, and decode_bindings validates every length against the remaining input, so a truncated / bad-magic / bad- version / bad-kind / varint-overflow buffer is a clean WireError, never a panic or OOB read (the crate is forbid(unsafe_code)). Position keys decode in a deterministic canonical s→p→o order, and the empty mapping μ₀ is skipped on encode (it does not restrict a fragment) exactly as the server's parse_bindings skips an all-blank line. HONEST scope — a codec only: it converts &[FragBinding] ↔ bytes / String; it issues no request. The native HTTP FragmentTransport (HttpFragmentTransport, sq-yzca, above) attaches the text form on the values query parameter (the carrier the server reads); this binary wire is the compact alternative a body-carrying transport emits.
• Phase 7 adaptive re-planning (sq-ij5x, the FINAL phase) — the client-side ANAPSID feedback loop, behind the extra default-OFF fedclient-adaptive feature (which pulls this planner's adaptive-replan). adaptive::execute_adaptive_single_source runs the plan as a leaf-scan phase (fetch each leaf once through the real adapter, record its REAL observed row count into RuntimeStats) followed by an adaptive join-ordering phase that drives this crate's AdaptiveExecutor: at each operator boundary it re-invokes the planner on the unjoined remainder when an observation diverges past divergence_factor, adopting the cheaper suffix only when it clears the hysteresis margin — at most once per boundary, no thrash. The re-plan DECISION engine is this planner's AdaptiveExecutor (the client does not re-write it); the client supplies real observed cardinalities and joins the re-ordered suffix with the SAME materialised natural_join. Re-planning changes the plan, never the answer: tests/adaptive_result_equals_static.rs asserts the adaptive result equals both the static interpreter and ground-truth local engine eval across a genuine large-divergence switch.
  • Multi-source (union-arm) adaptive loop (sq-xw8zz): adaptive::execute_adaptive_multi_source lifts the adaptive path's MultiSource guard exactly as sq-7yf0 did for the static/streaming interpreters — a leaf's retained arms are fetched per source (bag-union re-keyed onto the pattern header; a failed arm fails the leaf CLOSED, never a silent arm-drop), each successful arm's wall-clock fetch latency is recorded under its SOURCE index via RuntimeStats::record_source_latency (a failed arm records nothing — a transport error's duration would make a fast-failing source look attractively fast), and the observed leaf cardinality is the UNION count. That makes the latency-aware cost bias above — slowest-arm default AND the opt-in CardinalityWeighted (sq-s5kd) — consumable from a LIVE run, not just planner-level tests, and is the live seam the deferred source-failover work needs. The final RuntimeStats is exposed on AdaptiveOutcome::stats (carry observations across queries / assert what the re-planner saw). tests/adaptive_multi_source_result_equals_local.rs holds the loop to the same merged-graph engine oracle as sq-7yf0, with two distinct live per-arm latencies.

With Phase 7 the 8-phase streaming federation client is feature-complete (Phases 0–7 all landed; epic sq-dnko closed). Multi-source UNION-per-leaf fan-out has since landed under epic sq-3183 (sq-7yf0, above; the ADAPTIVE loop's union-arm counterpart is sq-xw8zz, above). Still ahead as future beads under epic sq-3183: the pushed-down streaming bind-join, and the ANAPSID "adaptive operator" refinement (estimate a leaf's cardinality from a prefix of its rows while still streaming it). See crates/sparq-fedclient/README.md.

Test-quality note — the mutation ratchet is measured features-ON (sq-3dyje.6). The whole sparq-fedclient surface (and every test file) is #[cfg(feature = "fedclient")]-gated, so cargo-mutants MUST run it with --features fedclient,fedclient-adaptive — a features-off run builds an EMPTY crate and reports every mutant as a spurious survivor (the same per-crate quirk .github/workflows/ci.yml applies to sparq-canon's rdf12-triple-terms and sparq-prov's reason). The feature-on suite pins EXACT observable values — rendered pushdown sub-queries, error-variant Display strings, SRJ/Service-Description parse outputs, both SSRF is_forbidden_ip boundary tables, and native-transport observables driven over a raw in-process loopback TCP server (a configured timeout actually bounds a stalled request; a >3 MiB body round-trips under the real byte cap) — so a mutated return value is caught, not merely executed. A small residue of genuinely-equivalent mutants remains and is documented rather than papered over: in wire.rs the |↔^ bit-op mutations are all equivalent — the header / read_varint flag-set |=→^= (each distinct flag bit is set at most once from a zero start / LEB128 groups occupy non-overlapping shift windows) and the write_varint continuation-bit byte | 0x80→byte ^ 0x80 (byte is masked to & 0x7f, so bit 7 is provably clear), so XOR ≡ OR on every reachable input; EgressGuard::deny_private→Default::default() is equivalent because the struct derives Default and deny_private constructs exactly the empty allowlist the default does (its own doc says "Equivalent to EgressGuard::default"); the exclusive_groups union-find inner-loop bound (i + 1)..n→(i * 1)..n is equivalent (the extra j == i iteration does an idempotent self-union — same source, self-shares-var, union(i,i) is a no-op — so the group set is unchanged); and the push_group SubQuery-project-FIELD &&→|| (line ~486) is equivalent because both branches yield an empty Vec in the only case they differ (output_vars empty ⇒ project empty). The proj_clause && (line ~449) is NOT equivalent — it is killed by push_group_projection_clause_by_output_var_membership. Finally the ScatterPool Drop::drop→() is equivalent: after any drop body Rust drops the struct's fields, and the tx: Option<SyncSender> field-drop closes the channel exactly as the explicit self.tx.take() does, so detached workers still exit (the un-joined JoinHandles make drop order unobservable). (ScatterPool::join→() is a DIFFERENT method and is NOT equivalent — its blocking drain is killed by scatter_pool_join_blocks_until_all_jobs_complete.) These equivalents are noted, not asserted on — a test that "kills" an equivalent mutant would be vacuous.

Test-quality note — the mutation ratchet for sparq-fedplan is measured features-ON (sq-3dyje.7). The entire sparq-fedplan surface (and every #[cfg(test)] block) is #[cfg(feature = "fedplan")]-gated, so cargo-mutants MUST run with --features fedplan,adaptive-replan — a features-off run builds an empty crate and every mutant trivially survives (the committed 534-surviving / 0-caught baseline was this feature-OFF artefact, mirroring the sparq-fedclient / sparq-canon / sparq-prov quirk). With features ON the ratchet drops to 52 surviving / 415 caught (89 %). Six externally-authored tests in crates/sparq-fedplan/tests/mutation_kill_assertions.rs pin additional observable decisions: plan_bgp refuses an empty BGP with a non-empty selection slice (pattern.rs:114); diverges() requires STRICTLY greater than k × estimate in both directions — o == k*e and e == k*o.max(1) do NOT trigger (two >= mutants killed via AdaptiveExecutor::maybe_replan on a 3-arm star BGP after one advance(), ensuring remaining.len() >= 2 so the short-circuit guard is not hit); evict_stale(0.0) evicts sources with age > 0 without short-circuiting (< 0.0 guard fires only for negative max_age, killed by <= 0.0 mutant); and iri_eq correctly excludes non-scs:CharacteristicSet rdf:type triples from the char-set classification (descriptor.rs:431, killed by → true mutant). A genuinely-equivalent residue of 46 survivors is documented below — these are NOT papered over with assertions. Key categories: (1) spill/budget bookkeeping — StreamJoin::spill_key -=→+=//= and spilled()-path mutations are equivalent because the spill key is only used as a de-duplication marker and any consistent alteration per spill run behaves identically under the existing test load; (2) Tuple key encoding — > 7→>= 7/== 7/< 7 mutations are equivalent because the test data either stays in or out of the short encoding regime under all variants; (3) run_streaming early-exit comparisons — < capacity→>= capacity/> capacity/== capacity mutations survive because the bounded-result test queries always exhaust results before hitting the capacity edge; (4) cost-formula — plan.rs:263/:262/:259/:258 delete-arm / operator mutations survive because the test plans use symmetric estimates that produce the same greedy choice under both forms; (5) plan_suffix_greedy — several <→==/>/<= and delete-arm mutations in adaptive.rs:902–908 are equivalent under the star-BGP test fixture's symmetric or dominated cost ratios; (6) descriptor — char_set builder < → ==/<= boundary mutations survive because the test fixtures use counts that avoid the exact boundary; (7) is_subset +=→*= is equivalent when accumulator values are 0 or 1 (the test uses binary indicator sums); (8) suffix_cost_greedy *→+ in the cost product is equivalent when all cost components are 1.0 (unit estimates). The record_source_latency_after contains_key → true mutant is equivalent because fold_latency's Entry::Vacant arm seeds unconditionally (no alpha applied), so the guard's value on the first call for a fresh source is unobservable. These equivalents are noted, not asserted on. [SONNET-4.6]

Deferred (NOT here)

Mid-operator adaptivity (tearing down a join while it is producing output and resuming its half-built hash tables under a new algorithm) and live source failover (switching to a replica mid-stage when a source goes dark — observed latency now biases the join order toward faster sources, but hard failover needs the live multi-source execution layer this pure crate does not own) are out of scope. Filed as roadmap beads under epic sq-3183.

[OPUS-4.8] sq-a35t / sq-vf7q / sq-7s4z / sq-b51o / sq-7byx — flag for Fable re-review.