///|
/// Sorted-array query evaluation.
///
/// Every posting list is doc-sorted, so a query's result can be carried as
/// two parallel arrays in ascending doc order instead of a
/// `HashMap[doc, score]`. That removes four hash probes per posting from the
/// hot path and makes And/Or/Not plain linear merges; the previous
/// map-based evaluator spent most of a saturated query inside hashing.
///
/// `DocScores` is strictly ascending in `docs` with no duplicates: the
/// invariants every operation below must preserve.

///|
/// Evaluation result: ascending docs with their accumulated scores.
pub struct DocScores {
  docs : Array[Int32]
  scores : Array[Double]
}

///|
pub fn DocScores::empty() -> DocScores {
  { docs: [], scores: [], }
}

///|
pub fn DocScores::length(self : DocScores) -> Int32 {
  self.docs.length()
}

///|
pub fn DocScores::is_empty(self : DocScores) -> Bool {
  self.docs.is_empty()
}

///|
/// Build from unsorted (doc, score) pairs: sorts by doc and sums duplicates.
pub fn DocScores::from_pairs(pairs : Array[(Int32, Double)]) -> DocScores {
  if pairs.is_empty() {
    return DocScores::empty()
  }
  pairs.sort_by(fn(a, b) {
    let c = a.0.compare(b.0)
    if c == 0 {
      a.1.compare(b.1)
    } else {
      c
    }
  })
  let docs : Array[Int32] = []
  let scores : Array[Double] = []
  for pair in pairs {
    let (doc, score) = pair
    if docs.length() > 0 && docs[docs.length() - 1] == doc {
      scores[scores.length() - 1] += score
    } else {
      docs.push(doc)
      scores.push(score)
    }
  }
  { docs, scores, }
}

///|
/// Append a posting whose doc is strictly greater than the last one.
pub fn DocScores::push(self : DocScores, doc : Int32, score : Double) -> Unit {
  self.docs.push(doc)
  self.scores.push(score)
}

///|
/// Intersection: docs in both, scores summed.
pub fn DocScores::inter(a : DocScores, b : DocScores) -> DocScores {
  let out = DocScores::empty()
  let mut i = 0
  let mut j = 0
  while i < a.docs.length() && j < b.docs.length() {
    let da = a.docs[i]
    let db = b.docs[j]
    if da == db {
      out.push(da, a.scores[i] + b.scores[j])
      i += 1
      j += 1
    } else if da < db {
      i += 1
    } else {
      j += 1
    }
  }
  out
}

///|
/// Union: every doc of either, scores summed on collision.
pub fn DocScores::union(a : DocScores, b : DocScores) -> DocScores {
  let out = DocScores::empty()
  let mut i = 0
  let mut j = 0
  while i < a.docs.length() && j < b.docs.length() {
    let da = a.docs[i]
    let db = b.docs[j]
    if da == db {
      out.push(da, a.scores[i] + b.scores[j])
      i += 1
      j += 1
    } else if da < db {
      out.push(da, a.scores[i])
      i += 1
    } else {
      out.push(db, b.scores[j])
      j += 1
    }
  }
  while i < a.docs.length() {
    out.push(a.docs[i], a.scores[i])
    i += 1
  }
  while j < b.docs.length() {
    out.push(b.docs[j], b.scores[j])
    j += 1
  }
  out
}

///|
/// Difference: docs of `a` not present in `b` (sorted set), scores kept.
pub fn DocScores::diff(a : DocScores, b : DocScores) -> DocScores {
  let out = DocScores::empty()
  let mut j = 0
  for i in 0.. Bool,
) -> DocScores {
  let docs : Array[Int32] = []
  let scores : Array[Double] = []
  let mut j = 0
  for num in 0.. Array[Hit] {
  let n = scored.docs.length()
  if n == 0 {
    return []
  }
  if limit <= 0 || n <= limit {
    let hits : Array[Hit] = []
    for i in 0.. worst.score {
        replace_worst(top, { id: self.doc_ids[scored.docs[i]], score, })
      } else if score == worst.score {
        let hit : Hit = { id: self.doc_ids[scored.docs[i]], score, }
        if hit_before_lex(hit, worst) {
          replace_worst(top, hit)
        }
      }
    }
  }
  top
}

///|
/// Ordering used for result ranking: higher score first, then smaller id in
/// lexicographic (not shortlex) order.
fn hit_before_lex(a : Hit, b : Hit) -> Bool {
  if a.score > b.score {
    true
  } else if a.score < b.score {
    false
  } else {
    a.id.lexical_compare(b.id) < 0
  }
}

///|
/// Insert into a best-first array of at most `limit` elements.
fn insert_best_first(top : Array[Hit], hit : Hit) -> Unit {
  let mut i = top.length() - 1
  top.push(hit)
  while i >= 0 && hit_before_lex(hit, top[i]) {
    top[i + 1] = top[i]
    i -= 1
  }
  top[i + 1] = hit
}

///|
/// Replace the worst (last) element of a best-first array and bubble the new
/// element up to its place.
fn replace_worst(top : Array[Hit], hit : Hit) -> Unit {
  let mut i = top.length() - 2
  while i >= 0 && hit_before_lex(hit, top[i]) {
    top[i + 1] = top[i]
    i -= 1
  }
  top[i + 1] = hit
}

///|
/// Offer a candidate to a best-first top-`limit` array, materializing the `Hit`
/// (with its id lookup) only when it can actually enter. A saturated query
/// scores thousands of candidates to keep ten; building a `Hit` for every one of
/// them was the same waste the bounded selection in `collect_hits` avoids.
fn offer(top : Array[Hit], score : Double, id : String, limit : Int32) -> Unit {
  if top.length() < limit {
    insert_best_first(top, { id, score, })
    return
  }
  let worst = top[top.length() - 1]
  if score > worst.score {
    replace_worst(top, { id, score, })
  } else if score == worst.score && id.lexical_compare(worst.id) < 0 {
    replace_worst(top, { id, score, })
  }
}

///|
/// A block-max top-k run, with the work it did: postings scored and postings
/// skipped because their whole block could not reach the threshold. The
/// counters exist so the optimization is measured, not assumed.
pub(all) struct BlockMaxResult {
  hits : Array[Hit]
  scored : Int32
  skipped : Int32
} derive(Eq, Debug)

///|
pub extend BlockMaxResult with Eq::{equal, not_equal}

///|
pub extend BlockMaxResult with @debug.Debug::{to_repr}

///|
/// One term of a conjunction, prepared for the block-max walk: its posting list
/// per field, the BM25 context for each, and a cursor per field.
priv struct TermCursor {
  name : String
  fields : Array[String]
  ctxs : Array[Bm25Context]
  lists : Array[Array[Posting]]
  curs : Array[Int32]
}

///|
fn Index::term_cursor_prepare(
  self : Index,
  term : String,
  field : String?,
) -> TermCursor {
  let fields = match field {
    Some(f) => [f]
    None => self.field_names()
  }
  let ctxs : Array[Bm25Context] = []
  let lists : Array[Array[Posting]] = []
  for f in fields {
    ctxs.push(self.bm25_context(f, term))
    lists.push(self.postings_of(f, term))
  }
  { name: term, fields, ctxs, lists, curs: Array::make(fields.length(), 0), }
}

///|
/// True when the term can match nothing: every field it would be read in has
/// either no postings or no usable statistics. The exact evaluator produces an
/// empty result in that case, so a conjunction containing such a term is empty.
fn TermCursor::is_empty(self : TermCursor) -> Bool {
  for i in 0.. Int32 {
  let mut n = 0
  for list in self.lists {
    n += list.length()
  }
  n
}

///|
/// The smallest document number any of this term's field cursors is sitting on.
fn TermCursor::peek(self : TermCursor) -> Int32? {
  let mut best : Int32? = None
  for i in 0.. if d < b { best = Some(d) }
        None => best = Some(d)
      }
    }
  }
  best
}

///|
/// Move every field cursor past the document `peek` just returned, dropping
/// duplicate occurrences of the same document across fields.
fn TermCursor::advance(self : TermCursor) -> Unit {
  let d = match self.peek() {
    Some(d) => d
    None => return
  }
  for i in 0.. (Bool, Double) {
  if self.lists.length() == 1 {
    // One field is the common case and needs neither the contribution array nor
    // the sort: a union walks a document per term, and allocating per document
    // was the dominant cost of the disjunction walk.
    let list = self.lists[0]
    let mut c = self.curs[0]
    while c < list.length() && list[c].doc < doc {
      c += 1
    }
    self.curs[0] = c
    if c < list.length() && list[c].doc == doc && self.ctxs[0].valid {
      return (true, self.ctxs[0].score(list[c]))
    }
    return (false, 0.0)
  }
  let contribs : Array[Double] = []
  for i in 0.. Bool {
  for i in 0.. Int32 {
  let mut n = 0
  for i in 0.. Int32? {
  let mut best : Int32? = None
  for t in cursors {
    match t.peek() {
      Some(d) =>
        match best {
          Some(b) => if d < b { best = Some(d) }
          None => best = Some(d)
        }
      None => ()
    }
  }
  best
}

///|
/// Top-k for a conjunction of terms (`a b`, `a AND b`, with each term possibly
/// field-scoped), by a leapfrog walk over the shortest term's postings plus the
/// same block-max pruning the single-term path uses.
///
/// Two things make it cheap and exact:
///
/// * **The shortest list drives.** Only documents of the most selective term can
///   be matches, so the walk visits them and asks the other terms whether they
///   occur at that document. A saturated term is never enumerated.
/// * **A block whose bound cannot reach the threshold is skipped.** The bound is
///   the sum, over the terms, of each term's own block maximum — a document in
///   the block must be in every term, so its total score cannot exceed that sum.
///   The test is strict, so a document that only ties is still scored and can
///   still win the id tiebreak.
///
/// Scores are summed in the query's term order, which is the order the exact
/// evaluator's nested intersections sum in, so the result is bit-for-bit the
/// full evaluation.
pub fn Index::and_terms_topk(
  self : Index,
  terms : Array[(String, String?)],
  limit : Int32,
) -> BlockMaxResult {
  let n_docs = self.doc_ids.length()
  if limit <= 0 || n_docs == 0 || terms.is_empty() {
    return { hits: [], scored: 0, skipped: 0, }
  }
  let prepared : Array[TermCursor] = []
  for pair in terms {
    prepared.push(self.term_cursor_prepare(pair.0, pair.1))
  }
  for t in prepared {
    if t.is_empty() {
      return { hits: [], scored: 0, skipped: 0, }
    }
  }
  // The most selective term drives the walk.
  let mut driver = 0
  let mut driver_count = 0
  for i in 0.. {
        done = true
        0
      }
      Some(d) => d
    }
    if done {
      continue
    }
    let b = d / BLOCK_SIZE
    // The bound only changes when the walk enters a new block; compute it once
    // per block rather than once per driver document.
    if b != last_block {
      last_block = b
      block_upper = 0.0
      for i in 0.. 0 {
            block_upper += prepared[i].ctxs[j].block_bound(stats_all[i][j][b])
          }
        }
      }
    }
    if top.length() >= limit &&
      b < n_blocks &&
      block_upper < top[top.length() - 1].score {
      // Every driver document in this block is dominated: skip the block.
      let hi = (b + 1) * BLOCK_SIZE
      let mut more = true
      while more {
        match prepared[driver].peek() {
          Some(dd) =>
            if dd < hi {
              prepared[driver].advance()
              skipped += 1
            } else {
              more = false
            }
          None => more = false
        }
      }
      continue
    }
    // Score the candidate against every term, then step past it.
    let mut total = 0.0
    let mut present = true
    for i in 0.. Bool {
  let live = self.doc_count()
  if live == 0 {
    return false
  }
  let mut min_df = -1
  let mut max_df = 0
  for pair in terms {
    let mut df = 0
    match pair.1 {
      Some(f) => df = self.postings_of(f, pair.0).length()
      None =>
        for f in self.field_names() {
          df += self.postings_of(f, pair.0).length()
        }
    }
    if min_df < 0 || df < min_df {
      min_df = df
    }
    if df > max_df {
      max_df = df
    }
  }
  // A dense operand to set the threshold, and one far rarer to create the gap.
  max_df * 10 > live && min_df * 10 < max_df
}

///|
/// Top-k for a disjunction of terms (`a OR b`), by the same block-max idea as the
/// conjunction: walk the union in document order and skip a whole doc block when
/// the sum of the terms' block maxima cannot reach the current k-th best.
///
/// A document's score is the sum of the terms that occur in it, accumulated in
/// the query's term order — the order the exact evaluator's nested unions add
/// in — so the result is bit-for-bit the full evaluation. Terms that match
/// nothing are dropped (an empty operand of a union contributes nothing).
///
/// `search` routes here through `or_walk_pays`, which is what keeps the walk off
/// the shapes where it loses; the numbers behind that gate are on that function.
pub fn Index::or_terms_topk(
  self : Index,
  terms : Array[(String, String?)],
  limit : Int32,
) -> BlockMaxResult {
  let n_docs = self.doc_ids.length()
  if limit <= 0 || n_docs == 0 || terms.is_empty() {
    return { hits: [], scored: 0, skipped: 0, }
  }
  let active : Array[TermCursor] = []
  for pair in terms {
    let cursor = self.term_cursor_prepare(pair.0, pair.1)
    if !cursor.is_empty() {
      active.push(cursor)
    }
  }
  if active.is_empty() {
    return { hits: [], scored: 0, skipped: 0, }
  }
  let n_blocks = (n_docs + BLOCK_SIZE - 1) / BLOCK_SIZE
  let stats_all : Array[Array[Array[BlockStat]]] = []
  for t in active {
    let per_field : Array[Array[BlockStat]] = []
    for i in 0.. {
        done = true
        0
      }
      Some(d) => d
    }
    if done {
      continue
    }
    let b = d / BLOCK_SIZE
    // The bound only changes when the walk enters a new block.
    if b != last_block {
      last_block = b
      block_upper = 0.0
      for i in 0.. 0 {
            block_upper += active[i].ctxs[j].block_bound(stats_all[i][j][b])
          }
        }
      }
    }
    if top.length() >= limit &&
      b < n_blocks &&
      block_upper < top[top.length() - 1].score {
      // Every document in this block is dominated: step past the whole block.
      let hi = (b + 1) * BLOCK_SIZE
      for i in 0.. BlockMaxResult {
  let fields = match field {
    Some(f) => [f]
    None => self.field_names()
  }
  let n_docs = self.doc_ids.length()
  if limit <= 0 || n_docs == 0 || fields.is_empty() {
    return { hits: [], scored: 0, skipped: 0, }
  }
  let n_blocks = (n_docs + BLOCK_SIZE - 1) / BLOCK_SIZE
  let any_deleted = !self.deleted.is_empty()
  let ctxs : Array[Bm25Context] = []
  let lists : Array[Array[Posting]] = []
  let stats_all : Array[Array[BlockStat]] = []
  for f in fields {
    ctxs.push(self.bm25_context(f, term))
    lists.push(self.postings_of(f, term))
    stats_all.push(self.term_block_stats(f, term))
  }
  let curs = Array::make(fields.length(), 0)
  let top : Array[Hit] = []
  let mut scored = 0
  let mut skipped = 0
  for b in 0.. 0 {
        upper += ctxs[i].block_bound(stats_all[i][b])
      }
    }
    // Skippable only once the top set is full: before that there is no
    // threshold to skip against.
    if top.length() >= limit && upper < top[top.length() - 1].score {
      for i in 0..