frs_candidates_pick: score + filter + dedup candidates per key (v0.31.0)

[NewGraphEnvironment]

Summary

Closes #207. Ships frs_candidates_pick() — fourth primitive in fresh's point-handling family (alongside frs_point_snap, frs_network_features, frs_point_match). Given a candidates table where multiple rows can share the same key, optionally score per row via a caller-supplied SQL expression, optionally filter via a caller-supplied WHERE clause, then keep one row per key via DISTINCT ON (col_key) ORDER BY ....

• Generic over any "score + filter + dedup per key" workflow where column-to-column comparisons disambiguate matches: stream-name match, watershed-group agreement, species-code overlap, assessment-date proximity, channel-width × stream-order compatibility, etc.
• Function signature follows the table_<role> / col_<role> / exp_<role> conventions (link/CLAUDE.md):

  frs_candidates_pick(
    conn, table_in, table_to, col_key,
    exp_score  = NULL,    # optional SQL fragment yielding a `score` column
    exp_filter = NULL,    # optional SQL WHERE clause for disqualifiers
    order_by              # vector of "expr ASC/DESC" strings; col_key prepended
  )

Live byte-identical parity vs bcfp

BULK PSCIS-to-stream selection: 102 / 102 ref picks identical, 0 missing. Closes the BULK 5-diff gap from fresh#206 at the dedup-step level.

Approach: stage bcfishpass.pscis_streams_150m (bcfp's pre-computed candidates with name_score, width_order_score, weighted_distance, multiple_match_ind) as input, call frs_candidates_pick with bcfp's exact filter + ORDER BY:

frs_candidates_pick(
  conn,
  table_in   = "fresh_test_207.candidates_bulk",
  table_to   = "fresh_test_207.picked_bulk",
  col_key    = "stream_crossing_id",
  exp_filter = "name_score != -100 AND width_order_score != -100 AND multiple_match_ind IS NULL",
  order_by   = c(
    "name_score DESC",
    "CASE WHEN modelled_xing_dist_instream IS NOT NULL
          THEN distance_to_stream - (distance_to_stream * 0.1)
          ELSE distance_to_stream END ASC"
  )
)

Diff vs bcfishpass.pscis (xref-excluded subset): ours 106 / ref 102 / identical 102 / only-ours 4 / only-ref 0. The 4 "extras" all live in bcfishpass.pscis_not_matched_to_streams — bcfp's downstream suspect_match filter (>50m distance) routes those to a separate table. That's a caller-level downstream filter, not a primitive responsibility.

Test plan

• 25 mocked tests covering validation (required args, identifier sanitization, exp_* non-empty character, reserved-column collision when exp_score set, existing-score-column-allowed-when-exp_score-NULL edge case) and SQL composition (DROP + CREATE, WITH scored AS CTE present/absent, SELECT DISTINCT ON (col_key), WHERE present/absent, ORDER BY containing col_key first then caller's clauses).
• Live byte-identical validation against bcfp BULK PSCIS dedup — 102/102.
• Full suite: 1012 PASS / 2 FAIL / 3 WARN — both FAIL and the WARNs pre-existing on main (same set as frs_point_match: match two point datasets along FWA network within instream distance #206 v0.30.0).
• R CMD check --no-tests: 0 errors / 4 warnings / 4 notes — identical to main.
• lintr::lint("R/frs_candidates_pick.R"): zero lints.
• DESCRIPTION 0.30.0 → 0.31.0 (minor bump — new exported function).
• NEWS.md 0.31.0 entry covers semantics, composition, parity numbers, and link#154 as the downstream consumer.

Composition story

With #206 + #207 shipped, the bcfp PSCIS-build pipeline reproduces byte-identically via composition:

fresh::frs_point_snap(conn, ..., num_features = 5)        # multi-stream candidates per PSCIS
fresh::frs_candidates_pick(conn, ..., exp_score = ..., order_by = ...)  # pick best stream
fresh::frs_point_match(conn, ..., distance_max = 100, tiebreak = "planar")  # match to modelled

This three-step chain is exactly what link#154 wires into lnk_pipeline_crossings to close the BULK mapping_code parity gap.

🤖 Generated with Claude Code