hdf5-pure 0.11.0

//! HDF5 Extensible Array index parsing for chunked datasets (v4 index type 4).
//!
//! Extensible Arrays are used for datasets with exactly one unlimited dimension.
//! Structures: AEHD (header), AEIB (index block), AEDB (data block), AESB (super block).

#[cfg(not(feature = "std"))]
extern crate alloc;

#[cfg(not(feature = "std"))]
use alloc::{format, vec, vec::Vec};

use crate::chunked_read::ChunkInfo;
use crate::convert::{TryToUsize, u32_from};
use crate::error::FormatError;
use crate::source::FileSource;

/// Parsed Extensible Array header (AEHD).
#[derive(Debug, Clone)]
pub struct ExtensibleArrayHeader {
    /// Client ID: 0 = non-filtered chunks, 1 = filtered chunks.
    pub client_id: u8,
    /// Size of each array element in bytes.
    pub element_size: u8,
    /// Max number of elements bits (log2 of the max number of data block elements per page).
    pub max_nelmts_bits: u8,
    /// Number of elements in the index block.
    pub idx_blk_elmts: u8,
    /// Minimum number of data block elements.
    pub min_dblk_nelmts: u8,
    /// Minimum number of elements in a super block.
    pub super_blk_min_nelmts: u8,
    /// Max number of data block elements bits.
    pub max_dblk_nelmts_bits: u8,
    /// Total number of elements stored.
    pub num_elements: u64,
    /// Address of the index block.
    pub index_block_address: u64,
}

fn read_offset(data: &[u8], pos: usize, size: u8) -> Result<u64, FormatError> {
    let s = size as usize;
    if pos + s > data.len() {
        return Err(FormatError::UnexpectedEof {
            expected: pos + s,
            available: data.len(),
        });
    }
    let slice = &data[pos..pos + s];
    Ok(match size {
        2 => u16::from_le_bytes([slice[0], slice[1]]) as u64,
        4 => u32::from_le_bytes([slice[0], slice[1], slice[2], slice[3]]) as u64,
        8 => u64::from_le_bytes([
            slice[0], slice[1], slice[2], slice[3], slice[4], slice[5], slice[6], slice[7],
        ]),
        _ => return Err(FormatError::InvalidOffsetSize(size)),
    })
}

fn is_undefined_addr(addr: u64, offset_size: u8) -> bool {
    match offset_size {
        2 => addr == 0xFFFF,
        4 => addr == 0xFFFF_FFFF,
        8 => addr == 0xFFFF_FFFF_FFFF_FFFF,
        _ => false,
    }
}

fn is_undefined(data: &[u8], pos: usize, size: u8) -> bool {
    let s = size as usize;
    if pos + s > data.len() {
        return false;
    }
    data[pos..pos + s].iter().all(|&b| b == 0xFF)
}

fn read_variable_length(data: &[u8], size: usize) -> Result<u64, FormatError> {
    if size > 8 || data.len() < size {
        return Err(FormatError::ChunkedReadError(
            "invalid variable-length size".into(),
        ));
    }
    let mut val = 0u64;
    for (i, &byte) in data.iter().enumerate().take(size) {
        val |= (byte as u64) << (i * 8);
    }
    Ok(val)
}

impl ExtensibleArrayHeader {
    /// Parse an Extensible Array header from file data at the given offset.
    pub fn parse(
        file_data: &[u8],
        offset: usize,
        offset_size: u8,
        length_size: u8,
    ) -> Result<Self, FormatError> {
        // EAHD: signature(4) + version(1) + client_id(1) + element_size(1) +
        //   max_nelmts_bits(1) + idx_blk_elmts(1) + min_dblk_nelmts(1) +
        //   super_blk_min_nelmts(1) + max_dblk_nelmts_bits(1) +
        //   6 stats fields (each length_size) + index_block_address(offset_size) + checksum(4)
        let min_size =
            4 + 1 + 1 + 1 + 1 + 1 + 1 + 1 + 1 + 6 * length_size as usize + offset_size as usize + 4;
        if offset + min_size > file_data.len() {
            return Err(FormatError::UnexpectedEof {
                expected: offset + min_size,
                available: file_data.len(),
            });
        }

        let d = &file_data[offset..];
        if &d[0..4] != b"EAHD" {
            return Err(FormatError::ChunkedReadError(
                "invalid Extensible Array header signature".into(),
            ));
        }

        let version = d[4];
        if version != 0 {
            return Err(FormatError::ChunkedReadError(format!(
                "unsupported Extensible Array header version: {version}"
            )));
        }

        let client_id = d[5];
        let element_size = d[6];
        let max_nelmts_bits = d[7];
        let idx_blk_elmts = d[8];
        let min_dblk_nelmts = d[9];
        let super_blk_min_nelmts = d[10];
        let max_dblk_nelmts_bits = d[11];

        let mut pos = 12;
        // 6 statistics fields, in the HDF5 C library's stored order:
        //   [0] nsuper_blks   [1] super_blk_size   [2] ndata_blks
        //   [3] data_blk_size [4] max_idx_set      [5] nelmts
        // We read max_idx_set (field [4]) — the count of elements that have been
        // set, which bounds the live region. For a densely written array it
        // equals the chunk count; `nelmts` (field [5]) is the larger
        // rounded-up-to-block-boundary slot count, which we do not need.
        let ls = length_size as usize;
        pos += 4 * ls; // skip [0]..[3]
        let num_elements = read_offset(d, pos, length_size)?; // [4] max_idx_set
        pos += ls;
        pos += ls; // skip [5] nelmts
        let index_block_address = read_offset(d, pos, offset_size)?;

        Ok(ExtensibleArrayHeader {
            client_id,
            element_size,
            max_nelmts_bits,
            idx_blk_elmts,
            min_dblk_nelmts,
            super_blk_min_nelmts,
            max_dblk_nelmts_bits,
            num_elements,
            index_block_address,
        })
    }

    /// Compute the size of this header in bytes (for write support).
    pub fn serialized_size(offset_size: u8, length_size: u8) -> usize {
        4 + 1 + 1 + 1 + 1 + 1 + 1 + 1 + 1 + 6 * length_size as usize + offset_size as usize + 4
    }

    /// Parse an Extensible Array header from a [`FileSource`] (bounded window).
    pub fn parse_from_source<S: FileSource + ?Sized>(
        source: &S,
        address: u64,
        offset_size: u8,
        length_size: u8,
    ) -> Result<Self, FormatError> {
        let size = Self::serialized_size(offset_size, length_size);
        let buf = source.read_exact_at(address, size)?;
        Self::parse(&buf, 0, offset_size, length_size)
    }
}

/// Per-element stride in an Extensible Array data block: just the address for
/// non-filtered chunks, or the full filtered element record otherwise.
fn ea_elem_stride(header: &ExtensibleArrayHeader, offset_size: u8) -> usize {
    if header.client_id == 0 {
        offset_size as usize
    } else {
        header.element_size as usize
    }
}

/// Geometry of an Extensible Array, derived from its header creation parameters.
///
/// This is the single source of truth for the super-block / data-block size
/// progression, shared by both the reader (this module) and the writer
/// (`chunked_write`). It mirrors libhdf5's `H5EA__hdr_init`: starting from
/// `(ndblks = 1, dblk_nelmts = min_dblk_nelmts)`, each successive super block
/// alternately doubles the elements-per-data-block (even index) and the
/// number-of-data-blocks (odd index):
///
/// ```text
///   super block i:  ndblks = 2^floor(i/2)   dblk_nelmts = min_dblk * 2^ceil(i/2)
///   SB0: 1 x 16   SB1: 1 x 32   SB2: 2 x 32   SB3: 2 x 64   SB4: 4 x 64 ...
/// ```
///
/// The first `super_blk_min_nelmts` super blocks have their data-block
/// addresses stored directly in the index block; the rest are reached through
/// on-disk super blocks (`EASB`) whose addresses are stored in the index block.
#[derive(Debug, Clone)]
pub(crate) struct EaGeometry {
    /// `(ndblks, dblk_nelmts)` for each super block index `0..nsblks`.
    pub sblks: Vec<(u64, u64)>,
    /// Element count of each direct data block whose address is stored in the
    /// index block. `len()` equals the number of direct data-block pointers.
    pub direct_dblk_nelmts: Vec<u64>,
    /// Number of super-block pointers stored in the index block.
    pub nsblk_addrs: usize,
    /// Super-block index of the first super block reached via a super-block
    /// pointer in the index block (i.e. `super_blk_min_nelmts`).
    pub first_indirect_sblk: usize,
}

impl EaGeometry {
    /// Derive the geometry from a parsed header.
    pub fn from_header(h: &ExtensibleArrayHeader) -> Self {
        let min_dblk = h.min_dblk_nelmts as u64;
        let sup_blk_min = h.super_blk_min_nelmts as usize;
        // nsblks = max_nelmts_bits - log2(min_dblk_nelmts) + 1
        let log2_min = if min_dblk <= 1 {
            0
        } else {
            min_dblk.trailing_zeros() as u64
        };
        let nsblks = (h.max_nelmts_bits as u64).saturating_sub(log2_min) as usize + 1;

        let mut sblks = Vec::with_capacity(nsblks);
        let mut ndblks = 1u64;
        let mut dblk_nelmts = min_dblk;
        for u in 0..nsblks {
            sblks.push((ndblks, dblk_nelmts));
            if u % 2 == 0 {
                dblk_nelmts = dblk_nelmts.saturating_mul(2);
            } else {
                ndblks = ndblks.saturating_mul(2);
            }
        }

        let mut direct_dblk_nelmts = Vec::new();
        for sb in sblks.iter().take(sup_blk_min.min(nsblks)) {
            let (nd, dn) = *sb;
            for _ in 0..nd {
                direct_dblk_nelmts.push(dn);
            }
        }

        let nsblk_addrs = nsblks.saturating_sub(sup_blk_min);
        EaGeometry {
            sblks,
            direct_dblk_nelmts,
            nsblk_addrs,
            first_indirect_sblk: sup_blk_min,
        }
    }
}

/// Read a single element from the extensible array element data.
/// Returns (chunk_info, bytes_consumed) or None if unallocated.
#[allow(clippy::too_many_arguments)]
fn read_element(
    data: &[u8],
    pos: usize,
    client_id: u8,
    element_size: u8,
    offset_size: u8,
    chunk_byte_size: u64,
    linear_index: usize,
    num_chunks_per_dim: &[u64],
    chunk_dimensions: &[u32],
) -> Result<(Option<ChunkInfo>, usize), FormatError> {
    let os = offset_size as usize;

    if client_id == 0 {
        // Non-filtered: just address
        if pos + os > data.len() {
            return Err(FormatError::UnexpectedEof {
                expected: pos + os,
                available: data.len(),
            });
        }
        if is_undefined(data, pos, offset_size) {
            return Ok((None, os));
        }
        let address = read_offset(data, pos, offset_size)?;
        let offsets = index_to_chunk_offsets(linear_index, num_chunks_per_dim, chunk_dimensions);
        Ok((
            Some(ChunkInfo {
                chunk_size: u32_from(chunk_byte_size)?,
                filter_mask: 0,
                offsets,
                address,
            }),
            os,
        ))
    } else {
        // Filtered: address + compressed_size + filter_mask
        let chunk_size_bytes = element_size as usize - os - 4;
        let elem_total = os + chunk_size_bytes + 4;
        if pos + elem_total > data.len() {
            return Err(FormatError::UnexpectedEof {
                expected: pos + elem_total,
                available: data.len(),
            });
        }
        if is_undefined(data, pos, offset_size) {
            return Ok((None, elem_total));
        }
        let address = read_offset(data, pos, offset_size)?;
        let chunk_size = read_variable_length(&data[pos + os..], chunk_size_bytes)?;
        let fm_off = pos + os + chunk_size_bytes;
        let filter_mask = u32::from_le_bytes([
            data[fm_off],
            data[fm_off + 1],
            data[fm_off + 2],
            data[fm_off + 3],
        ]);
        let offsets = index_to_chunk_offsets(linear_index, num_chunks_per_dim, chunk_dimensions);
        Ok((
            Some(ChunkInfo {
                chunk_size: u32_from(chunk_size)?,
                filter_mask,
                offsets,
                address,
            }),
            elem_total,
        ))
    }
}

/// Convert a linear chunk index to N-dimensional chunk offsets in dataset space.
fn index_to_chunk_offsets(
    index: usize,
    num_chunks_per_dim: &[u64],
    chunk_dimensions: &[u32],
) -> Vec<u64> {
    let rank = num_chunks_per_dim.len();
    let mut offsets = vec![0u64; rank];
    let mut remaining = index as u64;
    for d in (0..rank).rev() {
        let nchunks = num_chunks_per_dim[d];
        let chunk_idx = remaining % nchunks;
        remaining /= nchunks;
        offsets[d] = chunk_idx * chunk_dimensions[d] as u64;
    }
    offsets
}

/// Collect elements from a data block at the given offset.
#[allow(clippy::too_many_arguments)]
fn read_data_block_elements(
    file_data: &[u8],
    db_offset: usize,
    nelmts: usize,
    header: &ExtensibleArrayHeader,
    offset_size: u8,
    chunk_byte_size: u64,
    start_index: usize,
    total_elements: usize,
    num_chunks_per_dim: &[u64],
    chunk_dimensions: &[u32],
) -> Result<Vec<ChunkInfo>, FormatError> {
    // AEDB: signature(4) + version(1) + client_id(1) + header_address(offset_size)
    let db_header_size = 4 + 1 + 1 + offset_size as usize;
    if db_offset + db_header_size > file_data.len() {
        return Err(FormatError::UnexpectedEof {
            expected: db_offset + db_header_size,
            available: file_data.len(),
        });
    }

    let d = &file_data[db_offset..];
    if &d[0..4] != b"EADB" {
        return Err(FormatError::ChunkedReadError(
            "invalid Extensible Array data block signature".into(),
        ));
    }
    // Skip version(1) + client_id(1) + header_address(offset_size) + block_offset
    // Block offset is encoded in ceil(max_nelmts_bits/8) bytes
    let blk_off_size = (header.max_nelmts_bits as usize).div_ceil(8);
    let mut pos = db_offset + db_header_size + blk_off_size;

    // Direct data blocks and non-paged super-block data blocks store their
    // elements inline. Paged data blocks (only ever found inside a super block
    // whose `dblk_nelmts` exceeds the page size) are handled separately by
    // `read_paged_data_block`, which is driven by the page-init bitmap stored
    // in the owning super block.
    // Read only as many elements as the dataset count allows. A SWMR writer may
    // have written element slots beyond the published count (it grows the block
    // before bumping the count); those must be ignored, otherwise their linear
    // indices wrap modulo the current chunk-grid and overwrite earlier chunks.
    let limit = total_elements.saturating_sub(start_index).min(nelmts);
    let mut chunks = Vec::new();
    for i in 0..limit {
        let (info, consumed) = read_element(
            file_data,
            pos,
            header.client_id,
            header.element_size,
            offset_size,
            chunk_byte_size,
            start_index + i,
            num_chunks_per_dim,
            chunk_dimensions,
        )?;
        if let Some(ci) = info {
            chunks.push(ci);
        }
        pos += consumed;
    }

    Ok(chunks)
}

/// Test whether page `page_idx` is initialized in a super-block page-init
/// bitmap. The bitmap is a contiguous, MSB-first bit stream: page 0 is bit 7 of
/// byte 0, page 1 is bit 6, page 8 is bit 7 of byte 1, and so on.
fn page_is_initialized(bitmap: &[u8], page_idx: usize) -> bool {
    let byte = page_idx / 8;
    let mask = 0x80u8 >> (page_idx % 8);
    byte < bitmap.len() && (bitmap[byte] & mask) != 0
}

/// Read a paged Extensible Array data block.
///
/// A paged data block has a 22-byte header (signature, version, client id,
/// header address, block offset) *followed by its own checksum*, after which
/// the data block's pages are laid out contiguously. Each page holds
/// `page_nelmts` element slots followed by a 4-byte checksum, and is written
/// only when initialized. Whether page `p` of this data block is initialized is
/// recorded in the owning super block's bitmap at global page index
/// `db_local_idx * npages + p`. Pages are filled sequentially, so the first
/// uninitialized page marks the end of populated data for this block.
#[allow(clippy::too_many_arguments)]
fn read_paged_data_block(
    file_data: &[u8],
    db_offset: usize,
    page_nelmts: usize,
    npages: usize,
    db_local_idx: usize,
    page_bitmap: &[u8],
    header: &ExtensibleArrayHeader,
    offset_size: u8,
    chunk_byte_size: u64,
    start_index: usize,
    total_elements: usize,
    num_chunks_per_dim: &[u64],
    chunk_dimensions: &[u32],
) -> Result<Vec<ChunkInfo>, FormatError> {
    let blk_off_size = (header.max_nelmts_bits as usize).div_ceil(8);
    // Header includes its own checksum: sig(4)+ver(1)+cid(1)+hdr_addr+block_offset+checksum(4)
    let db_header_size = 4 + 1 + 1 + offset_size as usize + blk_off_size + 4;
    if db_offset + db_header_size > file_data.len() {
        return Err(FormatError::UnexpectedEof {
            expected: db_offset + db_header_size,
            available: file_data.len(),
        });
    }
    if &file_data[db_offset..db_offset + 4] != b"EADB" {
        return Err(FormatError::ChunkedReadError(
            "invalid Extensible Array data block signature".into(),
        ));
    }

    let mut chunks = Vec::new();
    let mut pos = db_offset + db_header_size;
    for page in 0..npages {
        let global_page = db_local_idx * npages + page;
        if !page_is_initialized(page_bitmap, global_page) {
            // Pages are populated sequentially; nothing initialized after this.
            break;
        }
        let page_start = start_index + page * page_nelmts;
        // Ignore element slots beyond the published count (a writer may have
        // written ahead of bumping the count); their wrapped linear indices
        // would otherwise overwrite earlier chunks.
        let limit = total_elements.saturating_sub(page_start).min(page_nelmts);
        for i in 0..limit {
            let (info, consumed) = read_element(
                file_data,
                pos,
                header.client_id,
                header.element_size,
                offset_size,
                chunk_byte_size,
                page_start + i,
                num_chunks_per_dim,
                chunk_dimensions,
            )?;
            if let Some(ci) = info {
                chunks.push(ci);
            }
            pos += consumed;
        }
        if limit < page_nelmts {
            break; // remaining slots/pages are beyond the dataset count
        }
        // Skip the page checksum.
        pos += 4;
    }

    Ok(chunks)
}

/// Read chunk records from an Extensible Array.
///
/// Traverses AEHD -> AEIB -> AEDB/AESB to collect all allocated chunks.
#[allow(clippy::too_many_arguments)]
pub fn read_extensible_array_chunks(
    file_data: &[u8],
    header: &ExtensibleArrayHeader,
    dataset_dims: &[u64],
    chunk_dimensions: &[u32],
    element_size: u32,
    offset_size: u8,
    _length_size: u8,
) -> Result<Vec<ChunkInfo>, FormatError> {
    let rank = chunk_dimensions.len();
    let os = offset_size as usize;

    let mut num_chunks_per_dim = Vec::with_capacity(rank);
    for d in 0..rank {
        let ds_dim = dataset_dims[d];
        let ch_dim = chunk_dimensions[d] as u64;
        num_chunks_per_dim.push(ds_dim.div_ceil(ch_dim));
    }

    let chunk_byte_size: u64 =
        chunk_dimensions.iter().map(|&d| d as u64).product::<u64>() * element_size as u64;

    // Parse index block (AEIB)
    let ib_offset = header.index_block_address.to_usize()?;
    let ib_header_size = 4 + 1 + 1 + offset_size as usize; // sig + ver + client + hdr_addr
    if ib_offset + ib_header_size > file_data.len() {
        return Err(FormatError::UnexpectedEof {
            expected: ib_offset + ib_header_size,
            available: file_data.len(),
        });
    }

    let ib = &file_data[ib_offset..];
    if &ib[0..4] != b"EAIB" {
        return Err(FormatError::ChunkedReadError(
            "invalid Extensible Array index block signature".into(),
        ));
    }
    // Skip version(1) + client_id(1) + header_address(offset_size)
    let mut pos = ib_offset + ib_header_size;

    let mut chunks = Vec::new();
    let mut global_index = 0usize;
    // Bound the traversal by the dataset's dimensions, not just the array's
    // stored element count. A SWMR writer publishes the grown chunk index
    // (header element count) before the grown dataspace dimension; reading only
    // as many chunks as the current dimensions imply means an interrupted append
    // (index ahead of dimensions) yields a consistent prefix rather than chunks
    // beyond the dataset bounds.
    let dims_chunks: usize = num_chunks_per_dim.iter().product::<u64>().to_usize()?;
    let total_elements = (header.num_elements.to_usize()?).min(dims_chunks);

    // 1. Read inline elements in index block
    let n_inline = header.idx_blk_elmts as usize;
    for i in 0..n_inline {
        if global_index + i >= total_elements {
            break;
        }
        let (info, consumed) = read_element(
            file_data,
            pos,
            header.client_id,
            header.element_size,
            offset_size,
            chunk_byte_size,
            global_index + i,
            &num_chunks_per_dim,
            chunk_dimensions,
        )?;
        if let Some(ci) = info {
            chunks.push(ci);
        }
        pos += consumed;
    }
    global_index += n_inline.min(total_elements);

    // If all elements were inline, we're done
    if global_index >= total_elements {
        return Ok(chunks);
    }

    // Derive the (shared) extensible-array geometry from the header. This is
    // the same progression the writer uses, so reader and writer cannot drift.
    let geom = EaGeometry::from_header(header);

    // 2. Direct data blocks: their addresses are listed in the index block,
    //    one per entry in `geom.direct_dblk_nelmts`.
    let mut direct_addrs: Vec<u64> = Vec::with_capacity(geom.direct_dblk_nelmts.len());
    for _ in 0..geom.direct_dblk_nelmts.len() {
        direct_addrs.push(read_offset(file_data, pos, offset_size)?);
        pos += os;
    }
    for (i, &addr) in direct_addrs.iter().enumerate() {
        if global_index >= total_elements {
            break;
        }
        let nelmts = geom.direct_dblk_nelmts[i].to_usize()?;
        if !is_undefined_addr(addr, offset_size) {
            let block_chunks = read_data_block_elements(
                file_data,
                addr.to_usize()?,
                nelmts,
                header,
                offset_size,
                chunk_byte_size,
                global_index,
                total_elements,
                &num_chunks_per_dim,
                chunk_dimensions,
            )?;
            chunks.extend(block_chunks);
        }
        // Advance even for undefined blocks so linear indices stay aligned.
        global_index += nelmts;
    }

    // 3. Super blocks: the remaining `geom.nsblk_addrs` index-block entries are
    //    addresses of on-disk super blocks (`EASB`). Super-block pointer `j`
    //    refers to super block `first_indirect_sblk + j`.
    let mut sblk_addrs: Vec<u64> = Vec::with_capacity(geom.nsblk_addrs);
    for _ in 0..geom.nsblk_addrs {
        if pos + os > file_data.len() {
            break;
        }
        sblk_addrs.push(read_offset(file_data, pos, offset_size)?);
        pos += os;
    }
    for (j, &sb_addr) in sblk_addrs.iter().enumerate() {
        if global_index >= total_elements {
            break;
        }
        let sblk_idx = geom.first_indirect_sblk + j;
        let (ndblks, dblk_nelmts) = geom.sblks[sblk_idx];
        let total_in_sb = (ndblks * dblk_nelmts).to_usize()?;
        if !is_undefined_addr(sb_addr, offset_size) {
            let sb_chunks = read_super_block(
                file_data,
                sb_addr.to_usize()?,
                ndblks.to_usize()?,
                dblk_nelmts.to_usize()?,
                header,
                offset_size,
                chunk_byte_size,
                global_index,
                total_elements,
                &num_chunks_per_dim,
                chunk_dimensions,
            )?;
            chunks.extend(sb_chunks);
        }
        global_index += total_in_sb;
    }

    Ok(chunks)
}

/// Read a super block (AESB) and its data blocks.
#[allow(clippy::too_many_arguments)]
fn read_super_block(
    file_data: &[u8],
    sb_offset: usize,
    ndblks: usize,
    nelmts_per_dblk: usize,
    header: &ExtensibleArrayHeader,
    offset_size: u8,
    chunk_byte_size: u64,
    start_index: usize,
    total_elements: usize,
    num_chunks_per_dim: &[u64],
    chunk_dimensions: &[u32],
) -> Result<Vec<ChunkInfo>, FormatError> {
    let os = offset_size as usize;

    // AESB: signature(4) + version(1) + client_id(1) + header_address(offset_size)
    //       + block_offset(ceil(max_nelmts_bits/8))
    //       + [page-init bitmap, if data blocks are paged]
    //       + data block addresses
    //       + checksum
    let blk_off_size = (header.max_nelmts_bits as usize).div_ceil(8);
    let sb_header_size = 4 + 1 + 1 + os + blk_off_size;
    if sb_offset + sb_header_size > file_data.len() {
        return Err(FormatError::UnexpectedEof {
            expected: sb_offset + sb_header_size,
            available: file_data.len(),
        });
    }

    if &file_data[sb_offset..sb_offset + 4] != b"EASB" {
        return Err(FormatError::ChunkedReadError(
            "invalid Extensible Array super block signature".into(),
        ));
    }

    let mut pos = sb_offset + sb_header_size;

    // A super block's data blocks are paged when their element count exceeds
    // the page size. When paged, a page-init bitmap precedes the data block
    // addresses. Its size is `ndblks * ceil(npages / 8)` bytes, and it is a
    // contiguous bit stream indexed by `db_local_idx * npages + page`.
    let page_nelmts = 1usize << header.max_dblk_nelmts_bits;
    let is_paged = nelmts_per_dblk > page_nelmts;
    let npages = if is_paged {
        nelmts_per_dblk / page_nelmts
    } else {
        0
    };
    let page_bitmap: Vec<u8> = if is_paged {
        let bitmap_size = ndblks * npages.div_ceil(8);
        if pos + bitmap_size > file_data.len() {
            return Err(FormatError::UnexpectedEof {
                expected: pos + bitmap_size,
                available: file_data.len(),
            });
        }
        let bm = file_data[pos..pos + bitmap_size].to_vec();
        pos += bitmap_size;
        bm
    } else {
        Vec::new()
    };

    // Read data block addresses.
    let mut dblk_addrs: Vec<u64> = Vec::with_capacity(ndblks);
    for _ in 0..ndblks {
        if pos + os > file_data.len() {
            return Err(FormatError::UnexpectedEof {
                expected: pos + os,
                available: file_data.len(),
            });
        }
        let addr = read_offset(file_data, pos, offset_size)?;
        dblk_addrs.push(addr);
        pos += os;
    }

    let mut chunks = Vec::new();
    let mut global_idx = start_index;

    for (db_local, &addr) in dblk_addrs.iter().enumerate() {
        if !is_undefined_addr(addr, offset_size) {
            let block_chunks = if is_paged {
                read_paged_data_block(
                    file_data,
                    addr.to_usize()?,
                    page_nelmts,
                    npages,
                    db_local,
                    &page_bitmap,
                    header,
                    offset_size,
                    chunk_byte_size,
                    global_idx,
                    total_elements,
                    num_chunks_per_dim,
                    chunk_dimensions,
                )?
            } else {
                read_data_block_elements(
                    file_data,
                    addr.to_usize()?,
                    nelmts_per_dblk,
                    header,
                    offset_size,
                    chunk_byte_size,
                    global_idx,
                    total_elements,
                    num_chunks_per_dim,
                    chunk_dimensions,
                )?
            };
            chunks.extend(block_chunks);
        }
        global_idx += nelmts_per_dblk;
    }

    Ok(chunks)
}

// ---------------------------------------------------------------------------
// Streaming traversal (read each block from a `FileSource` on demand)
// ---------------------------------------------------------------------------

/// Read chunk records from an Extensible Array via a [`FileSource`].
///
/// Streaming counterpart of [`read_extensible_array_chunks`]: reads the index
/// block, then each direct data block and super block (and its paged data
/// blocks) as bounded windows via `read_at`. The shared `read_element` /
/// `EaGeometry` drive the decoding, so the layout logic stays in one place.
#[allow(clippy::too_many_arguments)]
pub fn read_extensible_array_chunks_from_source<S: FileSource + ?Sized>(
    source: &S,
    header: &ExtensibleArrayHeader,
    dataset_dims: &[u64],
    chunk_dimensions: &[u32],
    element_size: u32,
    offset_size: u8,
    _length_size: u8,
) -> Result<Vec<ChunkInfo>, FormatError> {
    let rank = chunk_dimensions.len();
    let os = offset_size as usize;

    let mut num_chunks_per_dim = Vec::with_capacity(rank);
    for d in 0..rank {
        let ch_dim = chunk_dimensions[d] as u64;
        num_chunks_per_dim.push(dataset_dims[d].div_ceil(ch_dim));
    }
    let chunk_byte_size: u64 =
        chunk_dimensions.iter().map(|&d| d as u64).product::<u64>() * element_size as u64;

    let dims_chunks: usize = num_chunks_per_dim.iter().product::<u64>().to_usize()?;
    let total_elements = header.num_elements.to_usize()?.min(dims_chunks);

    let geom = EaGeometry::from_header(header);
    let elem_stride = ea_elem_stride(header, offset_size);

    // Read the whole index block: header + inline element slots + direct
    // data-block addresses + super-block addresses.
    let ib_header_size = 4 + 1 + 1 + os;
    let n_inline = header.idx_blk_elmts as usize;
    let ndirect = geom.direct_dblk_nelmts.len();
    let nsblk = geom.nsblk_addrs;
    let inline_bytes = n_inline
        .checked_mul(elem_stride)
        .ok_or(FormatError::OffsetOverflow {
            offset: n_inline as u64,
            length: elem_stride as u64,
        })?;
    let addr_bytes = (ndirect + nsblk)
        .checked_mul(os)
        .ok_or(FormatError::OffsetOverflow {
            offset: (ndirect + nsblk) as u64,
            length: os as u64,
        })?;
    let ib_len = ib_header_size + inline_bytes + addr_bytes;
    let ib = source.read_exact_at(header.index_block_address, ib_len)?;
    if &ib[0..4] != b"EAIB" {
        return Err(FormatError::ChunkedReadError(
            "invalid Extensible Array index block signature".into(),
        ));
    }

    let mut pos = ib_header_size;
    let mut chunks = Vec::new();
    let mut global_index = 0usize;

    // 1. Inline elements stored directly in the index block.
    for i in 0..n_inline {
        if global_index + i >= total_elements {
            break;
        }
        let (info, consumed) = read_element(
            &ib,
            pos,
            header.client_id,
            header.element_size,
            offset_size,
            chunk_byte_size,
            global_index + i,
            &num_chunks_per_dim,
            chunk_dimensions,
        )?;
        if let Some(ci) = info {
            chunks.push(ci);
        }
        pos += consumed;
    }
    global_index += n_inline.min(total_elements);
    if global_index >= total_elements {
        return Ok(chunks);
    }

    // After all inline slots, `pos` sits at the direct data-block addresses.
    let mut direct_addrs: Vec<u64> = Vec::with_capacity(ndirect);
    for _ in 0..ndirect {
        direct_addrs.push(read_offset(&ib, pos, offset_size)?);
        pos += os;
    }
    for (i, &addr) in direct_addrs.iter().enumerate() {
        if global_index >= total_elements {
            break;
        }
        let nelmts = geom.direct_dblk_nelmts[i].to_usize()?;
        if !is_undefined_addr(addr, offset_size) {
            chunks.extend(read_data_block_elements_from_source(
                source,
                addr,
                nelmts,
                header,
                offset_size,
                chunk_byte_size,
                global_index,
                total_elements,
                &num_chunks_per_dim,
                chunk_dimensions,
            )?);
        }
        global_index += nelmts;
    }

    // 3. Super-block addresses.
    let mut sblk_addrs: Vec<u64> = Vec::with_capacity(nsblk);
    for _ in 0..nsblk {
        sblk_addrs.push(read_offset(&ib, pos, offset_size)?);
        pos += os;
    }
    for (j, &sb_addr) in sblk_addrs.iter().enumerate() {
        if global_index >= total_elements {
            break;
        }
        let sblk_idx = geom.first_indirect_sblk + j;
        let (ndblks, dblk_nelmts) = geom.sblks[sblk_idx];
        let total_in_sb = (ndblks * dblk_nelmts).to_usize()?;
        if !is_undefined_addr(sb_addr, offset_size) {
            chunks.extend(read_super_block_from_source(
                source,
                sb_addr,
                ndblks.to_usize()?,
                dblk_nelmts.to_usize()?,
                header,
                offset_size,
                chunk_byte_size,
                global_index,
                total_elements,
                &num_chunks_per_dim,
                chunk_dimensions,
            )?);
        }
        global_index += total_in_sb;
    }

    Ok(chunks)
}

/// Collect elements from a (non-paged) data block read from the source.
#[allow(clippy::too_many_arguments)]
fn read_data_block_elements_from_source<S: FileSource + ?Sized>(
    source: &S,
    db_address: u64,
    nelmts: usize,
    header: &ExtensibleArrayHeader,
    offset_size: u8,
    chunk_byte_size: u64,
    start_index: usize,
    total_elements: usize,
    num_chunks_per_dim: &[u64],
    chunk_dimensions: &[u32],
) -> Result<Vec<ChunkInfo>, FormatError> {
    let os = offset_size as usize;
    let db_header_size = 4 + 1 + 1 + os;
    let blk_off_size = (header.max_nelmts_bits as usize).div_ceil(8);
    let elem_stride = ea_elem_stride(header, offset_size);
    let limit = total_elements.saturating_sub(start_index).min(nelmts);

    // Prefix + block-offset field + `limit` element slots (we only decode up to
    // the published count).
    let elem_bytes = limit
        .checked_mul(elem_stride)
        .ok_or(FormatError::OffsetOverflow {
            offset: limit as u64,
            length: elem_stride as u64,
        })?;
    let region_len = db_header_size + blk_off_size + elem_bytes;
    let block = source.read_exact_at(db_address, region_len)?;
    if &block[0..4] != b"EADB" {
        return Err(FormatError::ChunkedReadError(
            "invalid Extensible Array data block signature".into(),
        ));
    }

    let mut pos = db_header_size + blk_off_size;
    let mut chunks = Vec::new();
    for i in 0..limit {
        let (info, consumed) = read_element(
            &block,
            pos,
            header.client_id,
            header.element_size,
            offset_size,
            chunk_byte_size,
            start_index + i,
            num_chunks_per_dim,
            chunk_dimensions,
        )?;
        if let Some(ci) = info {
            chunks.push(ci);
        }
        pos += consumed;
    }
    Ok(chunks)
}

/// Read a paged Extensible Array data block from the source.
#[allow(clippy::too_many_arguments)]
fn read_paged_data_block_from_source<S: FileSource + ?Sized>(
    source: &S,
    db_address: u64,
    page_nelmts: usize,
    npages: usize,
    db_local_idx: usize,
    page_bitmap: &[u8],
    header: &ExtensibleArrayHeader,
    offset_size: u8,
    chunk_byte_size: u64,
    start_index: usize,
    total_elements: usize,
    num_chunks_per_dim: &[u64],
    chunk_dimensions: &[u32],
) -> Result<Vec<ChunkInfo>, FormatError> {
    let blk_off_size = (header.max_nelmts_bits as usize).div_ceil(8);
    let db_header_size = 4 + 1 + 1 + offset_size as usize + blk_off_size + 4;
    let elem_stride = ea_elem_stride(header, offset_size);
    let page_stride = page_nelmts
        .checked_mul(elem_stride)
        .and_then(|bytes| bytes.checked_add(4))
        .ok_or(FormatError::OffsetOverflow {
            offset: page_nelmts as u64,
            length: elem_stride as u64,
        })?;

    // Pages are populated sequentially; read only the leading initialized pages.
    let mut init_pages = 0usize;
    for page in 0..npages {
        if !page_is_initialized(page_bitmap, db_local_idx * npages + page) {
            break;
        }
        init_pages += 1;
    }
    // Clamp to the bytes actually available (the final partial page may omit its
    // trailing checksum).
    let avail = source
        .len()
        .saturating_sub(db_address)
        .to_usize()
        .unwrap_or(usize::MAX);
    let pages_bytes = init_pages
        .checked_mul(page_stride)
        .ok_or(FormatError::OffsetOverflow {
            offset: init_pages as u64,
            length: page_stride as u64,
        })?;
    let region_len = (db_header_size + pages_bytes).min(avail);
    let block = source.read_exact_at(db_address, region_len)?;
    if block.len() < 4 || &block[0..4] != b"EADB" {
        return Err(FormatError::ChunkedReadError(
            "invalid Extensible Array data block signature".into(),
        ));
    }

    let mut chunks = Vec::new();
    let mut pos = db_header_size;
    for page in 0..npages {
        let global_page = db_local_idx * npages + page;
        if !page_is_initialized(page_bitmap, global_page) {
            break;
        }
        let page_start = start_index + page * page_nelmts;
        let limit = total_elements.saturating_sub(page_start).min(page_nelmts);
        for i in 0..limit {
            let (info, consumed) = read_element(
                &block,
                pos,
                header.client_id,
                header.element_size,
                offset_size,
                chunk_byte_size,
                page_start + i,
                num_chunks_per_dim,
                chunk_dimensions,
            )?;
            if let Some(ci) = info {
                chunks.push(ci);
            }
            pos += consumed;
        }
        if limit < page_nelmts {
            break;
        }
        pos += 4; // page checksum
    }
    Ok(chunks)
}

/// Read a super block (AESB) and its data blocks from the source.
#[allow(clippy::too_many_arguments)]
fn read_super_block_from_source<S: FileSource + ?Sized>(
    source: &S,
    sb_address: u64,
    ndblks: usize,
    nelmts_per_dblk: usize,
    header: &ExtensibleArrayHeader,
    offset_size: u8,
    chunk_byte_size: u64,
    start_index: usize,
    total_elements: usize,
    num_chunks_per_dim: &[u64],
    chunk_dimensions: &[u32],
) -> Result<Vec<ChunkInfo>, FormatError> {
    let os = offset_size as usize;
    let blk_off_size = (header.max_nelmts_bits as usize).div_ceil(8);
    let sb_header_size = 4 + 1 + 1 + os + blk_off_size;

    let page_nelmts = 1usize << header.max_dblk_nelmts_bits;
    let is_paged = nelmts_per_dblk > page_nelmts;
    let npages = if is_paged {
        nelmts_per_dblk / page_nelmts
    } else {
        0
    };
    let bitmap_size = if is_paged {
        ndblks
            .checked_mul(npages.div_ceil(8))
            .ok_or(FormatError::OffsetOverflow {
                offset: ndblks as u64,
                length: npages.div_ceil(8) as u64,
            })?
    } else {
        0
    };

    // Header + (optional) page-init bitmap + data-block addresses.
    let addr_bytes = ndblks.checked_mul(os).ok_or(FormatError::OffsetOverflow {
        offset: ndblks as u64,
        length: os as u64,
    })?;
    let region_len = sb_header_size + bitmap_size + addr_bytes;
    let block = source.read_exact_at(sb_address, region_len)?;
    if &block[0..4] != b"EASB" {
        return Err(FormatError::ChunkedReadError(
            "invalid Extensible Array super block signature".into(),
        ));
    }

    let mut pos = sb_header_size;
    let page_bitmap: Vec<u8> = if is_paged {
        let bm = block[pos..pos + bitmap_size].to_vec();
        pos += bitmap_size;
        bm
    } else {
        Vec::new()
    };

    let mut dblk_addrs: Vec<u64> = Vec::with_capacity(ndblks);
    for _ in 0..ndblks {
        dblk_addrs.push(read_offset(&block, pos, offset_size)?);
        pos += os;
    }

    let mut chunks = Vec::new();
    let mut global_idx = start_index;
    for (db_local, &addr) in dblk_addrs.iter().enumerate() {
        if !is_undefined_addr(addr, offset_size) {
            let block_chunks = if is_paged {
                read_paged_data_block_from_source(
                    source,
                    addr,
                    page_nelmts,
                    npages,
                    db_local,
                    &page_bitmap,
                    header,
                    offset_size,
                    chunk_byte_size,
                    global_idx,
                    total_elements,
                    num_chunks_per_dim,
                    chunk_dimensions,
                )?
            } else {
                read_data_block_elements_from_source(
                    source,
                    addr,
                    nelmts_per_dblk,
                    header,
                    offset_size,
                    chunk_byte_size,
                    global_idx,
                    total_elements,
                    num_chunks_per_dim,
                    chunk_dimensions,
                )?
            };
            chunks.extend(block_chunks);
        }
        global_idx += nelmts_per_dblk;
    }

    Ok(chunks)
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn index_to_offsets_1d() {
        let num_chunks = vec![5u64];
        let chunk_dims = vec![20u32];
        assert_eq!(index_to_chunk_offsets(0, &num_chunks, &chunk_dims), vec![0]);
        assert_eq!(
            index_to_chunk_offsets(1, &num_chunks, &chunk_dims),
            vec![20]
        );
        assert_eq!(
            index_to_chunk_offsets(4, &num_chunks, &chunk_dims),
            vec![80]
        );
    }

    #[test]
    fn index_to_offsets_2d() {
        let num_chunks = vec![3u64, 2];
        let chunk_dims = vec![4u32, 3];
        assert_eq!(
            index_to_chunk_offsets(0, &num_chunks, &chunk_dims),
            vec![0, 0]
        );
        assert_eq!(
            index_to_chunk_offsets(1, &num_chunks, &chunk_dims),
            vec![0, 3]
        );
        assert_eq!(
            index_to_chunk_offsets(2, &num_chunks, &chunk_dims),
            vec![4, 0]
        );
    }

    #[test]
    fn parse_header_valid() {
        let os: u8 = 8;
        let ls: u8 = 8;
        let mut buf = vec![0u8; 256];
        buf[0..4].copy_from_slice(b"EAHD");
        buf[4] = 0; // version
        buf[5] = 0; // client_id = non-filtered
        buf[6] = 8; // element_size
        buf[7] = 10; // max_nelmts_bits
        buf[8] = 2; // idx_blk_elmts
        buf[9] = 4; // min_dblk_nelmts
        buf[10] = 2; // super_blk_min_nelmts
        buf[11] = 8; // max_dblk_nelmts_bits
        // 6 stats fields (each 8 bytes)
        buf[12..20].copy_from_slice(&0u64.to_le_bytes()); // stat[0]
        buf[20..28].copy_from_slice(&0u64.to_le_bytes()); // stat[1]
        buf[28..36].copy_from_slice(&0u64.to_le_bytes()); // stat[2]
        buf[36..44].copy_from_slice(&0u64.to_le_bytes()); // stat[3]
        buf[44..52].copy_from_slice(&5u64.to_le_bytes()); // stat[4] = num_elements
        buf[52..60].copy_from_slice(&0u64.to_le_bytes()); // stat[5]
        buf[60..68].copy_from_slice(&0x1000u64.to_le_bytes()); // index_block_address

        let hdr = ExtensibleArrayHeader::parse(&buf, 0, os, ls).unwrap();
        assert_eq!(hdr.client_id, 0);
        assert_eq!(hdr.element_size, 8);
        assert_eq!(hdr.idx_blk_elmts, 2);
        assert_eq!(hdr.min_dblk_nelmts, 4);
        assert_eq!(hdr.num_elements, 5);
        assert_eq!(hdr.index_block_address, 0x1000);
    }

    #[test]
    fn parse_header_invalid_signature() {
        let mut buf = vec![0u8; 256];
        buf[0..4].copy_from_slice(b"XXXX");
        let result = ExtensibleArrayHeader::parse(&buf, 0, 8, 8);
        assert!(result.is_err());
    }

    #[test]
    fn parse_header_invalid_version() {
        let mut buf = vec![0u8; 256];
        buf[0..4].copy_from_slice(b"EAHD");
        buf[4] = 1;
        let result = ExtensibleArrayHeader::parse(&buf, 0, 8, 8);
        assert!(result.is_err());
    }

    /// Build a synthetic Extensible Array with only inline elements (simplest case).
    /// All chunks fit in the index block.
    #[test]
    fn read_inline_only() {
        let os: u8 = 8;
        let ls: u8 = 8;
        let osv = os as usize;
        let num_chunks = 2usize;
        let chunk_byte_size = 20u64 * 8; // 20 elements × 8 bytes

        let mut file_data = vec![0u8; 0x3000];

        // AEHD at offset 0x100
        let aehd_offset = 0x100usize;
        let aeib_offset = 0x200usize;

        // Build AEHD
        file_data[aehd_offset..aehd_offset + 4].copy_from_slice(b"EAHD");
        file_data[aehd_offset + 4] = 0; // version
        file_data[aehd_offset + 5] = 0; // client_id = non-filtered
        file_data[aehd_offset + 6] = osv as u8; // element_size
        file_data[aehd_offset + 7] = 10; // max_nelmts_bits
        file_data[aehd_offset + 8] = num_chunks as u8; // idx_blk_elmts (all inline)
        file_data[aehd_offset + 9] = 4; // min_dblk_nelmts
        file_data[aehd_offset + 10] = 2; // super_blk_min_nelmts
        file_data[aehd_offset + 11] = 8; // max_dblk_nelmts_bits
        // 6 stats fields (each 8 bytes), nelmts at stat[4]
        file_data[aehd_offset + 44..aehd_offset + 52]
            .copy_from_slice(&(num_chunks as u64).to_le_bytes());
        file_data[aehd_offset + 60..aehd_offset + 68]
            .copy_from_slice(&(aeib_offset as u64).to_le_bytes());
        // checksum (4 bytes at +68) — not validated

        // Build AEIB at aeib_offset
        file_data[aeib_offset..aeib_offset + 4].copy_from_slice(b"EAIB");
        file_data[aeib_offset + 4] = 0; // version
        file_data[aeib_offset + 5] = 0; // client_id
        file_data[aeib_offset + 6..aeib_offset + 14]
            .copy_from_slice(&(aehd_offset as u64).to_le_bytes());

        // Inline elements
        let elem_start = aeib_offset + 6 + osv;
        let base_addr = 0x1000u64;
        for i in 0..num_chunks {
            let addr = base_addr + i as u64 * chunk_byte_size;
            let p = elem_start + i * osv;
            file_data[p..p + osv].copy_from_slice(&addr.to_le_bytes());
        }

        let header = ExtensibleArrayHeader::parse(&file_data, aehd_offset, os, ls).unwrap();
        let ds_dims = vec![40u64]; // 2 chunks × 20 elements
        let chunk_dims = vec![20u32];
        let chunks =
            read_extensible_array_chunks(&file_data, &header, &ds_dims, &chunk_dims, 8, os, ls)
                .unwrap();

        assert_eq!(chunks.len(), 2);
        assert_eq!(chunks[0].address, base_addr);
        assert_eq!(chunks[0].offsets, vec![0]);
        assert_eq!(chunks[0].chunk_size, chunk_byte_size as u32);
        assert_eq!(chunks[1].address, base_addr + chunk_byte_size);
        assert_eq!(chunks[1].offsets, vec![20]);

        #[cfg(feature = "std")]
        assert_ea_streams_match(&file_data, aehd_offset, &ds_dims, &chunk_dims, 8, os, ls);
    }

    /// Assert the streaming Extensible-Array reader matches the buffered one over
    /// both an in-memory and a `Read+Seek` source.
    #[cfg(feature = "std")]
    fn assert_ea_streams_match(
        file_data: &[u8],
        aehd_offset: usize,
        ds_dims: &[u64],
        chunk_dims: &[u32],
        element_size: u32,
        os: u8,
        ls: u8,
    ) {
        use crate::source::{BytesSource, ReadSeekSource};
        let h = ExtensibleArrayHeader::parse(file_data, aehd_offset, os, ls).unwrap();
        let buffered =
            read_extensible_array_chunks(file_data, &h, ds_dims, chunk_dims, element_size, os, ls)
                .unwrap();

        let mem = BytesSource::new(file_data);
        let hm =
            ExtensibleArrayHeader::parse_from_source(&mem, aehd_offset as u64, os, ls).unwrap();
        let from_mem = read_extensible_array_chunks_from_source(
            &mem,
            &hm,
            ds_dims,
            chunk_dims,
            element_size,
            os,
            ls,
        )
        .unwrap();

        let seek = ReadSeekSource::new(std::io::Cursor::new(file_data.to_vec())).unwrap();
        let hs =
            ExtensibleArrayHeader::parse_from_source(&seek, aehd_offset as u64, os, ls).unwrap();
        let from_seek = read_extensible_array_chunks_from_source(
            &seek,
            &hs,
            ds_dims,
            chunk_dims,
            element_size,
            os,
            ls,
        )
        .unwrap();

        assert_eq!(buffered, from_mem, "BytesSource mismatch");
        assert_eq!(buffered, from_seek, "ReadSeekSource mismatch");
    }

    /// Build a synthetic EA with inline elements + one direct data block.
    #[test]
    fn read_inline_plus_data_blocks() {
        let os: u8 = 8;
        let ls: u8 = 8;
        let osv = os as usize;
        let chunk_byte_size = 10u64 * 8; // 10 elements × 8 bytes
        let idx_blk_elmts = 2u8;
        let min_dblk_nelmts = 2u8;
        let sblk_min = 2u8;
        let total_chunks = 4usize; // 2 inline + 2 in data block (1 dblk from sb_level 0)

        let mut file_data = vec![0u8; 0x5000];
        let aehd_offset = 0x100usize;
        let aeib_offset = 0x200usize;
        let aedb_offset = 0x300usize;

        // EAHD
        file_data[aehd_offset..aehd_offset + 4].copy_from_slice(b"EAHD");
        file_data[aehd_offset + 4] = 0;
        file_data[aehd_offset + 5] = 0; // client_id
        file_data[aehd_offset + 6] = osv as u8; // element_size
        file_data[aehd_offset + 7] = 10;
        file_data[aehd_offset + 8] = idx_blk_elmts;
        file_data[aehd_offset + 9] = min_dblk_nelmts;
        file_data[aehd_offset + 10] = sblk_min;
        file_data[aehd_offset + 11] = 8;
        // 6 stats fields (each 8 bytes), nelmts at stat[4] (offset 12 + 4*8 = 44)
        file_data[aehd_offset + 44..aehd_offset + 52]
            .copy_from_slice(&(total_chunks as u64).to_le_bytes());
        // idx_blk_addr at offset 12 + 6*8 = 60
        file_data[aehd_offset + 60..aehd_offset + 68]
            .copy_from_slice(&(aeib_offset as u64).to_le_bytes());

        // AEIB
        file_data[aeib_offset..aeib_offset + 4].copy_from_slice(b"EAIB");
        file_data[aeib_offset + 4] = 0;
        file_data[aeib_offset + 5] = 0;
        file_data[aeib_offset + 6..aeib_offset + 14]
            .copy_from_slice(&(aehd_offset as u64).to_le_bytes());

        let mut pos = aeib_offset + 6 + osv;

        // Inline elements (2 chunks)
        let base_addr = 0x1000u64;
        for i in 0..idx_blk_elmts as usize {
            let addr = base_addr + i as u64 * chunk_byte_size;
            file_data[pos..pos + osv].copy_from_slice(&addr.to_le_bytes());
            pos += osv;
        }

        // Direct data block addresses: first sb_level=0 has 1 dblk, sb_level=1 has 1 dblk
        // Total direct dblks for sblk_min=2: 2^0 + 2^1 = 1 + 2 = 3 (oops)
        // Actually: sblk_min levels. level 0: 2^0=1 dblk, level 1: 2^1=2 dblks => 3 dblks
        // But we only have 2 remaining elements.
        // dblk sizes: level 0: 1 dblk of min_dblk=2; level 1: 2 dblks of 2 each (nelmts doubles at level > 0)
        // Wait, re-reading the code: at level 0, nelmts=min_dblk=2, 1 dblk.
        // At level 1, 1 dblk, nelmts still 2 (doubles only at level > 0... but the code says
        // `if sb_level > 0 { nelmts *= 2 }` after pushing). Let me re-check.
        // After push at level 0: nelmts=2. Then if 0>0 false, no double. Push 1 dblk of 2.
        // Level 1: ndblks=2. Push 2 dblks of 2. Then 1>0 true, nelmts=4.
        // Total: 3 dblks with sizes [2, 2, 2]. Total = 6.
        // We only need 2 more elements. So only the first dblk has data.
        let n_direct_dblks = 3;
        file_data[pos..pos + osv].copy_from_slice(&(aedb_offset as u64).to_le_bytes());
        pos += osv;
        // 2 more dblk addresses - undefined
        for _ in 1..n_direct_dblks {
            file_data[pos..pos + osv].copy_from_slice(&u64::MAX.to_le_bytes());
            pos += osv;
        }

        // EADB at aedb_offset (min_dblk_nelmts elements)
        file_data[aedb_offset..aedb_offset + 4].copy_from_slice(b"EADB");
        file_data[aedb_offset + 4] = 0;
        file_data[aedb_offset + 5] = 0;
        file_data[aedb_offset + 6..aedb_offset + 14]
            .copy_from_slice(&(aehd_offset as u64).to_le_bytes());
        // block_offset: ceil(max_nelmts_bits/8) = ceil(10/8) = 2 bytes
        // block_offset = 0 for first data block
        let blk_off_size = (10usize).div_ceil(8); // max_nelmts_bits=10
        let mut dbpos = aedb_offset + 6 + osv + blk_off_size;
        for i in 0..min_dblk_nelmts as usize {
            let addr = base_addr + (idx_blk_elmts as u64 + i as u64) * chunk_byte_size;
            file_data[dbpos..dbpos + osv].copy_from_slice(&addr.to_le_bytes());
            dbpos += osv;
        }

        let header = ExtensibleArrayHeader::parse(&file_data, aehd_offset, os, ls).unwrap();
        let ds_dims = vec![40u64];
        let chunk_dims = vec![10u32];
        let chunks =
            read_extensible_array_chunks(&file_data, &header, &ds_dims, &chunk_dims, 8, os, ls)
                .unwrap();

        assert_eq!(chunks.len(), 4);
        for (i, c) in chunks.iter().enumerate() {
            assert_eq!(c.address, base_addr + i as u64 * chunk_byte_size);
            assert_eq!(c.offsets, vec![i as u64 * 10]);
        }

        #[cfg(feature = "std")]
        assert_ea_streams_match(&file_data, aehd_offset, &ds_dims, &chunk_dims, 8, os, ls);
    }

    /// A large Extensible Array built by the writer reaches direct data blocks,
    /// super blocks, and paged super-block data blocks; the streaming reader
    /// must reproduce the buffered read across all those layers. Driving the
    /// layout from the writer avoids hand-building the doubling-table geometry.
    #[cfg(feature = "std")]
    #[test]
    fn streaming_ea_super_blocks_and_paged_match_buffered() {
        use crate::chunked_write::{WrittenChunk, build_extensible_array_at};
        use crate::source::{BytesSource, ReadSeekSource};

        // n covers: inline+direct (2000), several super blocks (50000), and
        // paged super-block data blocks (140000, since dblk_nelmts exceeds the
        // 1024-element page size at the higher super blocks).
        for &n in &[2000u64, 50000, 140000] {
            let chunks: Vec<WrittenChunk> = (0..n)
                .map(|i| WrittenChunk {
                    address: 0x10 + i * 8,
                    compressed_size: 8,
                    raw_size: 8,
                    filter_mask: 0,
                })
                .collect();
            let base = 0x1000u64;
            let ea = build_extensible_array_at(&chunks, 8, 8, false, base).unwrap();
            let mut file = vec![0u8; base as usize + ea.len()];
            file[base as usize..].copy_from_slice(&ea);

            let ds_dims = vec![n];
            let chunk_dims = vec![1u32];
            let header = ExtensibleArrayHeader::parse(&file, base as usize, 8, 8).unwrap();
            let buffered =
                read_extensible_array_chunks(&file, &header, &ds_dims, &chunk_dims, 8, 8, 8)
                    .unwrap();
            assert_eq!(buffered.len() as u64, n, "buffered chunk count at n={n}");

            let mem = BytesSource::new(&file);
            let hm = ExtensibleArrayHeader::parse_from_source(&mem, base, 8, 8).unwrap();
            let from_mem =
                read_extensible_array_chunks_from_source(&mem, &hm, &ds_dims, &chunk_dims, 8, 8, 8)
                    .unwrap();

            let seek = ReadSeekSource::new(std::io::Cursor::new(file)).unwrap();
            let hs = ExtensibleArrayHeader::parse_from_source(&seek, base, 8, 8).unwrap();
            let from_seek = read_extensible_array_chunks_from_source(
                &seek,
                &hs,
                &ds_dims,
                &chunk_dims,
                8,
                8,
                8,
            )
            .unwrap();

            assert_eq!(buffered, from_mem, "BytesSource mismatch at n={n}");
            assert_eq!(buffered, from_seek, "ReadSeekSource mismatch at n={n}");
        }
    }

    /// Test serialized_size computation.
    #[test]
    fn header_serialized_size() {
        // 12 fixed + 6*8 stats + 8 addr + 4 checksum = 72
        assert_eq!(ExtensibleArrayHeader::serialized_size(8, 8), 72);
        // 12 fixed + 6*4 stats + 4 addr + 4 checksum = 44
        assert_eq!(ExtensibleArrayHeader::serialized_size(4, 4), 44);
    }

    /// Verify read_element for unallocated slots.
    #[test]
    fn read_element_unallocated() {
        let data = vec![0xFFu8; 16];
        let num_chunks = vec![5u64];
        let chunk_dims = vec![10u32];
        let (info, consumed) =
            read_element(&data, 0, 0, 8, 8, 80, 0, &num_chunks, &chunk_dims).unwrap();
        assert!(info.is_none());
        assert_eq!(consumed, 8);
    }

    /// Verify filtered element reading.
    #[test]
    fn read_element_filtered() {
        let os: u8 = 8;
        let chunk_size_bytes = 4usize;
        let elem_size = os as usize + chunk_size_bytes + 4;
        let mut data = vec![0u8; elem_size + 16];
        // Address
        data[0..8].copy_from_slice(&0x2000u64.to_le_bytes());
        // Compressed size (4 bytes LE)
        data[8..12].copy_from_slice(&120u32.to_le_bytes());
        // Filter mask
        data[12..16].copy_from_slice(&0u32.to_le_bytes());

        let num_chunks = vec![5u64];
        let chunk_dims = vec![10u32];
        let (info, consumed) = read_element(
            &data,
            0,
            1,
            elem_size as u8,
            os,
            80,
            2,
            &num_chunks,
            &chunk_dims,
        )
        .unwrap();
        let ci = info.unwrap();
        assert_eq!(ci.address, 0x2000);
        assert_eq!(ci.chunk_size, 120);
        assert_eq!(ci.filter_mask, 0);
        assert_eq!(ci.offsets, vec![20]);
        assert_eq!(consumed, elem_size);
    }
}