// Swapping the last build side of a chain: when the rows the earlier steps
// produce are few, they become the hash side and the last table streams past
// them, so a large table named last is never held whole. The contract is in
// docs/ROADMAP.md ("The build side swap contract").

///|
/// The last join step of a chain with its roles swapped: the rows the earlier
/// steps produced are the build side, held whole behind an index, and the step's
/// own table is the probe side, arriving a batch at a time. Pairs leave in the
/// step's combined layout (prefix columns, then the table's) in batches of at
/// most `JOIN_OUT_BATCH`. Only INNER steps are swapped: a LEFT or FULL step owes
/// its unmatched rows to the side that would now be the build side. NULL keys
/// never match and the residual filters candidate pairs, as in `JoinCursor`.
priv struct SwapCursor {
  step : JoinStep
  prefix : @types.DataBatch
  prefix_keys : Array[@types.ColumnVector]
  mut int_index : IntKeyIndex
  str_map : Map[String, Array[Int32]]
  mut fast : Bool
  left_width : Int32
  next_right : () -> @types.DataBatch? raise @types.SqlError
  rplan : ResidualPlan?
  rleft_t : Array[FixedArray[Bool]]
  rleft_u : Array[FixedArray[Bool]]
  mut right_now : @types.DataBatch?
  mut out_p : Array[Int32] // pending prefix rows
  mut out_t : Array[Int32] // pending table rows, aligned with out_p
  mut out_at : Int32
  mut primed : Bool
  mut done : Bool
}

///|
fn SwapCursor::open(
  step : JoinStep,
  prefix : @types.DataBatch,
  next_right : () -> @types.DataBatch? raise @types.SqlError,
) -> SwapCursor raise @types.SqlError {
  let prefix_keys : Array[@types.ColumnVector] = []
  if prefix.row_count() > 0 {
    for e in step.keys_l {
      prefix_keys.push(eval_vec(e, prefix))
    }
  }
  // the residual's prefix half is the same for every table batch: evaluating it
  // once here is the difference between one pass over the prefix and one per
  // batch of the streamed table (§92)
  let left_width = step.out_dtypes.length() - step.right_width
  let rplan : ResidualPlan? = match step.residual {
    Some(re) => Some(residual_plan(re, left_width, step.right_width))
    None => None
  }
  let (rleft_t, rleft_u) = match rplan {
    Some(plan) => residual_side_state(plan, prefix, true)
    None => ([], [])
  }
  {
    step,
    prefix,
    prefix_keys,
    int_index: IntKeyIndex::empty(),
    str_map: Map([]),
    fast: false,
    left_width: step.out_dtypes.length() - step.right_width,
    next_right,
    rplan,
    rleft_t,
    rleft_u,
    right_now: None,
    out_p: [],
    out_t: [],
    out_at: 0,
    primed: false,
    done: false,
  }
}

///|
/// Index the prefix once, when the first table batch shows which of the two key
/// shapes this join has.
fn SwapCursor::prime(
  self : SwapCursor,
  right_keys : Array[@types.ColumnVector],
) -> Unit {
  let n = self.prefix.row_count()
  match int_key_cols(self.prefix_keys, right_keys) {
    Some((bd, bv, _, _)) => {
      self.fast = true
      self.int_index = IntKeyIndex::build(bd, bv, n)
    }
    None =>
      for r in 0.. ids.push(r)
          None => self.str_map[key] = [r]
        }
      }
  }
  self.primed = true
}

///|
/// Match one table batch against the prefix, appending its output pairs.
fn SwapCursor::match_batch(
  self : SwapCursor,
  right : @types.DataBatch,
  right_keys : Array[@types.ColumnVector],
  rstate : ResidualState?,
) -> Unit raise @types.SqlError {
  let n = right.row_count()
  if self.fast {
    let (rd, rv) = match int_key_vecs(right_keys) {
      Some(pair) => pair
      None => abort("fast-path keys changed shape between batches")
    }
    for r in 0..= 0 {
          let stops = self.int_index.starts
          for j in stops[slot]..
            for id in ids {
              if residual_ok(rstate, self.prefix, id, right, r) {
                self.out_p.push(id)
                self.out_t.push(r)
              }
            }
          None => ()
        }
      }
    }
  }
}

///|
/// Gather the next slice of pending pairs into a batch of the combined layout.
fn SwapCursor::slice(self : SwapCursor) -> @types.DataBatch {
  let total = self.out_p.length()
  let mut n = total - self.out_at
  if n > JOIN_OUT_BATCH {
    n = JOIN_OUT_BATCH
  }
  let p : Array[Int32] = []
  let t : Array[Int32] = []
  for i in self.out_at..<(self.out_at + n) {
    p.push(self.out_p[i])
    t.push(self.out_t[i])
  }
  self.out_at += n
  let right = match self.right_now {
    Some(b) => b
    None => abort("swap cursor: pending pairs without a table batch")
  }
  let columns : Array[@types.ColumnVector] = []
  for c in 0.. @types.DataBatch? raise @types.SqlError {
  while true {
    if self.out_at < self.out_p.length() {
      let batch = self.slice()
      if self.out_at >= self.out_p.length() {
        self.out_p = []
        self.out_t = []
        self.out_at = 0
      }
      return Some(batch)
    }
    if self.done {
      return None
    }
    match (self.next_right)() {
      Some(right) => {
        if right.row_count() == 0 || self.prefix.row_count() == 0 {
          continue
        }
        let right_keys : Array[@types.ColumnVector] = []
        for e in self.step.keys_r {
          right_keys.push(eval_vec(e, right))
        }
        if !self.primed {
          self.prime(right_keys)
        }
        let rstate : ResidualState? = match self.rplan {
          Some(plan) => {
            // only the table's half changes per batch
            let (bt, bu) = residual_side_state(plan, right, false)
            Some({ plan, lt: self.rleft_t, lu: self.rleft_u, bt, bu, })
          }
          None => None
        }
        self.right_now = Some(right)
        self.match_batch(right, right_keys, rstate)
      }
      None => {
        self.done = true
        return None
      }
    }
  }
  abort("swap cursor: the loop leaves only by returning")
}

///|
/// The most rows the earlier steps of a chain may hand over for the last step to
/// be swapped. Under a budget a quarter of it, in estimated bytes, is the other
/// bound, so the swapped side is the same kind of promise the fit check gives.
const SWAP_PREFIX_ROWS : Int32 = 65536

///|
/// Whether the last step of this chain may swap its sides. INNER only; an
/// aggregate whose answer depends on the order rows arrive in (`string_agg`,
/// `approx_quantile`) keeps the unswapped chain, and so does a LIMIT or OFFSET
/// over groups with no ORDER BY, because the groups it keeps would change.
fn join_agg_swappable(bq : BoundQuery) -> Bool {
  let last = bq.steps.length() - 1
  if last < 1 {
    return false
  }
  match bq.steps[last].kind {
    Inner => ()
    _ => return false
  }
  for spec in bq.aggs {
    match spec.kind {
      StringAgg | ApproxQuantile => return false
      _ => ()
    }
  }
  if bq.order_by.length() == 0 && (bq.limit is Some(_) || bq.offset is Some(_)) {
    return false
  }
  true
}

///|
/// Pull the output of every step but the last, up to the cap. `Some` carries the
/// rows and the rows table 0 delivered when the pull ended inside the cap; `None`
/// means it passed the cap, and everything collected is dropped.
fn join_agg_swap_prefix(
  src : JoinSource,
  launch : Launch,
  scans : Array[Array[@types.DataBatch]],
  opts : ExecOptions,
) -> (Array[@types.DataBatch], Int64)? raise @types.SqlError {
  let bq = launch.bq
  let steps = bq.steps
  let last = steps.length() - 1
  let head : Array[JoinStep] = []
  for i in 0.. @types.DataBatch? raise @types.SqlError = fn() -> @types.DataBatch? raise @types.SqlError {
    probe.next()
  }
  let chain = chain_pull(head, join_agg_sides(scans, steps), 1, probe_next)
  // step 1's probe filter already ran in the probe pull
  let input = if last == 1 {
    chain
  } else {
    filtered_pull(chain, steps[last].probe_filter)
  }
  let cap_bytes = if opts.budget > 0 { opts.budget / 4 } else { 2147483647 }
  let out : Array[@types.DataBatch] = []
  let mut rows = 0
  let mut bytes = 0
  while input() is Some(batch) {
    out.push(batch)
    rows += batch.row_count()
    bytes += estimate_batch_bytes(batch)
    if rows > SWAP_PREFIX_ROWS || bytes > cap_bytes {
      return None
    }
  }
  Some((out, probe.scanned))
}

///|
/// Run the aggregate from the swapped last step: the prefix is the build side and
/// the last table is the probe, streamed through its build filter. The counters
/// are the unswapped chain's: one `scan` per table, then `joins`.
fn run_join_agg_swap(
  src : JoinSource,
  launch : Launch,
  scans : Array[Array[@types.DataBatch]],
  prefix : (Array[@types.DataBatch], Int64),
  opts : ExecOptions,
  stats : Array[(String, Int64)],
  census~ : MemCensus? = None,
) -> SelectResult raise @types.SqlError {
  let bq = launch.bq
  let steps = bq.steps
  let last = steps.length() - 1
  let step = steps[last]
  let left_width = step.out_dtypes.length() - step.right_width
  let prefix_batch = concat_batches(
    prefix.0,
    head_dtypes(step.out_dtypes, left_width),
  )
  let table = table_input_pull(
    bq,
    launch.plan,
    src.builds[last - 1],
    last,
    None,
  )
  let table_next : () -> @types.DataBatch? raise @types.SqlError = fn() -> @types.DataBatch? raise @types.SqlError {
    table.next()
  }
  let cursor = SwapCursor::open(step, prefix_batch, table_next)
  let out : () -> @types.DataBatch? raise @types.SqlError = fn() -> @types.DataBatch? raise @types.SqlError {
    cursor.next()
  }
  let join : JoinFoldPull = {
    next_joined: out,
    deferred: bq.deferred,
    joined: 0L,
  }
  let next = fn() -> @types.DataBatch? raise @types.SqlError { join.next() }
  let folded = fold_pulled(next, launch, opts, census~)
  stats.push(("scan \{launch.schemas[0].table}", prefix.1))
  for i in 1..