///|
fn write_frame(out : Array[Byte], metadata : Bytes, body : Bytes) -> Unit {
  let start = reserve(out, 8)
  put_i32(out, start, -1)
  put_i32(out, start + 4, metadata.length())
  append_bytes(out, metadata)
  append_bytes(out, body)
}

///|
fn write_eos(out : Array[Byte]) -> Unit {
  let p = reserve(out, 8)
  put_i32(out, p, -1)
}

///|
fn validate_batches(
  schema : Schema,
  batches : Array[RecordBatch],
) -> Unit raise ArrowError {
  validate_schema(schema)
  for batch in batches {
    batch.validate()
    if batch.schema != schema {
      raise Invalid("batch schema mismatch")
    }
  }
}

///|
/// Write an uncompressed V5 IPC stream, including a schema and end marker.
pub fn write_stream(
  schema : Schema,
  batches : Array[RecordBatch],
) -> Bytes raise ArrowError {
  validate_batches(schema, batches)
  let out : Array[Byte] = []
  write_frame(
    out,
    schema_message(schema, replacement=schema_has_dictionary(schema)),
    b"",
  )
  let dictionaries : Array[DictionaryEntry] = []
  for batch in batches {
    emit_dictionary_updates(out, schema, batch, dictionaries, true)
    let layout = encode_body(batch)
    write_frame(out, batch_message(batch.num_rows, layout), layout.body)
  }
  write_eos(out)
  Bytes::from_array(out)
}

///|
/// Decode batches on demand from an in-memory stream. This retains the input
/// Bytes; it is not a network push parser. Both explicit EOS and clean EOF work.
pub struct StreamReader {
  schema : Schema
  priv data : Bytes
  priv limits : ReadLimits
  priv mut offset : Int
  priv mut count : Int
  priv mut finished : Bool
  priv dictionaries : Array[DictionaryEntry]
  priv mut dictionary_messages : Int
}

///|
pub fn StreamReader::new(
  data : Bytes,
  limits? : ReadLimits = ReadLimits::default(),
) -> StreamReader raise ArrowError {
  limits.validate()
  let frame = match read_frame(data, 0, data.length(), limits) {
    Some(frame) => frame
    None => raise Invalid("missing schema message")
  }
  if frame.kind != 1 {
    raise Invalid("stream must start with a schema")
  }
  let schema = decode_schema(frame.metadata, frame.header, limits)
  {
    schema,
    data,
    limits,
    offset: frame.next,
    count: 0,
    finished: false,
    dictionaries: [],
    dictionary_messages: 0,
  }
}

///|
/// Return the next batch. On error the reader becomes terminal.
pub fn StreamReader::next(self : StreamReader) -> RecordBatch? raise ArrowError {
  if self.finished {
    return None
  }
  self.finished = true
  while true {
    let frame = match
      read_frame(self.data, self.offset, self.data.length(), self.limits) {
      None => return None
      Some(frame) => frame
    }
    self.offset = frame.next
    if frame.kind == 2 {
      if self.dictionary_messages >= self.limits.max_dictionary_messages {
        raise LimitExceeded("dictionary message count")
      }
      decode_dictionary_frame(
        frame,
        self.schema,
        self.dictionaries,
        self.limits,
        true,
      )
      self.dictionary_messages += 1
      continue
    }
    if self.count >= self.limits.max_batches {
      raise LimitExceeded("batch count")
    }
    let result = decode_body(frame, self.schema, self.limits, self.dictionaries)
    self.count += 1
    self.finished = false
    return Some(result)
  }
  None
}

///|
/// Read and materialize all batches of an IPC stream.
pub fn read_stream(
  data : Bytes,
  limits? : ReadLimits = ReadLimits::default(),
) -> IpcData raise ArrowError {
  let reader = StreamReader::new(data, limits~)
  let batches = []
  while reader.next() is Some(batch) {
    batches.push(batch)
  }
  { schema: reader.schema, batches, }
}

///|
priv struct FileBlock {
  offset : Int
  metadata_size : Int
  body_size : Int
}

///|
fn encode_footer(
  schema : Schema,
  blocks : Array[FileBlock],
  dictionaries : Array[FileBlock],
) -> Bytes {
  let out : Array[Byte] = [b'\x00', b'\x00', b'\x00', b'\x00']
  let footer = fb_table(out, 4)
  put_i32(out, 0, footer.pos)
  fb_int(out, footer, 0, 4)
  fb_ref(out, footer, 1, encode_schema(out, schema))
  if !dictionaries.is_empty() {
    let dictionary_blocks = fb_vector(out, dictionaries.length(), 24, 8)
    fb_ref(out, footer, 2, dictionary_blocks)
    for i in 0.. Bytes raise ArrowError {
  validate_batches(schema, batches)
  let out : Array[Byte] = []
  append_bytes(out, b"ARROW1")
  align_bytes(out, 8)
  write_frame(out, schema_message(schema), b"")
  let blocks = []
  let dictionary_blocks = []
  let dictionary_state : Array[DictionaryEntry] = []
  for batch in batches {
    for i in 0..= 0 {
          if !dictionary_equal(dictionary_state[slot].values, values) {
            raise Unsupported("dictionary replacement in file")
          }
        } else {
          let layout = dictionary_layout(values)
          let metadata = dictionary_message(id, values.length(), layout, false)
          dictionary_blocks.push(FileBlock::{
            offset: out.length(),
            metadata_size: 8 + metadata.length(),
            body_size: layout.body.length(),
          })
          write_frame(out, metadata, layout.body)
          dictionary_state.push({ id, values, })
        }
      }
    }
    let layout = encode_body(batch)
    let metadata = batch_message(batch.num_rows, layout)
    blocks.push(FileBlock::{
      offset: out.length(),
      metadata_size: 8 + metadata.length(),
      body_size: layout.body.length(),
    })
    write_frame(out, metadata, layout.body)
  }
  write_eos(out)
  let footer = encode_footer(schema, blocks, dictionary_blocks)
  append_bytes(out, footer)
  put_i32(out, reserve(out, 4), footer.length())
  append_bytes(out, b"ARROW1")
  Bytes::from_array(out)
}

///|
/// Footer-indexed access to record batches in an in-memory Arrow IPC file.
pub struct FileReader {
  schema : Schema
  priv data : Bytes
  priv limits : ReadLimits
  priv blocks : Array[FileBlock]
  priv dictionary_snapshots : Array[Array[DictionaryEntry]]
}

///|
fn file_blocks(
  footer : Bytes,
  root : Int,
  slot : Int,
  limits : ReadLimits,
  footer_start : Int,
) -> Array[FileBlock] raise ArrowError {
  let limit = if slot == 2 {
    limits.max_dictionary_messages
  } else {
    limits.max_batches
  }
  let (p, count) = vector_info(footer, root, slot, 24, limit)
  let blocks = []
  let mut previous = -1
  for i in 0.. footer_start - offset ||
      body_size > footer_start - offset - metadata_size {
      raise Invalid("invalid file block")
    }
    if metadata_size - 8 > limits.max_metadata_bytes {
      raise LimitExceeded("file block metadata size")
    }
    blocks.push({ offset, metadata_size, body_size, })
    previous = offset
  }
  blocks
}

///|
pub fn FileReader::new(
  data : Bytes,
  limits? : ReadLimits = ReadLimits::default(),
) -> FileReader raise ArrowError {
  limits.validate()
  if data.length() < 18 ||
    !data.has_prefix(b"ARROW1") ||
    !data.has_suffix(b"ARROW1") {
    raise Invalid("invalid Arrow file magic or size")
  }
  let footer_size = read_i32(data, data.length() - 10)
  if footer_size < 0 || footer_size > data.length() - 18 {
    raise Invalid("invalid footer size")
  }
  if footer_size > limits.max_metadata_bytes {
    raise LimitExceeded("footer size")
  }
  let footer_start = data.length() - 10 - footer_size
  let footer = slice_bytes(data, footer_start, footer_size)
  let root = follow_offset(footer, 0)
  let version = field_int(footer, root, 0, 2, 0)
  if version != 3 && version != 4 {
    raise Unsupported("footer metadata version")
  }
  let schema = decode_schema(footer, required_ref(footer, root, 1), limits)
  let (_, custom) = vector_info(footer, root, 4, 4, limits.max_fields)
  if custom != 0 {
    raise Unsupported("footer custom metadata")
  }
  let first = match read_frame(data, 8, footer_start, limits) {
    Some(frame) => frame
    None => raise Invalid("missing file schema message")
  }
  // PyArrow 21 writes V5 footers even when its stream metadata is V4.
  // Accept that established form, but reject a footer older than the header.
  if first.version > version ||
    first.kind != 1 ||
    decode_schema(first.metadata, first.header, limits) != schema {
    raise Invalid("header/footer schema mismatch")
  }
  let blocks = file_blocks(footer, root, 3, limits, footer_start)
  let dictionary_blocks = file_blocks(footer, root, 2, limits, footer_start)
  let state : Array[DictionaryEntry] = []
  let snapshots : Array[Array[DictionaryEntry]] = []
  let mut offset = first.next
  let mut batch_index = 0
  let mut dictionary_index = 0
  while batch_index < blocks.length() ||
        dictionary_index < dictionary_blocks.length() {
    let is_dictionary = dictionary_index < dictionary_blocks.length() &&
      dictionary_blocks[dictionary_index].offset == offset
    let is_batch = batch_index < blocks.length() &&
      blocks[batch_index].offset == offset
    if is_dictionary == is_batch {
      raise Invalid("missing, overlapping or unindexed file block")
    }
    let block = if is_dictionary {
      dictionary_blocks[dictionary_index]
    } else {
      blocks[batch_index]
    }
    let end = offset + block.metadata_size + block.body_size
    let frame = match read_frame(data, offset, end, limits) {
      Some(frame) => frame
      None => raise Invalid("file block points to EOS")
    }
    if frame.metadata_size != block.metadata_size ||
      frame.body.length() != block.body_size ||
      frame.next != end {
      raise Invalid("footer block lengths differ from message")
    }
    if is_dictionary {
      decode_dictionary_frame(frame, schema, state, limits, false)
      dictionary_index += 1
    } else {
      if frame.kind != 3 {
        raise Invalid("file record block is not a batch")
      }
      snapshots.push(state.map(entry => entry))
      batch_index += 1
    }
    offset = end
  }
  if read_frame(data, offset, footer_start, limits) is Some(_) {
    raise Invalid("unindexed file message")
  }
  { schema, data, limits, blocks, dictionary_snapshots: snapshots, }
}

///|
pub fn FileReader::num_batches(self : FileReader) -> Int {
  self.blocks.length()
}

///|
pub fn FileReader::is_empty(self : FileReader) -> Bool {
  self.blocks.is_empty()
}

///|
pub fn FileReader::get_batch(
  self : FileReader,
  index : Int,
) -> RecordBatch raise ArrowError {
  if index < 0 || index >= self.blocks.length() {
    raise Invalid("batch index out of range")
  }
  let block = self.blocks[index]
  let frame = match
    read_frame(
      self.data,
      block.offset,
      block.offset + block.metadata_size + block.body_size,
      self.limits,
    ) {
    None => raise Invalid("file block points to EOS")
    Some(frame) => frame
  }
  if frame.metadata_size != block.metadata_size ||
    frame.body.length() != block.body_size {
    raise Invalid("footer block lengths differ from message")
  }
  decode_body(frame, self.schema, self.limits, self.dictionary_snapshots[index])
}

///|
pub fn read_file(
  data : Bytes,
  limits? : ReadLimits = ReadLimits::default(),
) -> IpcData raise ArrowError {
  let reader = FileReader::new(data, limits~)
  let batches = []
  for i in 0..