(*---------------------------------------------------------------------------
  Copyright (c) 2026 Thomas Gazagnaire. All rights reserved.
  SPDX-License-Identifier: ISC
  ---------------------------------------------------------------------------*)

type layout = (string * string list) list

type key_source =
  | Const of Ast.expr
  | Outer of int * int
  | In_list of Ast.expr list

type path = {
  descr : string;
  unique : bool;
  bound : int;
  rows_per_key : int option;
  covering : bool;
}

type access =
  | Scan
  | Seek of { table : string; path : path; keys : (string * key_source) list }
  | Probe of {
      key : int;
      src : int;
      pos : int;
      coll : Catalog.Collation.t option;
      rest : (int * int * int * Catalog.Collation.t option) list;
      spill : access option;
      filter : (string * Ast.expr) option;
    }
  | Pair of {
      key : int;
      other : int;
      other_key : int;
      coll : Catalog.Collation.t option;
    }

type affinity = [ `Blob | `Integer | `Numeric | `Real | `Text ]

type facts = {
  columns : string -> string list;
  seekable : Ast.table_ref -> string option;
  best_index : string -> (string * key_source) list -> path option;
  affinity : string -> string -> affinity option;
  row_count : string -> upto:int -> int;
  seek_answers_rowid : bool;
  const_subquery : bool;
  seek_lists : bool;
}

type t = { accesses : access list; order : int list }

let ident_equal = Ast.ident_equal
let mem_ident col cols = List.exists (ident_equal col) cols

(* ── Reading the query ───────────────────────────────────────── *)

(* Every position of [layout] a reference to [col] under [qual] could name: a
   qualifier restricts the search to the source it names, and each source is
   asked for a column of that name. *)
let column_positions (layout : layout) qual col =
  List.filteri
    (fun i _ ->
      match qual with
      | Some q -> ident_equal (fst (List.nth layout i)) q
      | None -> true)
    (List.mapi
       (fun i (_, cols) ->
         Option.map (fun j -> (i, j)) (List.find_index (ident_equal col) cols))
       layout)
  |> List.filter_map Fun.id

(* The scan position a column reference names, or [None] when it names no single
   one. A qualifier picks its source outright; a bare name is resolved against
   every source's columns and accepted only when exactly one has it, since
   resolving it against the wrong frame would silently drop rows. *)
let column_position (layout : layout) qual col =
  match column_positions layout qual col with [ p ] -> Some p | _ -> None

(* [column_position] for a seek key: a name no source declares that spells the
   rowid reaches the hidden rowid slot, as {!read_columns} charges it, so the
   rowid is sought under any of its spellings. *)
let key_position (layout : layout) qual col =
  match column_positions layout qual col with
  | [ p ] -> Some p
  | [] when Ast.is_rowid_name col ->
      column_position layout qual Ast.rowid_sentinel
  | _ -> None

(* What a query reads from each source. Charged from every position a reference
   could name rather than the one it does: this decides whether a cursor can be
   answered out of an index, so a reference nothing here can attribute has to be
   assumed of each source that could answer it. *)
let read_columns (layout : layout) (s : Ast.select) =
  let n = List.length layout in
  let acc = Array.make n [] in
  let add (i, j) = if not (List.mem j acc.(i)) then acc.(i) <- j :: acc.(i) in
  (* [*] never returns the hidden rowid slot, so it does not ask for it *)
  let whole i =
    List.iteri
      (fun j c -> if not (String.equal c Ast.rowid_sentinel) then add (i, j))
      (snd (List.nth layout i))
  in
  let qualified q =
    List.iteri (fun i (a, _) -> if ident_equal a q then whole i) layout
  in
  let col _ qual name =
    match column_positions layout qual name with
    | [] ->
        (* The declared columns are consulted first, so a real column named
           [rowid] shadows the implicit one -- and the hidden slot's own name
           cannot be written, so nothing else reaches it. *)
        if Ast.is_rowid_name name then
          List.iter add (column_positions layout qual Ast.rowid_sentinel)
    | ps -> List.iter add ps
  in
  (* [*] and [t.*] name no column, so the walk does not report them; they are
     read off each select's own result list instead, which [nest] collects on
     the way past. A [*] standing as [count]'s argument is not one of these: it
     is an argument, not a result column, and it reads nothing. *)
  let nested = ref [] in
  let nest ctx u =
    nested := u :: !nested;
    ctx
  in
  Ast.walk_select
    ~table:(fun _ _ -> ())
    ~col ~nest
    ~cte:(fun ctx c -> Some (nest ctx c.Ast.query))
    () s;
  List.iter
    (fun (u : Ast.select) ->
      List.iter
        (fun (rc : Ast.result_col) ->
          match rc.Ast.expr with
          | Ast.Star -> List.iteri (fun i _ -> whole i) layout
          | Ast.Qualified_star q -> qualified q
          | _ -> ())
        u.Ast.cols)
    (s :: !nested);
  Array.map (List.sort_uniq compare) acc

let seek_keys ~const (layout : layout) ~earlier i e =
  let mine qual col =
    match key_position layout qual col with
    | Some (k, _) when k = i -> true
    | _ -> false
  in
  (* [c] is a column of the source being planned, [other] what it is equated to;
     [None] when [other] is not a key this source can seek on. *)
  let key c other =
    match other with
    | Ast.Col (q, oc) -> (
        match key_position layout q oc with
        | Some (k, j) when earlier k -> Some (c, Outer (k, j))
        | _ -> None)
    | e when const e -> Some (c, Const e)
    | _ -> None
  in
  let rec go acc = function
    | Ast.Binary (Ast.And, a, b) -> go (go acc a) b
    | Ast.Binary (Ast.Eq, x, y) -> (
        match (x, y) with
        | Ast.Col (q, c), other when mine q c -> (
            match key c other with Some k -> k :: acc | None -> acc)
        | other, Ast.Col (q, c) when mine q c -> (
            match key c other with Some k -> k :: acc | None -> acc)
        | _ -> acc)
    | Ast.In_list (Ast.Col (q, c), values)
      when mine q c && List.for_all const values ->
        (c, In_list values) :: acc
    | _ -> acc
  in
  go [] e

(* The equality classes the top-level AND conjuncts of [preds] sort the scan
   positions into, and the constant each position is directly equated to. Two
   positions are linked only where [comparable] holds. A class is merged one
   link at a time: a join names a handful of columns, so these lists stay
   shorter than the union-find that would replace them. *)
let equality_classes (layout : layout) ~comparable preds =
  let rec conjuncts acc = function
    | Ast.Binary (Ast.And, a, b) -> conjuncts (conjuncts acc a) b
    | e -> e :: acc
  in
  let const = function (Ast.Lit _ | Ast.Param _) as e -> Some e | _ -> None in
  let pos = function
    | Ast.Col (q, c) -> column_position layout q c
    | _ -> None
  in
  let links = ref [] and consts = ref [] in
  let note x y =
    match (pos x, pos y) with
    | Some p, Some q -> if comparable p q then links := (p, q) :: !links
    | Some p, None -> (
        match const y with Some v -> consts := (p, v) :: !consts | None -> ())
    | None, (Some _ | None) -> ()
  in
  List.iter
    (function
      | Ast.Binary (Ast.Eq, x, y) ->
          note x y;
          note y x
      | _ -> ())
    (List.fold_left conjuncts [] preds);
  let merge classes (p, q) =
    let hit, rest =
      List.partition (fun cl -> List.mem p cl || List.mem q cl) classes
    in
    List.sort_uniq compare (p :: q :: List.concat hit) :: rest
  in
  (List.fold_left merge [] !links, !consts)

let equality_closure (layout : layout) ~affinity preds =
  let comparable p q =
    match (affinity p, affinity q) with
    | Some a, Some b -> a = b
    | None, _ | _, None -> false
  in
  let classes, consts = equality_classes layout ~comparable preds in
  let column (i, j) =
    let alias, cols = List.nth layout i in
    Ast.Col (Some alias, List.nth cols j)
  in
  List.concat_map
    (fun cl ->
      (* two constants in one class contradict each other, so the query has no
         rows and either narrows to none of them; the first will do *)
      match List.find_map (fun p -> List.assoc_opt p consts) cl with
      | None -> []
      | Some v ->
          List.filter_map
            (fun p ->
              if List.mem_assoc p consts then None
              else Some (Ast.Binary (Ast.Eq, column p, v)))
            cl)
    classes

(* Every equality of [e] between a column of the source at [i] and one of an
   earlier source, as (inner column, outer source, outer column, collation), in
   the order [e] states them. *)
let probe_keys (layout : layout) ~earlier i e =
  let column = function
    | Ast.Col (q, c) -> Some (q, c, None)
    | Ast.Collate (Ast.Col (q, c), coll) ->
        Option.map
          (fun cl -> (q, c, Some cl))
          (Catalog.Collation.of_string coll)
    | _ -> None
  in
  let pair (q1, c1, k1) (q2, c2, k2) =
    (* at most one operand carries a wrapper, and it names the whole
       comparison's sequence *)
    let coll = if k1 = None then k2 else k1 in
    match (column_position layout q1 c1, column_position layout q2 c2) with
    | Some (k, key), Some (src, pos) when k = i && earlier src ->
        Some (key, src, pos, coll)
    | _ -> None
  in
  let rec go = function
    | Ast.Binary (Ast.And, a, b) -> go a @ go b
    | Ast.Binary (Ast.Eq, x, y) -> (
        match (column x, column y) with
        | Some a, Some b -> (
            match pair a b with
            | Some k -> [ k ]
            | None -> Option.to_list (pair b a))
        | _ -> [])
    | _ -> []
  in
  go e

(* The hash on every equality [preds] offer the source at [i]: the first is the
   key the probe is built on, and each further one narrows its buckets. An
   equality stated twice is one key. *)
(* The conjuncts of [preds] that read columns of the source at [i] and no
   other, with no subquery, no window and no call whose answer could change
   between two evaluations: the rows of that source a build may drop, as the
   joined row's WHERE would. *)
let local_filter (layout : layout) i preds =
  let rec conjuncts acc = function
    | Ast.Binary (Ast.And, a, b) -> conjuncts (conjuncts acc a) b
    | e -> e :: acc
  in
  let reads_here e =
    let ok = ref true and seen = ref false in
    let rec go e =
      match e with
      | Ast.Col (q, c) -> (
          match key_position layout q c with
          | Some (k, _) when k = i -> seen := true
          | _ -> ok := false)
      | Ast.Func (n, args)
        when Func.deterministic n
             && not (Func.aggregate n ~nargs:(List.length args)) ->
          List.iter go args
      | Ast.Func _ | Ast.Window _ | Ast.Raise _ | Ast.Distinct_agg _
      | Ast.Filter _ | Ast.Ordered_agg _ ->
          ok := false
      | e ->
          Ast.fold_expr
            ~select:(fun () _ -> ok := false)
            (fun () x -> go x)
            () e
    in
    go e;
    !ok && !seen
  in
  match
    List.filter reads_here (List.rev (List.fold_left conjuncts [] preds))
  with
  | [] -> None
  | e :: es ->
      Some (List.fold_left (fun a b -> Ast.Binary (Ast.And, a, b)) e es)

let probe_key ?(filter = None) layout ~earlier i preds =
  let keys =
    List.fold_left
      (fun acc k -> if List.mem k acc then acc else k :: acc)
      []
      (List.concat_map (probe_keys layout ~earlier i) preds)
  in
  match List.rev keys with
  | [] -> None
  | (key, src, pos, coll) :: rest ->
      Some (Probe { key; src; pos; coll; rest; spill = None; filter })

(* The index seek for the source at scan position [i], from the keys [preds]
   offer it, or [None] when it has no table to address or no index covers the
   keys. Each key column is folded onto the schema's own spelling, so a backend
   that matches names byte for byte still finds the column the query meant. *)
let seek_path facts ~const layout ~earlier i (tr : Ast.table_ref option) preds =
  Option.bind (Option.bind tr facts.seekable) (fun table ->
      let declared = facts.columns table in
      let spelled c =
        Option.value ~default:c (List.find_opt (ident_equal c) declared)
      in
      let keys =
        List.concat_map (seek_keys ~const layout ~earlier i) preds
        |> List.filter (function
          | _, In_list _ -> facts.seek_lists
          | _, (Const _ | Outer _) -> true)
        (* a column equated twice keeps the first key -- whichever case each
           equality spelled it in; the predicate still runs, so the second only
           narrows further *)
        |> List.fold_left
             (fun acc (c, k) ->
               if mem_ident c (List.map fst acc) then acc
               else acc @ [ (spelled c, k) ])
             []
      in
      Option.map
        (fun path -> Seek { table; path; keys })
        (facts.best_index table keys))

(* The outer kind {!Catalog.Ir} names this join by, [None] for an inner one. *)
let ir_kind = function
  | Ast.Inner -> None
  | Ast.Left -> Some Catalog.Ir.Left
  | Ast.Right -> Some Catalog.Ir.Right
  | Ast.Full -> Some Catalog.Ir.Full

let loop_order (s : Ast.select) k =
  if List.length s.Ast.joins <> k - 1 then List.init k Fun.id
  else
    Catalog.Ir.cursor_order
      (List.map
         (fun (j : Ast.join) ->
           { Catalog.Ir.kind = ir_kind j.Ast.type_; on = Catalog.Ir.Always })
         s.Ast.joins)

(* ── What each source costs ──────────────────────────────────── *)

let index_equality_rows = Catalog.Join.index_equality_rows

(* How far the rewinds a loop makes are tracked. {!Catalog.Join.max_rewinds}
   stops at 100 000, and a join driven by more rows than that was planned as
   if it had 100 000: the seek it then chose costs a descent per row, two
   million page fetches over 877 221 flows against 45 059 for the hash it
   declined. A count past 100 000 rows is an estimate off a table's interior
   pages, so tracking further costs those pages and not the table's. *)
let max_rewinds = 1 lsl 30

(* {!Catalog.Join.count_bound} under {!max_rewinds}: as far as a source is
   counted to decide its own access, and, where a source follows, to estimate
   how often that one is rewound. *)
let count_bound ~rewinds ~next =
  let decides = Catalog.Join.count_bound ~rewinds ~next:false in
  if next then max (max_rewinds / max 1 rewinds) decides else decides

(* How many keys a seek on [keys] looks up per rewind: the product of the
   lengths of its IN lists, saturating at {!max_rewinds}. *)
let key_count keys =
  List.fold_left
    (fun n (_, k) ->
      match k with
      | In_list es -> min max_rewinds (n * List.length es)
      | Const _ | Outer _ -> n)
    1 keys

(* The rows a source offers each time its cursor is rewound: a scan offers the
   whole relation, an access path offers what one key selects -- one row for a
   rowid or fully-bound unique index, what the backend counted where it can,
   sqlite3's default estimate otherwise -- for each key it looks up. *)
let rewind_yield access rows =
  match access with
  | Scan -> rows
  | Seek { path = { unique = true; _ }; keys; _ } -> min rows (key_count keys)
  | Seek { path = { rows_per_key = Some n; _ }; keys; _ } ->
      min rows (key_count keys * n)
  | Seek { keys; _ } -> min rows (key_count keys * index_equality_rows)
  | Probe _ | Pair _ -> min rows index_equality_rows

(* How often the loop rewinds the source after [i], given [i] reads through
   [access] and holds [inner_rows] as far as it was counted. *)
let rewinds_after ~access ~rewinds ~inner_rows =
  min max_rewinds (rewinds * rewind_yield access inner_rows)

let source_rows facts ~upto (tr : Ast.table_ref option) =
  match tr with Some tr -> facts.row_count tr.Ast.name ~upto | None -> upto

(* [tr] counted to [upto], first no further than [within], the bound
   {!Catalog.Join.count_bound} sets: a source smaller than that is the count it
   always was, and only a table the store addresses, whose count past it is an
   estimate off its interior pages, is counted further. A derived source, a
   view and a table the store cannot seek keep the bound they had. *)
let source_rows_past facts ~within ~upto tr =
  let n = source_rows facts ~upto:(min upto within) tr in
  let addressed = Option.bind tr facts.seekable <> None in
  if n < min upto within || upto <= within || not addressed then n
  else source_rows facts ~upto tr

(* The access a source after the first reads through, of the [seek] and the
   [hash] it is offered, for [inner_rows] read [rewinds] times. *)
let choose_access ~rewinds ~inner_rows seek hash =
  (* what one of this seek's keys selects, which is how much of the relation it
     re-reads on every rewind, and whether reading one of those rows costs the
     relation a second look *)
  let seek_yield =
    match seek with Some a -> rewind_yield a inner_rows | None -> 1
  in
  let covering =
    match seek with Some (Seek { path; _ }) -> path.covering | _ -> false
  in
  match
    Catalog.Join.choose ~rewinds ~inner_rows ~seek_yield ~covering
      ~seek:(seek <> None) ~hash:(hash <> None)
  with
  | Catalog.Join.Seek -> Option.get seek
  | Catalog.Join.Hash -> (
      (* the seek the ratio declined is where a build too large to hold goes,
         rather than back to the product *)
      match Option.get hash with
      | Probe p -> Probe { p with spill = seek }
      | a -> a)
  | Catalog.Join.Scan -> Scan

(* Counting a source reads it, so it is counted no further than an answer
   needs, and every answer is the one a count to [full] gives. The choice of an
   access reads the rows only through what a seek's key selects and the bound
   {!count_bound} sets for a bucket no wider than
   [index_equality_rows]; with no seek it does not read them at all. Where a
   source follows, the rewinds it causes read the rows only as far as one
   rewind offers -- the whole relation for a scan, what one key selects for a
   seek. A relation past those bounds moves nothing, so the outer side of a
   join pinned to one key no longer walks its table to be planned. *)
let nested_loop facts ~const layout ~order ~keys refs =
  let n = List.length refs in
  let accesses = Array.make n Scan in
  let seen = Array.make n false in
  let earlier k = k >= 0 && k < n && seen.(k) in
  let last = List.nth order (List.length order - 1) in
  let rewinds = ref 1 in
  List.iter
    (fun i ->
      let tr = List.nth refs i in
      let preds = keys i in
      let seek = seek_path facts ~const layout ~earlier i tr preds in
      let filter =
        match tr with
        | Some tr ->
            Option.map (fun e -> (tr.Ast.name, e)) (local_filter layout i preds)
        | None -> None
      in
      let hash = probe_key ~filter layout ~earlier i preds in
      let full = count_bound ~rewinds:!rewinds ~next:(i <> last) in
      let within =
        Catalog.Join.count_bound ~rewinds:!rewinds ~next:(i <> last)
      in
      let decides =
        match seek with
        | None -> count_bound ~rewinds:!rewinds ~next:false
        | Some a ->
            let yield = rewind_yield a max_int in
            if yield <= index_equality_rows then
              max yield (count_bound ~rewinds:!rewinds ~next:false)
            else within
      in
      let inner_rows = source_rows facts ~upto:decides tr in
      let a = choose_access ~rewinds:!rewinds ~inner_rows seek hash in
      accesses.(i) <- a;
      seen.(i) <- true;
      if i <> last then
        let upto = min full (rewind_yield a max_int) in
        let within = min within (rewind_yield a max_int) in
        let inner_rows =
          if upto > decides then source_rows_past facts ~within ~upto tr
          else inner_rows
        in
        rewinds := rewinds_after ~access:a ~rewinds:!rewinds ~inner_rows)
    order;
  Array.to_list accesses

(* Both sides of a two-cursor FULL join, hashed on the one equality its ON
   offers. Each side is the padded one of a pass, so neither may be narrowed by
   the WHERE and neither is the inner throughout; without an equality to hash on
   they scan, and the join costs the product. *)
let full_pair layout on =
  match probe_key layout ~earlier:(fun k -> k = 0) 1 [ on ] with
  | Some (Probe { key; pos; coll; _ }) ->
      [
        Pair { key = pos; other = 1; other_key = key; coll };
        Pair { key; other = 0; other_key = pos; coll };
      ]
  | _ -> [ Scan; Scan ]

(* ── The plan ────────────────────────────────────────────────── *)

(* Does [s] name the implicit rowid anywhere, under any alias? An index answers
   a key, not a row's identity, so a backend whose sought rows carry no rowid
   has to answer such a query from the relation. *)
let references_rowid (s : Ast.select) =
  let found = ref false in
  let col () _ name = if Ast.is_rowid_name name then found := true in
  Ast.walk_select
    ~table:(fun _ _ -> ())
    ~col
    ~nest:(fun ctx _ -> ctx)
    ~cte:(fun ctx _ -> Some ctx)
    () s;
  !found

(* The table each source of [s] addresses, in scan order, or [None] where it
   addresses none: a derived table, or a name the engine materialises for
   itself. The store does not have such a relation, and a table that happens to
   share its name is not it, so offering no reference leaves the source scanned
   and sized as the derived relation it is. *)
let sources ~bound (s : Ast.select) =
  let source (tr : Ast.table_ref) =
    if tr.Ast.subquery <> None || List.exists (ident_equal tr.Ast.name) bound
    then None
    else Some tr
  in
  (match s.Ast.from with Ast.Table tr -> source tr | _ -> None)
  :: List.map (fun (j : Ast.join) -> source j.Ast.table) s.Ast.joins

(* The select each source of [s] is a derived table of, in scan order, and
   [None] where the source is not one. A derived table has no schema to read,
   but its columns are the expressions of the select behind it, which is what
   sqlite3 types it from -- so this is what the affinity lookup follows. *)
let derived (s : Ast.select) =
  let of_ref (tr : Ast.table_ref) = tr.Ast.subquery in
  (match s.Ast.from with
  | Ast.Table tr -> of_ref tr
  | Ast.Select { fs_select; _ } -> Some fs_select
  | Ast.Unit | Ast.Values _ | Ast.Table_function _ -> None)
  :: List.map (fun (j : Ast.join) -> of_ref j.Ast.table) s.Ast.joins

let column_affinity ~affinity env s (layout : layout) =
  let refs = sources ~bound:[] s in
  let subs = derived s in
  fun (i, j) ->
    match List.nth_opt subs i with
    | Some (Some sub) -> Resolve.output_affinity env sub j
    | _ -> (
        match (List.nth_opt refs i, List.nth_opt layout i) with
        | Some (Some tr), Some (_, cols) ->
            Option.bind (List.nth_opt cols j) (affinity tr.Ast.name)
        | _ -> None)

(* Which predicates the source at each scan position may take keys from, and
   the order the loop enters the cursors, for a chain of [k] sources whose
   trailing join is [trailing]. [None] for a FULL join, whose two sides are
   planned together and neither of which may be narrowed at all. *)
let key_sets ~trailing ~k ~where ~ons =
  (* the ONs of the leading block, which are inner and so qualify the rows the
     trailing join preserves without ever padding one *)
  let leading_ons = List.filteri (fun i _ -> i < k - 2) ons in
  let forward = List.init k Fun.id in
  match trailing with
  | Ast.Inner -> Some (forward, fun _ -> where @ ons)
  | Ast.Left ->
      Some
        ( forward,
          fun i ->
            if i = k - 1 then [ List.nth ons (k - 2) ] else where @ leading_ons
        )
  | Ast.Right ->
      Some
        ( (k - 1) :: List.init (k - 1) Fun.id,
          fun i -> if i = k - 1 then where else ons )
  | Ast.Full -> None

let access_paths facts env ~bound (s : Ast.select) (layout : layout) =
  let where = Option.to_list s.Ast.where in
  let ons = List.map (fun (j : Ast.join) -> j.Ast.on) s.joins in
  let refs = sources ~bound s in
  let k = List.length layout in
  let scans = List.map (fun _ -> Scan) layout in
  let kinds = List.map (fun (j : Ast.join) -> j.Ast.type_) s.joins in
  (* the last join's kind, and whether every join before it is inner *)
  let trailing, leading_all_inner =
    match List.rev kinds with
    | [] -> (Ast.Inner, true)
    | last :: lead -> (last, List.for_all (fun kd -> kd = Ast.Inner) lead)
  in
  let affinity = column_affinity ~affinity:facts.affinity env s layout in
  (* A closed subquery yields one value before the scan starts, which is all a
     lookup key has to be -- and sqlite3 plans that shape as a seek. Whether the
     backend can reach the value there is its own answer. *)
  let const = function
    | Ast.Lit _ | Ast.Param _ -> true
    | Ast.Scalar_select sub ->
        facts.const_subquery && not (Resolve.correlated env [] sub)
    | _ -> false
  in
  let facts =
    if facts.seek_answers_rowid || not (references_rowid s) then facts
    else { facts with seekable = (fun _ -> None) }
  in
  let nested ~order ~keys =
    let keys i =
      let preds = keys i in
      preds @ equality_closure layout ~affinity preds
    in
    nested_loop facts ~const layout ~order ~keys refs
  in
  (* a NATURAL / USING join's equality is implicit, and the layout it publishes
     hides the duplicated columns; leave those to the plain scan *)
  if
    List.exists
      (fun (j : Ast.join) -> j.Ast.natural || j.Ast.using <> [])
      s.joins
    || List.length refs <> k
    || not leading_all_inner
  then scans
  else
    match key_sets ~trailing ~k ~where ~ons with
    | Some (order, keys) -> nested ~order ~keys
    | None -> if k = 2 then full_pair layout (List.hd ons) else scans

let v facts env ~bound s layout =
  {
    accesses = access_paths facts env ~bound s layout;
    order = loop_order s (List.length layout);
  }

let pp_key_source ppf = function
  | Const e -> Fmt.string ppf (Render.expr e)
  | Outer (i, j) -> Fmt.pf ppf "%d.%d" i j
  | In_list es ->
      Fmt.pf ppf "(%s)" (String.concat ", " (List.map Render.expr es))

let pp_access ppf = function
  | Scan -> Fmt.string ppf "scan"
  | Seek { table; path; keys } ->
      Fmt.pf ppf "seek %s using %s on %a" table path.descr
        Fmt.(list ~sep:(any ", ") (pair ~sep:(any "=") string pp_key_source))
        keys
  | Probe { key; src; pos; rest; _ } ->
      Fmt.pf ppf "probe %d on %d.%d" key src pos;
      List.iter
        (fun (key, src, pos, _) -> Fmt.pf ppf ", %d on %d.%d" key src pos)
        rest
  | Pair { key; other; other_key; _ } ->
      Fmt.pf ppf "pair %d with %d.%d" key other other_key

let pp ppf t =
  Fmt.pf ppf "@[<v>%a@,order: %a@]"
    Fmt.(list ~sep:cut (fun ppf (i, a) -> Fmt.pf ppf "%d: %a" i pp_access a))
    (List.mapi (fun i a -> (i, a)) t.accesses)
    Fmt.(list ~sep:(any " ") int)
    t.order

let explain t ~names =
  let line what access =
    match (what, access) with
    | `Other d, _ -> "SCAN " ^ d
    | `Table n, (Scan | Probe _ | Pair _) -> "SCAN " ^ n
    | `Table n, Seek { path; _ } -> Fmt.str "SEARCH %s USING %s" n path.descr
  in
  String.concat "\n" (List.map2 line names t.accesses)

(* The alias and column names each source publishes, enough to decide an access
   path without materialising anything: an access path needs them only to match
   a qualified column against the source it belongs to. *)
let layout env ~bound (s : Ast.select) =
  let alias (tr : Ast.table_ref) =
    Option.value tr.Ast.alias ~default:tr.Ast.name
  in
  let cols (tr : Ast.table_ref) =
    match tr.Ast.subquery with
    | Some q -> Resolve.select_columns env q
    | None -> (
        match List.find_opt (fun (n, _) -> ident_equal n tr.Ast.name) bound with
        | Some (_, cs) -> cs
        | None -> env.Resolve.columns tr.Ast.name)
  in
  let leading =
    match s.Ast.from with
    | Ast.Table tr -> (alias tr, cols tr)
    | Ast.Select { fs_select; fs_alias } ->
        (Option.value fs_alias ~default:"", Resolve.select_columns env fs_select)
    | Ast.Table_function { tf_name; tf_alias; _ } ->
        ( Option.value tf_alias ~default:tf_name,
          env.Resolve.table_function_columns tf_name )
    | Ast.Values rows ->
        let n = match rows with r :: _ -> List.length r | [] -> 0 in
        ("", List.init n (fun i -> Fmt.str "column%d" (i + 1)))
    | Ast.Unit -> ("", [])
  in
  leading
  :: List.map
       (fun (j : Ast.join) -> (alias j.Ast.table, cols j.Ast.table))
       s.joins