///|
/// 读取路径的统一单元格表示（将 parquet 的值收敛成 Delta 语义类型）
pub(all) enum Cell {
  CellI32(Int32)
  CellI64(Int64)
  CellF32(Float)
  CellF64(Double)
  CellBool(Bool)
  CellStr(String)
  CellBin(Bytes)
  CellTs(Int64)
  CellNull
  /// 嵌套值（M2）：struct 按字段名、list 按序、map 按键值对装配
  CellStruct(Array[(String, Cell)])
  CellList(Array[Cell])
  CellMap(Array[(Cell, Cell)])
} derive(Debug)

///|
pub(all) struct ReadColumn {
  name : String
  type_ : DeltaType
  cells : Array[Cell]
} derive(Debug)

///|
pub(all) struct ReadResult {
  columns : Array[ReadColumn]
  row_count : Int32
  files_read : Int32
} derive(Debug)

///|
/// parquet 值 → Cell
fn convert_cell(v : @parquet.Value) -> Cell {
  match v {
    @parquet.Int32(x) => CellI32(x)
    @parquet.Int64(x) => CellI64(x)
    @parquet.TimestampMicros(x) => CellTs(x)
    @parquet.Float(x) => CellF32(x)
    @parquet.Double(x) => CellF64(x)
    @parquet.Boolean(x) => CellBool(x)
    @parquet.String(x) => CellStr(x)
    @parquet.Binary(x) => CellBin(x)
    @parquet.Null => CellNull
  }
}

///|
/// Delta 逻辑类型与 Parquet 物理类型必须相容。当前 Parquet 层只可靠识别
/// INT96 timestamp；INT64 TIMESTAMP_MICROS 的 logical type 尚未解析，不能把
/// 普通 long 当成 timestamp。
fn parquet_type_matches(
  delta : DeltaType,
  physical : @parquet.ColumnType,
) -> Bool {
  match delta {
    BooleanType => physical is @parquet.Boolean
    ByteType | ShortType | Int32Type => physical is @parquet.Int32
    DateType => physical is @parquet.Date
    Int64Type => physical is @parquet.Int64
    FloatType => physical is @parquet.Float
    DoubleType => physical is @parquet.Double
    StringType => physical is @parquet.String
    BinaryType => physical is @parquet.Binary
    TimestampType =>
      physical is @parquet.TimestampMicros ||
      physical is @parquet.TimestampMicros64
    _ => false
  }
}

///|
fn find_column_index(columns : Array[@parquet.Column], name : String) -> Int32? {
  let mut i = 0
  while i < columns.length() {
    if columns[i].name() == name {
      return Some(i)
    }
    i = i + 1
  }
  None
}

///|
fn partition_value_of(file : AddFile, name : String) -> String? {
  for pair in file.partition_values {
    let (k, v) = pair
    if k == name {
      return Some(v)
    }
  }
  None
}

///|
/// ISO 日期 `YYYY-MM-DD` → Unix epoch 天数；纯整数仍按 Delta 的 date 天数值解析。
fn partition_date_days(raw : String) -> Int32 raise DeltaError {
  let parts = Array::from_iter(raw.split("-"))
  if parts.length() == 1 {
    return @string.parse_int(raw, base=10) catch {
      _ =>
        raise DeltaError::ParseError("invalid date partition value '\{raw}'")
    }
  }
  if parts.length() != 3 {
    raise DeltaError::ParseError("invalid date partition value '\{raw}'")
  }
  let year = @string.parse_int(parts[0], base=10) catch {
    _ => raise DeltaError::ParseError("invalid date partition value '\{raw}'")
  }
  let month = @string.parse_int(parts[1], base=10) catch {
    _ => raise DeltaError::ParseError("invalid date partition value '\{raw}'")
  }
  let day = @string.parse_int(parts[2], base=10) catch {
    _ => raise DeltaError::ParseError("invalid date partition value '\{raw}'")
  }
  if month < 1 || month > 12 || day < 1 || day > 31 {
    raise DeltaError::ParseError("invalid date partition value '\{raw}'")
  }
  let y = if month <= 2 { year - 1 } else { year }
  let era = if y >= 0 { y / 400 } else { (y - 399) / 400 }
  let yoe = y - era * 400
  let mp = month + (if month > 2 { -3 } else { 9 })
  let doy = (153 * mp + 2) / 5 + day - 1
  let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy
  era * 146097 + doe - 719468
}

///|
fn cell_from_partition(ty : DeltaType, raw : String) -> Cell raise DeltaError {
  match ty {
    ByteType | ShortType | Int32Type | DateType =>
      CellI32(
        if ty is DateType {
          partition_date_days(raw)
        } else {
          @string.parse_int(raw, base=10) catch {
            _ =>
              raise DeltaError::ParseError(
                "invalid \{delta_type_name(ty)} partition value '\{raw}'",
              )
          }
        },
      )
    Int64Type =>
      CellI64(
        @string.parse_int64(raw, base=10) catch {
          _ =>
            raise DeltaError::ParseError(
              "invalid int64 partition value '\{raw}'",
            )
        },
      )
    TimestampType =>
      CellTs(
        @string.parse_int64(raw, base=10) catch {
          _ =>
            raise DeltaError::ParseError(
              "timestamp partition value must be microseconds since epoch: '\{raw}'",
            )
        },
      )
    FloatType =>
      CellF32(
        Float::from_double(
          @string.parse_double(raw) catch {
            _ =>
              raise DeltaError::ParseError(
                "invalid float partition value '\{raw}'",
              )
          },
        ),
      )
    DoubleType =>
      CellF64(
        @string.parse_double(raw) catch {
          _ =>
            raise DeltaError::ParseError(
              "invalid double partition value '\{raw}'",
            )
        },
      )
    BooleanType =>
      if raw == "true" {
        CellBool(true)
      } else if raw == "false" {
        CellBool(false)
      } else {
        raise DeltaError::ParseError(
          "invalid boolean partition value '\{raw}'",
        )
      }
    StringType => CellStr(raw)
    _ =>
      raise DeltaError::Unsupported(
        "partition column of type \{delta_type_name(ty)} is not supported by the reader",
      )
  }
}

///|
/// 读取 snapshot 中的数据文件。
///
/// - `root` 是表目录（parquet 路径相对它）
/// - `limit` 限制读入行数（None = 全部）
/// 分区列不会出现在数据文件里，这里用 `partitionValues` 补齐常量列。
/// schema 含嵌套字段时走 shredded 装配路径（src/shred.mbt）。
pub fn read_snapshot(
  root : String,
  snapshot : Snapshot,
  limit : Int32?,
) -> ReadResult raise DeltaError {
  ensure_readable_schema(snapshot.metadata.schema)
  if schema_has_nested(snapshot.metadata.schema) {
    return read_snapshot_shredded(root, snapshot, limit)
  }
  let fields = snapshot.metadata.schema.fields
  let columns : Array[ReadColumn] = []
  for f in fields {
    columns.push({ name: f.name, type_: f.type_, cells: [], })
  }
  let max_rows = match limit {
    Some(l) => l
    None => 1000000000
  }
  let mut row_count = 0
  let mut files_read = 0
  for file in snapshot.files {
    if row_count >= max_rows {
      break
    }
    let path = root + "/" + file.path
    let pf = @parquet.read_file(path) catch {
      @parquet.Unsupported(msg) =>
        raise DeltaError::Unsupported("cannot read '\{file.path}': \{msg}")
      @parquet.InvalidData(msg) =>
        raise DeltaError::ParseError("corrupt parquet '\{file.path}': \{msg}")
      @parquet.Io(msg) =>
        raise DeltaError::IoError("cannot read '\{file.path}': \{msg}")
    }
    files_read = files_read + 1
    let rows = pf.rows()
    let remaining = max_rows - row_count
    let take = if rows.length() < remaining { rows.length() } else { remaining }
    let pcols = pf.columns()
    let mut idx = 0
    while idx < fields.length() {
      let field = fields[idx]
      let col = columns[idx]
      match partition_value_of(file, field.name) {
        Some(raw) => {
          let cell = cell_from_partition(field.type_, raw)
          let mut r = 0
          while r < take {
            col.cells.push(cell)
            r = r + 1
          }
        }
        None =>
          match find_column_index(pcols, field.name) {
            Some(pidx) => {
              let physical = pcols[pidx].column_type()
              if !parquet_type_matches(field.type_, physical) {
                raise DeltaError::ParseError(
                  "column '\{field.name}' is \{delta_type_name(field.type_)} in the Delta schema but \{physical} in parquet file '\{file.path}'",
                )
              }
              let mut r = 0
              while r < take {
                let row = rows[r]
                col.cells.push(convert_cell(row[pidx]))
                r = r + 1
              }
            }
            None =>
              raise DeltaError::Unsupported(
                "column '\{field.name}' is missing from parquet file '\{file.path}'",
              )
          }
      }
      idx = idx + 1
    }
    row_count = row_count + take
  }
  { columns, row_count, files_read, }
}

///|
/// 单元格的展示形式（CLI / 示例用）。嵌套值以 JSON 风格渲染。
pub fn cell_to_display(cell : Cell) -> String {
  match cell {
    CellI32(x) => x.to_string()
    CellI64(x) => x.to_string()
    CellF32(x) => x.to_string()
    CellF64(x) => x.to_string()
    CellBool(x) => x.to_string()
    CellStr(x) => x
    CellBin(x) => x.to_unchecked_string()
    CellTs(x) => x.to_string()
    CellNull => "NULL"
    CellStruct(members) => {
      let sb = StringBuilder()
      sb.write_char('{')
      let mut i = 0
      while i < members.length() {
        if i > 0 {
          sb.write_string(", ")
        }
        let (name, value) = members[i]
        sb.write_string("\"" + name + "\": ")
        sb.write_string(cell_to_display(value))
        i = i + 1
      }
      sb.write_char('}')
      sb.to_string()
    }
    CellList(items) => {
      let sb = StringBuilder()
      sb.write_char('[')
      let mut i = 0
      while i < items.length() {
        if i > 0 {
          sb.write_string(", ")
        }
        sb.write_string(cell_to_display(items[i]))
        i = i + 1
      }
      sb.write_char(']')
      sb.to_string()
    }
    CellMap(entries) => {
      let sb = StringBuilder()
      sb.write_char('{')
      let mut i = 0
      while i < entries.length() {
        if i > 0 {
          sb.write_string(", ")
        }
        let (k, v) = entries[i]
        sb.write_string(cell_to_display(k))
        sb.write_string(": ")
        sb.write_string(cell_to_display(v))
        i = i + 1
      }
      sb.write_char('}')
      sb.to_string()
    }
  }
}