///|
/// Predicate operators available to the trace query planner.
pub enum QueryPredicateKind {
  SignalEquals
  SignalMatches
  TimeAtLeast
  TimeAtMost
  ValueKindIs
  HasTransition
} derive(Debug, Eq)

///|
/// Physical operations used in a query plan.
pub enum QueryPlanNodeKind {
  FullScan
  SignalIndex
  TimeIndex
  ValueFilter
  Projection
  Sort
  Limit
} derive(Debug, Eq)

///|
/// One normalized query predicate.
pub struct QueryPredicate {
  kind : QueryPredicateKind
  field : String
  argument : String
  lower : Int64?
  upper : Int64?
} derive(Debug)

///|
/// One node in a costed query plan.
pub struct QueryPlanNode {
  index : Int
  kind : QueryPlanNodeKind
  input_rows : Int
  output_rows : Int
  estimated_cost : Double
  detail : String
} derive(Debug)

///|
/// A complete query plan with predicates and projections.
pub struct QueryPlanV2 {
  predicates : Array[QueryPredicate]
  signals : Array[String]
  limit : Int?
  nodes : Array[QueryPlanNode]
  estimated_rows : Int
  estimated_cost : Double
} derive(Debug)

///|
/// Materialized result from executing a query plan.
pub struct QueryResultV2 {
  trace : Trace
  scanned_rows : Int
  returned_rows : Int
  elapsed_units : Int64
  plan : QueryPlanV2
} derive(Debug)

///|
/// List query predicates.
pub fn query_predicate_catalog() -> Array[QueryPredicateKind] {
  [
    SignalEquals,
    SignalMatches,
    TimeAtLeast,
    TimeAtMost,
    ValueKindIs,
    HasTransition,
  ]
}

///|
/// List physical query plan nodes.
pub fn query_plan_node_catalog() -> Array[QueryPlanNodeKind] {
  [FullScan, SignalIndex, TimeIndex, ValueFilter, Projection, Sort, Limit]
}

///|
/// Start a query plan for a trace.
pub fn query_plan(trace : Trace) -> QueryPlanV2 {
  let rows = trace.changes.length()
  {
    predicates: [],
    signals: [],
    limit: None,
    nodes: [
      {
        index: 0,
        kind: FullScan,
        input_rows: rows,
        output_rows: rows,
        estimated_cost: rows.to_double(),
        detail: "scan changes",
      },
    ],
    estimated_rows: rows,
    estimated_cost: rows.to_double(),
  }
}

///|
/// Add an equality predicate for a signal identifier.
pub fn QueryPlanV2::where_signal(
  self : QueryPlanV2,
  signal_id : String,
) -> QueryPlanV2 {
  let predicates = self.predicates.copy()
  predicates.push({
    kind: SignalEquals,
    field: "id",
    argument: signal_id,
    lower: None,
    upper: None,
  })
  let signals = self.signals.copy()
  if !signals.contains(signal_id) {
    signals.push(signal_id)
  }
  { ..self, predicates, signals }
}

///|
/// Add a signal glob/prefix predicate.
pub fn QueryPlanV2::where_signal_matches(
  self : QueryPlanV2,
  pattern : String,
) -> QueryPlanV2 {
  let predicates = self.predicates.copy()
  predicates.push({
    kind: SignalMatches,
    field: "id",
    argument: pattern,
    lower: None,
    upper: None,
  })
  { ..self, predicates, }
}

///|
/// Restrict the query to an inclusive time range.
pub fn QueryPlanV2::where_time(
  self : QueryPlanV2,
  start : Int64?,
  end : Int64?,
) -> QueryPlanV2 {
  let predicates = self.predicates.copy()
  predicates.push({
    kind: TimeAtLeast,
    field: "time",
    argument: "",
    lower: start,
    upper: end,
  })
  { ..self, predicates, }
}

///|
/// Restrict the query to a textual value kind.
pub fn QueryPlanV2::where_value_kind(
  self : QueryPlanV2,
  kind : String,
) -> QueryPlanV2 {
  let predicates = self.predicates.copy()
  predicates.push({
    kind: ValueKindIs,
    field: "value",
    argument: kind,
    lower: None,
    upper: None,
  })
  { ..self, predicates, }
}

///|
/// Limit the number of returned changes.
pub fn QueryPlanV2::take(self : QueryPlanV2, count : Int) -> QueryPlanV2 {
  { ..self, limit: Some(Int::max(0, count)) }
}

///|
/// Build a costed plan against the actual trace shape.
pub fn QueryPlanV2::optimize(self : QueryPlanV2, trace : Trace) -> QueryPlanV2 {
  let nodes : Array[QueryPlanNode] = []
  let mut input = trace.changes.length()
  let mut total_cost = 0.0
  let mut index = 0
  for predicate in self.predicates {
    let selectivity = query_selectivity(predicate, trace)
    let output = (input.to_double() * selectivity).to_int()
    let kind = query_plan_kind(predicate.kind)
    let cost = query_predicate_cost(kind, input)
    nodes.push({
      index,
      kind,
      input_rows: input,
      output_rows: output,
      estimated_cost: cost,
      detail: predicate.argument,
    })
    total_cost += cost
    input = output
    index += 1
  }
  if self.signals.length() > 0 {
    let output = Int::min(
      input,
      trace.changes.filter(fn(item) { self.signals.contains(item.id) }).length(),
    )
    nodes.push({
      index,
      kind: Projection,
      input_rows: input,
      output_rows: output,
      estimated_cost: input.to_double() * 0.1,
      detail: "project selected signals",
    })
    total_cost += input.to_double() * 0.1
    input = output
    index += 1
  }
  match self.limit {
    Some(limit) => {
      let output = Int::min(input, limit)
      nodes.push({
        index,
        kind: Limit,
        input_rows: input,
        output_rows: output,
        estimated_cost: 0.01,
        detail: "limit result",
      })
      total_cost += 0.01
      input = output
    }
    None => ()
  }
  { ..self, nodes, estimated_rows: input, estimated_cost: total_cost }
}

///|
/// Execute a plan by filtering and preserving source order.
pub fn execute_query(trace : Trace, plan : QueryPlanV2) -> QueryResultV2 {
  let started = trace.changes.length().to_int64()
  let mut selected = trace.changes.copy()
  for predicate in plan.predicates {
    selected = selected.filter(fn(change) { query_matches(change, predicate) })
  }
  if plan.signals.length() > 0 {
    selected = selected.filter(fn(change) { plan.signals.contains(change.id) })
  }
  match plan.limit {
    Some(limit) if selected.length() > limit => {
      let limited : Array[Change] = []
      for index, change in selected {
        if index >= limit {
          break
        }
        limited.push(change)
      }
      selected = limited
    }
    _ => ()
  }
  let result_trace = { ..trace, changes: selected }
  let optimized = plan.optimize(trace)
  {
    trace: result_trace,
    scanned_rows: trace.changes.length(),
    returned_rows: result_trace.changes.length(),
    elapsed_units: started + result_trace.changes.length().to_int64(),
    plan: optimized,
  }
}

///|
/// Find the first returned change for a signal.
pub fn QueryResultV2::first_change(
  self : QueryResultV2,
  signal_id : String,
) -> Change? {
  for change in self.trace.changes {
    if change.id == signal_id {
      return Some(change)
    }
  }
  None
}

///|
/// Render the plan as a table suitable for performance diagnostics.
pub fn QueryPlanV2::to_text(self : QueryPlanV2) -> String {
  let out = StringBuilder()
  out.write_string(
    "rows=" +
    self.estimated_rows.to_string() +
    " cost=" +
    self.estimated_cost.to_string() +
    "\n",
  )
  for node in self.nodes {
    out.write_string(
      node.index.to_string() +
      " " +
      query_plan_node_name(node.kind) +
      " " +
      node.input_rows.to_string() +
      "->" +
      node.output_rows.to_string() +
      " cost=" +
      node.estimated_cost.to_string() +
      " " +
      node.detail +
      "\n",
    )
  }
  out.to_string()
}

///|
/// Render the query result as JSON metadata plus trace JSON.
pub fn QueryResultV2::to_json(self : QueryResultV2) -> String {
  "{\"scanned_rows\":" +
  self.scanned_rows.to_string() +
  ",\"returned_rows\":" +
  self.returned_rows.to_string() +
  ",\"elapsed_units\":" +
  self.elapsed_units.to_string() +
  ",\"plan_cost\":" +
  self.plan.estimated_cost.to_string() +
  ",\"trace\":" +
  trace_json(self.trace) +
  "}"
}

///|
/// Return a rough selectivity estimate from actual signal indexes.
fn query_selectivity(predicate : QueryPredicate, trace : Trace) -> Double {
  match predicate.kind {
    SignalEquals => {
      let count = trace.changes
        .filter(fn(item) { item.id == predicate.argument })
        .length()
      if trace.changes.length() == 0 {
        0.0
      } else {
        count.to_double() / trace.changes.length().to_double()
      }
    }
    SignalMatches => {
      let count = trace.changes
        .filter(fn(item) { signal_glob_match(item.id, predicate.argument) })
        .length()
      if trace.changes.length() == 0 {
        0.0
      } else {
        count.to_double() / trace.changes.length().to_double()
      }
    }
    TimeAtLeast | TimeAtMost => 0.5
    ValueKindIs => 0.5
    HasTransition => 0.25
  }
}

///|
/// Estimate cost for a physical predicate operator.
fn query_predicate_cost(kind : QueryPlanNodeKind, rows : Int) -> Double {
  match kind {
    SignalIndex => rows.to_double() * 0.08
    TimeIndex => rows.to_double() * 0.05
    ValueFilter => rows.to_double() * 0.35
    FullScan => rows.to_double()
    Projection => rows.to_double() * 0.1
    Sort => rows.to_double() * 0.8
    Limit => 0.01
  }
}

///|
/// Map a predicate to its preferred physical operator.
fn query_plan_kind(kind : QueryPredicateKind) -> QueryPlanNodeKind {
  match kind {
    SignalEquals | SignalMatches => SignalIndex
    TimeAtLeast | TimeAtMost => TimeIndex
    ValueKindIs => ValueFilter
    HasTransition => FullScan
  }
}

///|
/// Evaluate one predicate against one change.
fn query_matches(change : Change, predicate : QueryPredicate) -> Bool {
  match predicate.kind {
    SignalEquals => change.id == predicate.argument
    SignalMatches => signal_glob_match(change.id, predicate.argument)
    TimeAtLeast =>
      match predicate.lower {
        Some(value) => change.time >= value
        None => true
      }
    TimeAtMost =>
      match predicate.upper {
        Some(value) => change.time <= value
        None => true
      }
    ValueKindIs => change.value.kind_name() == predicate.argument
    HasTransition => true
  }
}

///|
/// Small glob matcher supporting '*' and '?' for signal queries.
fn signal_glob_match(value : String, pattern : String) -> Bool {
  if pattern == "*" {
    return true
  }
  let value_chars = value.to_array()
  let pattern_chars = pattern.to_array()
  query_glob_match_at(value_chars, pattern_chars, 0, 0)
}

///|
/// Recursive glob matcher with bounded input sizes from a signal identifier.
fn query_glob_match_at(
  value : Array[Char],
  pattern : Array[Char],
  vi : Int,
  pi : Int,
) -> Bool {
  if pi == pattern.length() {
    return vi == value.length()
  }
  if pattern[pi] == '*' {
    if query_glob_match_at(value, pattern, vi, pi + 1) {
      return true
    }
    vi < value.length() && query_glob_match_at(value, pattern, vi + 1, pi)
  } else if vi < value.length() &&
    (pattern[pi] == '?' || pattern[pi] == value[vi]) {
    query_glob_match_at(value, pattern, vi + 1, pi + 1)
  } else {
    false
  }
}

///|
/// Name a physical plan node.
fn query_plan_node_name(value : QueryPlanNodeKind) -> String {
  match value {
    FullScan => "full-scan"
    SignalIndex => "signal-index"
    TimeIndex => "time-index"
    ValueFilter => "value-filter"
    Projection => "projection"
    Sort => "sort"
    Limit => "limit"
  }
}