///|
fn default_protocol() -> Protocol {
  { min_reader_version: 1, min_writer_version: 1, }
}

///|
fn empty_schema() -> DeltaSchema {
  { fields: [], raw_json: "", }
}

///|
fn empty_metadata() -> TableInfo {
  {
    id: "",
    name: None,
    schema: empty_schema(),
    partition_columns: [],
    format_provider: "parquet",
    configuration: [],
  }
}

///|
fn upsert_file(files : Array[AddFile], f : AddFile) -> Unit {
  let rebuilt : Array[AddFile] = []
  let mut replaced = false
  for x in files {
    if x.path == f.path {
      rebuilt.push(f)
      replaced = true
    } else {
      rebuilt.push(x)
    }
  }
  if !replaced {
    rebuilt.push(f)
  }
  files.clear()
  for x in rebuilt {
    files.push(x)
  }
}

///|
fn drop_file(files : Array[AddFile], path : String) -> Unit {
  let kept : Array[AddFile] = []
  for x in files {
    if x.path != path {
      kept.push(x)
    }
  }
  files.clear()
  for x in kept {
    files.push(x)
  }
}

///|
fn stats_num_records(stats : String?) -> Int64 {
  match stats {
    Some(s) => {
      let j = @json.parse(s) catch { _ => return 0L }
      match json_field_int64(j, "numRecords") {
        Some(v) => v
        None => 0L
      }
    }
    None => 0L
  }
}

///|
pub(all) struct DeltaTable {
  root : String
}

///|
pub fn DeltaTable::open(root : String) -> DeltaTable raise DeltaError {
  if !@fs.path_exists(log_dir(root)) {
    raise DeltaError::NotFound(
      "'\{root}' is not a Delta table (no _delta_log directory)",
    )
  }
  { root, }
}

///|
pub fn DeltaTable::versions(self : DeltaTable) -> Array[Int32] raise DeltaError {
  list_versions(self.root)
}

///|
pub fn DeltaTable::current_version(self : DeltaTable) -> Int32 raise DeltaError {
  let versions = list_versions(self.root)
  if versions.length() == 0 {
    raise DeltaError::NotFound("table '\{self.root}' has no log entries")
  }
  versions[versions.length() - 1]
}

///|
/// 回放日志到指定版本（None = 最新），返回 snapshot 与历史
fn replay_until(
  root : String,
  target : Int32?,
) -> (Snapshot, Array[HistoryEntry]) raise DeltaError {
  let versions = list_versions(root)
  if versions.length() == 0 {
    raise DeltaError::NotFound("table '\{root}' has no log entries")
  }
  if versions[0] != 0 {
    raise DeltaError::Unsupported(
      "log starts at version \{versions[0]}; checkpoint-only or truncated tables are not supported (replay requires the initial log entries)",
    )
  }
  let latest = versions[versions.length() - 1]
  let target_version = match target {
    Some(v) => v
    None => latest
  }
  if target_version < 0 || target_version > latest {
    raise DeltaError::NotFound(
      "requested version \{target_version} is out of range (latest = \{latest})",
    )
  }
  if !versions.contains(target_version) {
    raise DeltaError::NotFound("log version \{target_version} does not exist")
  }
  let mut protocol = default_protocol()
  let mut metadata = empty_metadata()
  let files : Array[AddFile] = []
  let history : Array[HistoryEntry] = []
  let mut v = 0
  while v <= target_version {
    let actions = read_actions(root, v)
    for action in actions {
      match action {
        ProtocolAction(p) => protocol = p
        MetaDataAction(m) => metadata = m
        AddAction(f) => upsert_file(files, f)
        RemoveAction(r) => drop_file(files, r.path)
        CommitInfoAction(c) =>
          history.push({
            version: c.version,
            timestamp: c.timestamp,
            operation: c.operation,
          })
        UnknownAction(_) => ()
      }
    }
    v = v + 1
  }
  let mut total_rows = 0L
  let mut total_bytes = 0L
  for f in files {
    total_rows = total_rows + stats_num_records(f.stats)
    total_bytes = total_bytes + f.size
  }
  let snapshot = {
    version: target_version,
    protocol,
    metadata,
    files,
    total_rows,
    total_bytes,
  }
  (snapshot, history)
}

///|
pub fn DeltaTable::snapshot(self : DeltaTable) -> Snapshot raise DeltaError {
  let (snapshot, _) = replay_until(self.root, None)
  snapshot
}

///|
pub fn DeltaTable::snapshot_at(
  self : DeltaTable,
  version : Int32,
) -> Snapshot raise DeltaError {
  let (snapshot, _) = replay_until(self.root, Some(version))
  snapshot
}

///|
pub fn DeltaTable::history(
  self : DeltaTable,
) -> Array[HistoryEntry] raise DeltaError {
  let (_, history) = replay_until(self.root, None)
  history
}

///|
/// 读取表的最新版本号 + snapshot + 是否检测到 checkpoint（用于 CLI 提示）
pub fn DeltaTable::inspect(
  self : DeltaTable,
) -> (Snapshot, CheckpointInfo?, Bool) raise DeltaError {
  let (snapshot, _) = replay_until(self.root, None)
  let cp = checkpoint_info(self.root)
  let has_cp = has_checkpoint_file(self.root)
  (snapshot, cp, has_cp)
}