///|
/// Rendering of the expression tree: a handwritten `Show` presenting the
/// documented operator form, its `to_string` exposing that rendering as a
/// `String` (shared later by `LazyFrame::explain`'s plan printer), and the
/// handwritten `Debug` delegating to it. All total — they walk shape,
/// nothing can fail.
///
/// Two rendering conventions share this module, and only one of them is bound
/// to method spellings: the arithmetic / comparison operators render as
/// **display syntax** (`//`, `%`, `**` — glyphs, not the `floor_div` /
/// `modulo` / `pow` call sites they came from), while the postfix method
/// renderings (`agg_suffix`, `write_str_suffix`) spell the *public method
/// names* so a reader can copy an `explain` line back into code. The price
/// of the second convention is that renaming a public `Expr` method does
/// not rename its suffix here — a rename must update the suffix table in
/// the same change, which no guard can force (the outputs are strings, and
/// the old spelling is still a plausible diagnostic).

///|
/// Render the documented operator form, e.g.
/// `(col(revenue) - col(cost)) as profit`: columns as `col(name)`, bare
/// literals (strings quoted, the null literal as `null`), binary nodes
/// parenthesised infix, prefix `(-e)` / `(not e)`, the remaining unary
/// probes and `Agg` / `Cast` as postfix method calls, `Alias` as
/// `expr as name`, `Ternary` mirroring its `when(c).then(a).otherwise(b)`
/// construction route, a `Map` as `map("label", [inputs])`, and a
/// `LitSeries` as `lit_series("name", len)`. Handwritten because
/// `derive(Show)` is deprecated — and the derived variant form would leak
/// construction-route noise into `explain` output anyway.
pub impl Show for Expr with fn output(self, logger) {
  write_expr(logger, self)
}

///|
pub extend Expr with Show::{to_string, output}

///|
/// Handwritten `Debug`: the `ExprNode` `Map` / `MapBatches` closures
/// have no `Debug`, so neither the AST node nor the `Expr` wrapping it can
/// `derive` one. The debug form delegates to
/// the `Show` rendering (`to_string`), which already presents every variant —
/// `Map` by its `(label, inputs)`, the closure opaque — in the documented
/// operator form. Debug output surfaces in `debug_inspect` and in the failure
/// message of any assertion over something that embeds an expression — the
/// handle itself has no equality to fail. `Repr`'s public constructor wraps
/// the `Show` rendering's string directly.
pub impl @debug.Debug for Expr with fn to_repr(self) {
  @debug.Repr(self.to_string())
}

///|
pub extend Expr with Debug::{to_repr}

///|
/// A pending step for the iterative `write_expr`: either a literal string
/// fragment to emit, a sub-expression still to expand, or one of the three leaf
/// writers that take more than a plain string (`format_scalar`
/// for a `Lit`, a `Cast` target, a `Str` suffix).
priv enum WriteTask {
  Frag(String)
  Node(Expr)
  ScalarLit(@types.Scalar)
  CastTarget(@types.DataType)
  StrSuffix(@ir.StrOp)
}

///|
/// Renderer behind `Show`, driven by an explicit heap stack instead of the call
/// stack so a deeply nested tree can't overflow and abort (the same discipline
/// as `ExprNode::referenced_columns`). Every composite form is
/// self-delimiting (parentheses or
/// a closing `)` of a method call), so a node expands into a fixed sequence of
/// fragments and child placeholders; the emitted text is byte-for-byte the
/// recursive form's.
fn write_expr(logger : &Logger, expr : Expr) -> Unit {
  let stack : Array[WriteTask] = [Node(expr)]
  for ;; {
    match stack.pop() {
      None => break
      Some(task) =>
        match task {
          Frag(s) => logger.write_string(s)
          ScalarLit(v) => logger.write_string(format_scalar(v))
          CastTarget(t) => t.output(logger)
          StrSuffix(op) => write_str_suffix(logger, op)
          Node(e) => push_write_tasks(stack, e)
        }
    }
  }
}

///|
/// Expand one node into its fragment / child sequence, pushed so the parts emit
/// left to right (built forward, then pushed in reverse so the first pops
/// first). The per-variant shapes mirror the recursive renderer exactly.
fn push_write_tasks(stack : Array[WriteTask], expr : Expr) -> Unit {
  let tasks : Array[WriteTask] = match expr.node {
    // `col(name)` and `Alias`'s ` as name` render the name UNQUOTED, unlike a
    // string literal (`format_scalar` quotes those), so
    // `col("x")` stays distinct
    // from `lit_str("x")` in a plan line. A name containing literal parens can
    // therefore read paren-ambiguously (`col(a)b)`); that is display-only and
    // accepted — `explain` output is diagnostic, not re-parsed.
    Col(name) => [Frag("col(\{@text.escape_debug(name)})")]
    Lit(value) => [ScalarLit(value)]
    // The embedded data is opaque to the rendering, so a `LitSeries` shows its
    // name (quoted / escaped like a string literal) and length —
    // `lit_series("flag", 4)` — enough to read a plan line without dumping
    // every cell.
    LitSeries(s) =>
      [Frag("lit_series(\"\{@text.escape_debug(s.name())}\", \{s.len()})")]
    // A `Node` holds the opaque `Expr`, so every AST child is wrapped with
    // `Expr::of_node` before it is pushed for expansion.
    Binary(op, l, r) =>
      [
        Frag("("),
        Node(Expr::of_node(l)),
        Frag(" \{binop_symbol(op)} "),
        Node(Expr::of_node(r)),
        Frag(")"),
      ]
    Unary(op, e) =>
      match op {
        Neg => [Frag("(-"), Node(Expr::of_node(e)), Frag(")")]
        Not => [Frag("(not "), Node(Expr::of_node(e)), Frag(")")]
        IsNull => [Node(Expr::of_node(e)), Frag(".is_null()")]
        IsNotNull => [Node(Expr::of_node(e)), Frag(".is_not_null()")]
        IsNan => [Node(Expr::of_node(e)), Frag(".is_nan()")]
        IsFinite => [Node(Expr::of_node(e)), Frag(".is_finite()")]
        IsInfinite => [Node(Expr::of_node(e)), Frag(".is_infinite()")]
        DtYear => [Node(Expr::of_node(e)), Frag(".dt_year()")]
        DtMonth => [Node(Expr::of_node(e)), Frag(".dt_month()")]
        DtDay => [Node(Expr::of_node(e)), Frag(".dt_day()")]
        DtWeekday => [Node(Expr::of_node(e)), Frag(".dt_day_of_week()")]
        IsNotNan => [Node(Expr::of_node(e)), Frag(".is_not_nan()")]
        Abs => [Node(Expr::of_node(e)), Frag(".abs()")]
        Floor => [Node(Expr::of_node(e)), Frag(".floor()")]
        Ceil => [Node(Expr::of_node(e)), Frag(".ceil()")]
        Sign => [Node(Expr::of_node(e)), Frag(".sign()")]
        Round(0) => [Node(Expr::of_node(e)), Frag(".round()")]
        Round(d) => [Node(Expr::of_node(e)), Frag(".round(decimals=\{d})")]
        Sin => [Node(Expr::of_node(e)), Frag(".sin()")]
        Cos => [Node(Expr::of_node(e)), Frag(".cos()")]
        Tan => [Node(Expr::of_node(e)), Frag(".tan()")]
        Asin => [Node(Expr::of_node(e)), Frag(".asin()")]
        Acos => [Node(Expr::of_node(e)), Frag(".acos()")]
        Atan => [Node(Expr::of_node(e)), Frag(".atan()")]
        Sinh => [Node(Expr::of_node(e)), Frag(".sinh()")]
        Cosh => [Node(Expr::of_node(e)), Frag(".cosh()")]
        Tanh => [Node(Expr::of_node(e)), Frag(".tanh()")]
        Asinh => [Node(Expr::of_node(e)), Frag(".asinh()")]
        Acosh => [Node(Expr::of_node(e)), Frag(".acosh()")]
        Atanh => [Node(Expr::of_node(e)), Frag(".atanh()")]
        Ln => [Node(Expr::of_node(e)), Frag(".ln()")]
        Log2 => [Node(Expr::of_node(e)), Frag(".log2()")]
        Log10 => [Node(Expr::of_node(e)), Frag(".log10()")]
        Exp => [Node(Expr::of_node(e)), Frag(".exp()")]
        Cbrt => [Node(Expr::of_node(e)), Frag(".cbrt()")]
        Sqrt => [Node(Expr::of_node(e)), Frag(".sqrt()")]
      }
    Agg(op, e) => [Node(Expr::of_node(e)), Frag(agg_suffix(op))]
    Str(op, e) => [Node(Expr::of_node(e)), StrSuffix(op)]
    Positional(op, e) => [Node(Expr::of_node(e)), Frag(positional_suffix(op))]
    Cast(e, target) =>
      [Node(Expr::of_node(e)), Frag(".cast("), CastTarget(target), Frag(")")]
    Alias(e, name) =>
      [Node(Expr::of_node(e)), Frag(" as \{@text.escape_debug(name)}")]
    Ternary(c, t, f) =>
      [
        Frag("when("),
        Node(Expr::of_node(c)),
        Frag(").then("),
        Node(Expr::of_node(t)),
        Frag(").otherwise("),
        Node(Expr::of_node(f)),
        Frag(")"),
      ]
    // Postfix method-call form, mirroring the construction route — the
    // operand appears once, so a chained coalesce renders linearly.
    FillNull(o, v) =>
      [
        Node(Expr::of_node(o)),
        Frag(".fill_null("),
        Node(Expr::of_node(v)),
        Frag(")"),
      ]
    FillNan(o, v) =>
      [
        Node(Expr::of_node(o)),
        Frag(".fill_nan("),
        Node(Expr::of_node(v)),
        Frag(")"),
      ]
    // `col(a).diff(1)` — the step count renders as a plain integer, the
    // default reading `diff()` itself spells.
    Diff(e, n) => [Node(Expr::of_node(e)), Frag(".diff(\{n})")]
    // `col(a).clip(lower, upper)` — each side renders only when it exists,
    // the omitted sides leaving the argument list shorter (and both omitted
    // never reaching a node at all: the builder returns the operand).
    Clip(x, lo, hi) => {
      let ts : Array[WriteTask] = [Node(Expr::of_node(x)), Frag(".clip(")]
      let mut wrote = false
      if lo is Some(l) {
        ts.push(Node(Expr::of_node(l)))
        wrote = true
      }
      if hi is Some(h) {
        if wrote {
          ts.push(Frag(", "))
        }
        ts.push(Node(Expr::of_node(h)))
      }
      ts.push(Frag(")"))
      ts
    }
    // `col(a).is_in([1, 2])`: the operand, then each set member rendered as a
    // literal (`format_scalar`, like a `Lit`) inside `[...]`.
    IsIn(operand, members) => {
      let ts : Array[WriteTask] = [
        Node(Expr::of_node(operand)),
        Frag(".is_in(["),
      ]
      for i in 0.. 0 {
          ts.push(Frag(", "))
        }
        ts.push(ScalarLit(members[i]))
      }
      ts.push(Frag("])"))
      ts
    }
    // `col(a).is_between(lo, hi)` — the operand once, then both bounds.
    // A non-default `closed` renders as the named argument that builds it.
    IsBetween(x, lo, hi, closed) => {
      let ts : Array[WriteTask] = [
        Node(Expr::of_node(x)),
        Frag(".is_between("),
        Node(Expr::of_node(lo)),
        Frag(", "),
        Node(Expr::of_node(hi)),
      ]
      match closed {
        @types.ClosedInterval::Both => ()
        @types.ClosedInterval::Left => ts.push(Frag(", closed=Left"))
        @types.ClosedInterval::Right => ts.push(Frag(", closed=Right"))
        @types.ClosedInterval::None => ts.push(Frag(", closed=None"))
      }
      ts.push(Frag(")"))
      ts
    }
    // The closure is opaque, so a `Map` renders by its label and inputs:
    // `map("label", [in1, in2])`. The label is quoted / escaped like a string
    // literal so it stays on one unambiguous line.
    Map(label, inputs, _) => {
      let ts : Array[WriteTask] = [
        Frag("map(\"\{@text.escape_debug(label)}\", ["),
      ]
      for i in 0.. 0 {
          ts.push(Frag(", "))
        }
        ts.push(Node(Expr::of_node(inputs[i])))
      }
      ts.push(Frag("])"))
      ts
    }
    // Like `Map`, as `map_batches("label", [inputs])`, and — when the node is
    // flagged as a reduction — a trailing `, returns_scalar=true` so the two
    // forms render distinctly.
    MapBatches(label, inputs, returns_scalar, _) => {
      let ts : Array[WriteTask] = [
        Frag("map_batches(\"\{@text.escape_debug(label)}\", ["),
      ]
      for i in 0.. 0 {
          ts.push(Frag(", "))
        }
        ts.push(Node(Expr::of_node(inputs[i])))
      }
      ts.push(
        Frag(if returns_scalar { "], returns_scalar=true)" } else { "])" }),
      )
      ts
    }
  }
  for i = tasks.length() - 1; i >= 0; i = i - 1 {
    stack.push(tasks[i])
  }
}

///|
/// Infix symbol for a binary tag — the operator where one exists (`&` / `|`
/// for the Kleene connectives), the conventional mathematical glyph where the
/// builder is a method: `<` for `.lt()`, `//` / `%` / `**` for
/// `.floor_div()` / `.modulo()` / `.pow()`, and the function name as a word
/// for the two-argument math pair `.atan2()` / `.hypot()` — none of which
/// MoonBit spells as an operator. Display syntax, not a call site to copy.
fn binop_symbol(op : @ir.BinOp) -> String {
  match op {
    Add => "+"
    Sub => "-"
    Mul => "*"
    Div => "/"
    FloorDiv => "//"
    Mod => "%"
    Pow => "**"
    Atan2 => "atan2"
    Hypot => "hypot"
    Eq => "=="
    Ne => "!="
    Lt => "<"
    Le => "<="
    Gt => ">"
    Ge => ">="
    And => "&"
    Or => "|"
  }
}

///|
/// Postfix method spelling for an aggregation tag. The parameterised
/// `Quantile` renders its arguments display-syntax like a `Str` suffix's
/// literal arguments (`.str_contains("foo")`): a `Double::to_string` `q` and
/// the lowercase interpolation tag, omitted when it is the `Nearest` default
/// so the common spelling stays `.quantile(0.5)`.
fn agg_suffix(op : @series.ReduceOp) -> String {
  match op {
    Sum => ".sum()"
    Mean => ".mean()"
    Min => ".min()"
    Max => ".max()"
    Count => ".count()"
    Std => ".std()"
    Var => ".variance()"
    Median => ".median()"
    Quantile(q, interpolation) => {
      let tag = match interpolation {
        Nearest => ""
        Linear => ", linear"
        Lower => ", lower"
        Higher => ", higher"
        Midpoint => ", midpoint"
      }
      ".quantile(\{q}\{tag})"
    }
    NUnique => ".n_unique()"
    Mode => ".mode()"
    ArgMin => ".arg_min()"
    ArgMax => ".arg_max()"
    First => ".first()"
    Last => ".last()"
  }
}

///|
/// Postfix method spelling for a positional transform, display-syntax like
/// `agg_suffix`'s parameterised `quantile`: the non-default arguments ride
/// the call (a `shift` fill, a `reverse`d cumulative, `rank`'s method and
/// direction, `arg_sort`'s two flags), each default omitted.
fn positional_suffix(op : @ir.PositionalOp) -> String {
  match op {
    Shift(n, fill) => {
      let f = if fill.is_null() { "" } else { ", fill=\{format_scalar(fill)}" }
      ".shift(\{n}\{f})"
    }
    CumSum(reverse) => cum_suffix("cum_sum", reverse)
    CumMin(reverse) => cum_suffix("cum_min", reverse)
    CumMax(reverse) => cum_suffix("cum_max", reverse)
    CumProd(reverse) => cum_suffix("cum_prod", reverse)
    CumCount(reverse) => cum_suffix("cum_count", reverse)
    Rank(how, descending) => {
      let m = match how {
        Average => ""
        Min => "min"
        Max => "max"
        Dense => "dense"
        Ordinal => "ordinal"
      }
      let d = if descending { "descending" } else { "" }
      ".rank(\{[m, d].filter(x => x != "").join(", ")})"
    }
    ArgSort(descending, nulls_last) => {
      let d = if descending { "descending" } else { "" }
      let nl = if nulls_last { "nulls_last" } else { "" }
      ".arg_sort(\{[d, nl].filter(x => x != "").join(", ")})"
    }
    Rolling(stat, window, min_samples, center) => {
      let verb = match stat {
        Sum => "rolling_sum"
        Mean => "rolling_mean"
        Min => "rolling_min"
        Max => "rolling_max"
        Std => "rolling_std"
        Var => "rolling_variance"
        Median => "rolling_median"
      }
      let ms = match min_samples {
        Some(k) => "min_samples=\{k}"
        None => ""
      }
      let c = if center { "center" } else { "" }
      ".\{verb}(\{[ "\{window}", ms, c ].filter(x => x != "").join(", ")})"
    }
    UniqueCounts => ".unique_counts()"
    IsFirstDistinct => ".is_first_distinct()"
    IsLastDistinct => ".is_last_distinct()"
    ForwardFill(limit) =>
      ".forward_fill(\{match limit { Some(k) => "\{k}"; None => "" }})"
    BackwardFill(limit) =>
      ".backward_fill(\{match limit { Some(k) => "\{k}"; None => "" }})"
  }
}

///|
/// The `cum_*` family's shared suffix — the verb, plus `reverse` when set.
fn cum_suffix(verb : String, reverse : Bool) -> String {
  if reverse {
    ".\{verb}(reverse)"
  } else {
    ".\{verb}()"
  }
}

///|
/// Postfix method rendering for a string-namespace tag, written straight to
/// `logger` (unlike `agg_suffix`'s fixed string) because the parameterised
/// operations carry literal arguments that are quoted and escaped like a
/// `lit_str` — `.str_contains("foo")`, `.str_replace("a", "b")` — so a plan
/// line stays unambiguous and on one line.
fn write_str_suffix(logger : &Logger, op : @ir.StrOp) -> Unit {
  // Render one parameterised string-op call `.name("arg", …)`, each argument
  // quoted and `escape_debug`-escaped exactly like a rendered `lit_str`
  // (`format_scalar`), so a pattern containing a `"` or a
  // control character stays on one unambiguous line.
  fn write_str_call(name : String, args : Array[String]) -> Unit {
    logger.write_string(".")
    logger.write_string(name)
    logger.write_string("(")
    logger.write_string(
      args.map(a => "\"\{@text.escape_debug(a)}\"").join(", "),
    )
    logger.write_string(")")
  }

  match op {
    ToUppercase => logger.write_string(".str_to_uppercase()")
    ToLowercase => logger.write_string(".str_to_lowercase()")
    StripChars => logger.write_string(".str_strip_chars()")
    LenChars => logger.write_string(".str_len_chars()")
    Contains(p) => write_str_call("str_contains", [p])
    StartsWith(p) => write_str_call("str_starts_with", [p])
    EndsWith(p) => write_str_call("str_ends_with", [p])
    Replace(a, b) => write_str_call("str_replace", [a, b])
    ReplaceAll(a, b) => write_str_call("str_replace_all", [a, b])
    Reverse => logger.write_string(".str_reverse()")
    // The width renders bare, the fill as a `'x'` char literal (escaped like a
    // string so a control or quote character stays on one line).
    PadStart(width, fill) =>
      logger <+
        ".str_pad_start(\{width}, '\{@text.escape_debug(fill.to_string())}')"
    PadEnd(width, fill) =>
      logger <+
        ".str_pad_end(\{width}, '\{@text.escape_debug(fill.to_string())}')"
    // The regex forms render as the spelling that builds them: the same
    // method with `literal=false`.
    ContainsRegex(p) =>
      logger <+ ".str_contains(\"\{@text.escape_debug(p)}\", literal=false)"
    ReplaceRegex(a, b) =>
      logger <+
        ".str_replace(\"\{@text.escape_debug(a)}\", \"\{@text.escape_debug(b)}\", literal=false)"
    ReplaceAllRegex(a, b) =>
      logger <+
        ".str_replace_all(\"\{@text.escape_debug(a)}\", \"\{@text.escape_debug(b)}\", literal=false)"
    // The pattern renders quoted, the group index bare.
    Extract(pattern, group) =>
      logger <+ ".str_extract(\"\{@text.escape_debug(pattern)}\", \{group})"
    CountMatches(p) => write_str_call("str_count_matches", [p])
    // The offset renders bare; the length only when present.
    Slice(offset, length) =>
      match length {
        None => logger <+ ".str_slice(\{offset})"
        Some(n) => logger <+ ".str_slice(\{offset}, \{n})"
      }
    LenBytes => logger.write_string(".str_len_bytes()")
    // The custom strip renders its charset as a quoted argument, so it stays
    // distinct from the default `.str_strip_chars()`.
    StripCharsCustom(chars) => write_str_call("str_strip_chars", [chars])
    // The separator renders quoted, the index bare.
    SplitGet(sep, index) =>
      logger <+ ".str_split_get(\"\{@text.escape_debug(sep)}\", \{index})"
    ZFill(width) => logger <+ ".str_zfill(\{width})"
  }
}