use crate::error::{Error, Result};
use crate::filters;
use crate::io::reader::HdfReader;
use std::io::{Read, Seek};
use std::thread;

use super::chunk_read::{BorrowedChunkPayloadRead, ChunkReadContext};
use super::{read_le_u32_at, read_le_uint_at, usize_from_u64, Dataset, DatasetInfo};

struct DecodedBTreeV2Chunk {
    coords: Vec<u64>,
    addr: u64,
    nbytes: u64,
    read_size: usize,
    filter_mask: u32,
}

const MIN_PARALLEL_BTREE_V2_DEFLATE_CHUNKS_1D: usize = 8;
const MIN_PARALLEL_BTREE_V2_DEFLATE_BYTES_1D: usize = 64 * 1024;

impl Dataset {
    pub(super) fn read_chunked_btree_v2<R: Read + Seek>(
        reader: &mut HdfReader<R>,
        info: &DatasetInfo,
        chunk_ctx: &ChunkReadContext<'_>,
    ) -> Result<Vec<u8>> {
        let mut output = Self::scratch_output(chunk_ctx.total_bytes);
        Self::read_chunked_btree_v2_into(reader, info, chunk_ctx, &mut output)?;
        Ok(output)
    }

    pub(super) fn read_chunked_btree_v2_into<R: Read + Seek>(
        reader: &mut HdfReader<R>,
        info: &DatasetInfo,
        chunk_ctx: &ChunkReadContext<'_>,
        output: &mut [u8],
    ) -> Result<()> {
        if output.len() != chunk_ctx.total_bytes {
            return Err(Error::InvalidFormat(format!(
                "v2-B-tree chunk output buffer has {} bytes, expected {}",
                output.len(),
                chunk_ctx.total_bytes
            )));
        }
        let filtered = info
            .filter_pipeline
            .as_ref()
            .map(|pipeline| !pipeline.filters.is_empty())
            .unwrap_or(false);
        let chunk_size_len = if filtered {
            Self::filtered_chunk_size_len(
                info,
                chunk_ctx.chunk_bytes,
                usize::from(reader.sizeof_size()),
            )?
        } else {
            0
        };
        let mut records = Vec::new();
        crate::format::btree_v2::collect_all_records_into(
            reader,
            chunk_ctx.idx_addr,
            &mut records,
        )?;
        Self::filled_data_into(
            chunk_ctx.total_bytes / chunk_ctx.element_size,
            chunk_ctx.element_size,
            info,
            output,
        )?;
        let mut compressed_scratch = Vec::new();
        let mut shuffle_scratch = Vec::new();
        let mut raw_scratch = Vec::new();

        if !(Self::btree_v2_uses_unfiltered_coalescing(info)
            && chunk_ctx.data_dims.len() == 1
            && chunk_ctx.chunk_dims.len() == 1)
        {
            let mut coords = Vec::with_capacity(chunk_ctx.data_dims.len());
            let mut filtered_scratch = Vec::new();
            let mut chunks = Vec::with_capacity(records.len());
            for record in &records {
                let (addr, nbytes, filter_mask) = Self::decode_btree_v2_chunk_record_into(
                    record,
                    filtered,
                    chunk_size_len,
                    usize::from(reader.sizeof_addr()),
                    chunk_ctx.data_dims.len(),
                    chunk_ctx.chunk_bytes,
                    &mut coords,
                )?;
                Self::trace_btree2_chunk_lookup(
                    chunk_ctx.idx_addr,
                    &coords,
                    addr,
                    nbytes,
                    filter_mask,
                );
                if crate::io::reader::is_undef_addr(addr) {
                    continue;
                }
                Self::scale_btree_v2_chunk_coords(&mut coords, chunk_ctx.chunk_dims)?;
                let read_size = usize_from_u64(nbytes, "v2-B-tree chunk size")?;
                chunks.push(DecodedBTreeV2Chunk {
                    coords: coords.clone(),
                    addr,
                    nbytes,
                    read_size,
                    filter_mask,
                });
            }

            let mut handled = vec![false; chunks.len()];
            Self::try_read_parallel_deflate_btree_v2_chunks_1d(
                reader,
                info,
                chunk_ctx,
                &chunks,
                output,
                &mut handled,
            )?;

            for (chunk_index, chunk) in chunks.iter().enumerate() {
                if handled.get(chunk_index).copied().unwrap_or(false) {
                    continue;
                }
                if Self::try_read_full_chunk_1d_into_output(
                    reader,
                    info,
                    chunk_ctx,
                    &chunk.coords,
                    chunk.addr,
                    chunk.read_size,
                    chunk.filter_mask,
                    output,
                    &mut compressed_scratch,
                    &mut shuffle_scratch,
                )? {
                    continue;
                }
                Self::read_btree_v2_chunk_payload_into_scratch(
                    reader,
                    chunk.addr,
                    chunk.nbytes,
                    chunk.read_size,
                    &mut raw_scratch,
                )?;

                if let Some(ref pipeline) = info.filter_pipeline {
                    if !pipeline.filters.is_empty() {
                        filters::apply_pipeline_reverse_with_mask_expected_into(
                            &raw_scratch,
                            pipeline,
                            chunk_ctx.element_size,
                            chunk.filter_mask,
                            chunk_ctx.chunk_bytes,
                            &mut filtered_scratch,
                        )?;
                        Self::copy_chunk_to_output(
                            &filtered_scratch,
                            &chunk.coords,
                            chunk_ctx.data_dims,
                            chunk_ctx.chunk_dims,
                            chunk_ctx.element_size,
                            output,
                        )?;
                        continue;
                    }
                }

                Self::copy_chunk_to_output(
                    &raw_scratch,
                    &chunk.coords,
                    chunk_ctx.data_dims,
                    chunk_ctx.chunk_dims,
                    chunk_ctx.element_size,
                    output,
                )?;
            }
            return Ok(());
        }

        let mut chunks = Vec::with_capacity(records.len());
        for record in &records {
            let (addr, nbytes, filter_mask, mut coords) = Self::decode_btree_v2_chunk_record(
                record,
                filtered,
                chunk_size_len,
                usize::from(reader.sizeof_addr()),
                chunk_ctx.data_dims.len(),
                chunk_ctx.chunk_bytes,
            )?;
            Self::trace_btree2_chunk_lookup(chunk_ctx.idx_addr, &coords, addr, nbytes, filter_mask);
            if crate::io::reader::is_undef_addr(addr) {
                continue;
            }

            Self::scale_btree_v2_chunk_coords(&mut coords, chunk_ctx.chunk_dims)?;
            let read_size = usize_from_u64(nbytes, "v2-B-tree chunk size")?;
            chunks.push(DecodedBTreeV2Chunk {
                coords,
                addr,
                nbytes,
                read_size,
                filter_mask,
            });
        }

        let handled = {
            Self::try_read_coalesced_borrowed_unfiltered_chunks_1d(
                reader,
                info,
                chunk_ctx,
                chunks.len(),
                chunks.iter().map(|chunk| {
                    Ok(BorrowedChunkPayloadRead {
                        coords: &chunk.coords,
                        addr: chunk.addr,
                        read_size: chunk.read_size,
                        filter_mask: chunk.filter_mask,
                    })
                }),
                output,
            )?
        };
        for (chunk_index, chunk) in chunks.iter().enumerate() {
            if handled.get(chunk_index).copied().unwrap_or(false) {
                continue;
            }
            if Self::try_read_full_chunk_1d_into_output(
                reader,
                info,
                chunk_ctx,
                &chunk.coords,
                chunk.addr,
                chunk.read_size,
                chunk.filter_mask,
                output,
                &mut compressed_scratch,
                &mut shuffle_scratch,
            )? {
                continue;
            }
            Self::read_btree_v2_chunk_payload_into_scratch(
                reader,
                chunk.addr,
                chunk.nbytes,
                chunk.read_size,
                &mut raw_scratch,
            )?;

            if let Some(ref pipeline) = info.filter_pipeline {
                if !pipeline.filters.is_empty() {
                    raw_scratch = filters::apply_pipeline_reverse_with_mask_expected(
                        &raw_scratch,
                        pipeline,
                        chunk_ctx.element_size,
                        chunk.filter_mask,
                        chunk_ctx.chunk_bytes,
                    )?;
                }
            }

            Self::copy_chunk_to_output(
                &raw_scratch,
                &chunk.coords,
                chunk_ctx.data_dims,
                chunk_ctx.chunk_dims,
                chunk_ctx.element_size,
                output,
            )?;
        }

        Ok(())
    }

    fn try_read_parallel_deflate_btree_v2_chunks_1d<R: Read + Seek>(
        reader: &mut HdfReader<R>,
        info: &DatasetInfo,
        chunk_ctx: &ChunkReadContext<'_>,
        chunks: &[DecodedBTreeV2Chunk],
        output: &mut [u8],
        handled: &mut [bool],
    ) -> Result<()> {
        // libhdf5 reads each filtered chunk through the raw data chunk cache
        // and runs H5Z_pipeline serially in-process. This Rust-only fast path
        // keeps that serial behavior as the semantic baseline but decompresses
        // independent deflate-only 1D full-prefix chunks on scoped workers.
        let Some(pipeline) = info.filter_pipeline.as_ref() else {
            return Ok(());
        };
        if !Self::is_deflate_only_pipeline(pipeline) || chunk_ctx.data_dims.len() != 1 {
            return Ok(());
        }
        if chunk_ctx.chunk_dims.len() != 1 || output.len() != chunk_ctx.total_bytes {
            return Ok(());
        }

        let data_size = usize_from_u64(chunk_ctx.data_dims[0], "dataset dimension")?;
        let chunk_size = usize_from_u64(chunk_ctx.chunk_dims[0], "chunk dimension")?;
        if chunk_size == 0 {
            return Ok(());
        }
        let full_chunk_count = data_size / chunk_size;
        if full_chunk_count < MIN_PARALLEL_BTREE_V2_DEFLATE_CHUNKS_1D
            || chunks.len() < full_chunk_count
        {
            return Ok(());
        }
        let chunk_bytes = chunk_size
            .checked_mul(chunk_ctx.element_size)
            .ok_or_else(|| Error::InvalidFormat("chunk byte size overflow".into()))?;
        if chunk_bytes == 0 || chunk_bytes != chunk_ctx.chunk_bytes {
            return Ok(());
        }
        let full_output_len = full_chunk_count
            .checked_mul(chunk_bytes)
            .ok_or_else(|| Error::InvalidFormat("parallel chunk output length overflow".into()))?;
        if full_output_len > output.len()
            || full_output_len < MIN_PARALLEL_BTREE_V2_DEFLATE_BYTES_1D
        {
            return Ok(());
        }

        let worker_count = super::support::parallel_deflate_worker_count(full_chunk_count);
        if worker_count <= 1 {
            return Ok(());
        }

        let mut payloads = Vec::with_capacity(full_chunk_count);
        for (chunk_index, chunk) in chunks.iter().take(full_chunk_count).enumerate() {
            let expected_coord = u64::try_from(chunk_index)
                .ok()
                .and_then(|index| index.checked_mul(chunk_ctx.chunk_dims[0]))
                .ok_or_else(|| {
                    Error::InvalidFormat("v2-B-tree chunk coordinate overflow".into())
                })?;
            if handled.get(chunk_index).copied().unwrap_or(false)
                || chunk.filter_mask != 0
                || chunk.coords.len() != 1
                || chunk.coords[0] != expected_coord
                || crate::io::reader::is_undef_addr(chunk.addr)
            {
                return Ok(());
            }
            let mut payload = vec![0u8; chunk.read_size];
            reader.seek(chunk.addr)?;
            reader.read_bytes_into(&mut payload).map_err(|err| {
                Error::InvalidFormat(format!(
                    "failed to read v2-B-tree chunk {chunk_index} at address {} with size {}: {err}",
                    chunk.addr, chunk.nbytes
                ))
            })?;
            payloads.push(payload);
        }

        let chunks_per_worker = full_chunk_count.div_ceil(worker_count);
        let full_output = &mut output[..full_output_len];
        let parallel_result: Result<()> = thread::scope(|scope| {
            let mut handles = Vec::new();
            let mut remaining_output = full_output;
            let mut payload_start = 0usize;
            while payload_start < full_chunk_count {
                let payload_end = (payload_start + chunks_per_worker).min(full_chunk_count);
                let group_chunks = payload_end - payload_start;
                let group_bytes = group_chunks * chunk_bytes;
                let (group_output, next_output) = remaining_output.split_at_mut(group_bytes);
                let group_payloads = &payloads[payload_start..payload_end];
                handles.push(scope.spawn(move || -> Result<()> {
                    for (chunk_offset, payload) in group_payloads.iter().enumerate() {
                        let start = chunk_offset * chunk_bytes;
                        let end = start + chunk_bytes;
                        crate::filters::deflate::decompress_exact_into(
                            payload,
                            &mut group_output[start..end],
                        )?;
                    }
                    Ok(())
                }));
                remaining_output = next_output;
                payload_start = payload_end;
            }
            for handle in handles {
                handle.join().map_err(|_| {
                    Error::InvalidFormat("parallel deflate worker panicked".into())
                })??;
            }
            Ok(())
        });
        parallel_result?;
        super::support::record_parallel_deflate_chunks_handled(full_chunk_count);

        for flag in handled.iter_mut().take(full_chunk_count) {
            *flag = true;
        }
        Ok(())
    }

    fn scale_btree_v2_chunk_coords(coords: &mut [u64], chunk_dims: &[u64]) -> Result<()> {
        for (coord, &chunk) in coords.iter_mut().zip(chunk_dims) {
            *coord = (*coord).checked_mul(chunk).ok_or_else(|| {
                Error::InvalidFormat("v2-B-tree chunk coordinate overflow".into())
            })?;
        }
        Ok(())
    }

    fn read_btree_v2_chunk_payload_into_scratch<R: Read + Seek>(
        reader: &mut HdfReader<R>,
        addr: u64,
        nbytes: u64,
        read_size: usize,
        scratch: &mut Vec<u8>,
    ) -> Result<()> {
        reader.seek(addr).map_err(|err| {
            Error::InvalidFormat(format!(
                "failed to seek to v2-B-tree chunk address {addr}: {err}",
            ))
        })?;
        scratch.resize(read_size, 0);
        reader.read_bytes_into(scratch).map_err(|err| {
            Error::InvalidFormat(format!(
                "failed to read v2-B-tree chunk at address {addr} with size {nbytes}: {err}"
            ))
        })
    }

    fn btree_v2_uses_unfiltered_coalescing(info: &DatasetInfo) -> bool {
        info.filter_pipeline
            .as_ref()
            .map(|pipeline| pipeline.filters.is_empty())
            .unwrap_or(true)
    }

    #[cfg(feature = "tracehash")]
    fn trace_btree2_chunk_lookup(
        index_addr: u64,
        scaled: &[u64],
        addr: u64,
        nbytes: u64,
        filter_mask: u32,
    ) {
        let mut th = tracehash::th_call!("hdf5.chunk_index.btree2.lookup");
        th.input_u64(index_addr);
        for coord in scaled {
            th.input_u64(*coord);
        }
        th.output_value(&(true));
        th.output_u64(addr);
        th.output_u64(if crate::io::reader::is_undef_addr(addr) {
            0
        } else {
            nbytes
        });
        th.output_u64(u64::from(filter_mask));
        th.finish();
    }

    #[cfg(not(feature = "tracehash"))]
    fn trace_btree2_chunk_lookup(
        _index_addr: u64,
        _scaled: &[u64],
        _addr: u64,
        _nbytes: u64,
        _filter_mask: u32,
    ) {
    }

    #[cfg(feature = "tracehash")]
    fn trace_btree2_record_decode(
        record: &[u8],
        addr: u64,
        nbytes: u64,
        filter_mask: u32,
        scaled: &[u64],
    ) {
        let mut th = tracehash::th_call!("hdf5.chunk_index.btree2.record_decode");
        th.input_bytes(record);
        th.output_value(&(true));
        th.output_u64(addr);
        th.output_u64(nbytes);
        let Ok(scaled_len) = u64::try_from(scaled.len()) else {
            return;
        };
        th.output_u64(u64::from(filter_mask));
        th.output_u64(scaled_len);
        for coord in scaled {
            th.output_u64(*coord);
        }
        th.finish();
    }

    #[cfg(not(feature = "tracehash"))]
    fn trace_btree2_record_decode(
        _record: &[u8],
        _addr: u64,
        _nbytes: u64,
        _filter_mask: u32,
        _scaled: &[u64],
    ) {
    }

    pub(super) fn decode_btree_v2_chunk_record(
        record: &[u8],
        filtered: bool,
        chunk_size_len: usize,
        sizeof_addr: usize,
        ndims: usize,
        chunk_bytes: usize,
    ) -> Result<(u64, u64, u32, Vec<u64>)> {
        let mut scaled = Vec::with_capacity(ndims);
        let (addr, nbytes, filter_mask) = Self::decode_btree_v2_chunk_record_into(
            record,
            filtered,
            chunk_size_len,
            sizeof_addr,
            ndims,
            chunk_bytes,
            &mut scaled,
        )?;
        Ok((addr, nbytes, filter_mask, scaled))
    }

    pub(super) fn decode_btree_v2_chunk_record_into(
        record: &[u8],
        filtered: bool,
        chunk_size_len: usize,
        sizeof_addr: usize,
        ndims: usize,
        chunk_bytes: usize,
        scaled: &mut Vec<u64>,
    ) -> Result<(u64, u64, u32)> {
        let mut pos = 0usize;
        let addr = read_le_uint_at(record, &mut pos, sizeof_addr)?;

        let (nbytes, filter_mask) = if filtered {
            let nbytes = read_le_uint_at(record, &mut pos, chunk_size_len)?;
            let filter_mask = read_le_u32_at(record, &mut pos).map_err(|err| {
                if err.to_string().contains("truncated u32 field") {
                    Error::InvalidFormat("truncated v2-B-tree filter mask".into())
                } else {
                    err
                }
            })?;
            (nbytes, filter_mask)
        } else {
            (
                u64::try_from(chunk_bytes)
                    .map_err(|_| Error::InvalidFormat("v2-B-tree chunk size overflow".into()))?,
                0,
            )
        };

        scaled.clear();
        scaled.reserve(ndims);
        for _ in 0..ndims {
            scaled.push(read_le_uint_at(record, &mut pos, 8)?);
        }

        Self::trace_btree2_record_decode(record, addr, nbytes, filter_mask, scaled);

        Ok((addr, nbytes, filter_mask))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::format::messages::data_layout::{ChunkIndexType, DataLayoutMessage, LayoutClass};
    use crate::format::messages::dataspace::{DataspaceMessage, DataspaceType};
    use crate::format::messages::datatype::{DatatypeClass, DatatypeMessage};
    use crate::format::messages::filter_pipeline::{
        FilterDesc, FilterPipelineMessage, FILTER_DEFLATE,
    };
    use std::io::Cursor;

    #[test]
    fn decode_btree_v2_record_rejects_truncated_filter_mask() {
        let record = [1u8; 8 + 1 + 3];
        let err = Dataset::decode_btree_v2_chunk_record(&record, true, 1, 8, 0, 16).unwrap_err();
        assert!(
            err.to_string().contains("truncated v2-B-tree filter mask"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn decode_btree_v2_record_rejects_truncated_scaled_coordinate() {
        let record = [1u8; 8 + 7];
        let err = Dataset::decode_btree_v2_chunk_record(&record, false, 0, 8, 1, 16).unwrap_err();
        assert!(
            err.to_string().contains("truncated integer field")
                || err
                    .to_string()
                    .contains("invalid little-endian integer size"),
            "unexpected error: {err}"
        );
    }

    fn deflate_info_and_context<'a>(
        data_dims: &'a [u64],
        chunk_dims: &'a [u64],
        chunk_bytes: usize,
        total_bytes: usize,
    ) -> (DatasetInfo, ChunkReadContext<'a>) {
        let info = DatasetInfo {
            dataspace: DataspaceMessage {
                version: 2,
                space_type: DataspaceType::Simple,
                ndims: u8::try_from(data_dims.len()).unwrap(),
                dims: data_dims.to_vec(),
                max_dims: None,
            },
            datatype: DatatypeMessage {
                version: 4,
                class: DatatypeClass::FixedPoint,
                class_bits: [0, 0, 0],
                size: 4,
                properties: vec![0; 4],
            },
            layout: DataLayoutMessage {
                version: 4,
                layout_class: LayoutClass::Chunked,
                compact_data: None,
                contiguous_addr: None,
                contiguous_size: None,
                chunk_dims: Some(chunk_dims.to_vec()),
                chunk_index_addr: Some(0),
                chunk_index_type: Some(ChunkIndexType::BTreeV2),
                chunk_element_size: None,
                chunk_flags: None,
                chunk_encoded_dims: Some(chunk_dims.to_vec()),
                single_chunk_filtered_size: None,
                single_chunk_filter_mask: None,
                data_addr: None,
                virtual_heap_addr: None,
                virtual_heap_index: None,
            },
            filter_pipeline: Some(FilterPipelineMessage {
                version: 2,
                filters: vec![FilterDesc {
                    id: FILTER_DEFLATE,
                    name: None,
                    flags: 0,
                    client_data: vec![4],
                }],
            }),
            fill_value: None,
            external_file_list: None,
        };
        let chunk_ctx = ChunkReadContext {
            idx_addr: 0,
            data_dims,
            chunk_dims,
            chunk_bytes,
            element_size: 4,
            total_bytes,
        };
        (info, chunk_ctx)
    }

    fn synthetic_btree_v2_deflate_payloads(
        chunk_elems: usize,
        full_chunk_count: usize,
    ) -> (Vec<u8>, Vec<DecodedBTreeV2Chunk>, Vec<u8>) {
        let mut image = Vec::new();
        let mut chunks = Vec::new();
        let mut expected = Vec::new();

        for chunk_index in 0..full_chunk_count {
            let start = chunk_index * chunk_elems;
            let mut raw = Vec::with_capacity(chunk_elems * 4);
            for value in start..start + chunk_elems {
                let value = value as i32 * 13 - 23;
                raw.extend_from_slice(&value.to_le_bytes());
            }
            expected.extend_from_slice(&raw);

            let addr = image.len() as u64;
            let mut compressed = Vec::new();
            crate::filters::deflate::compress_into(&raw, 4, &mut compressed).unwrap();
            let read_size = compressed.len();
            image.extend_from_slice(&compressed);

            chunks.push(DecodedBTreeV2Chunk {
                coords: vec![(start as u64)],
                addr,
                nbytes: read_size as u64,
                read_size,
                filter_mask: 0,
            });
        }

        (image, chunks, expected)
    }

    #[test]
    fn parallel_deflate_btree_v2_helper_decodes_full_prefix() {
        const CHUNK_ELEMS: usize = 2048;
        const FULL_CHUNKS: usize = 8;
        const TAIL_ELEMS: usize = 17;
        let data_dims = [(CHUNK_ELEMS * FULL_CHUNKS + TAIL_ELEMS) as u64];
        let chunk_dims = [CHUNK_ELEMS as u64];
        let chunk_bytes = CHUNK_ELEMS * 4;
        let total_bytes = (CHUNK_ELEMS * FULL_CHUNKS + TAIL_ELEMS) * 4;
        let (info, chunk_ctx) =
            deflate_info_and_context(&data_dims, &chunk_dims, chunk_bytes, total_bytes);
        let (image, chunks, expected_prefix) =
            synthetic_btree_v2_deflate_payloads(CHUNK_ELEMS, FULL_CHUNKS);
        let mut reader = HdfReader::new(Cursor::new(image));
        let mut output = vec![0u8; total_bytes];
        let mut handled = vec![false; chunks.len()];

        super::super::support::set_parallel_deflate_worker_override(2);
        super::super::support::reset_parallel_deflate_chunks_handled();
        Dataset::try_read_parallel_deflate_btree_v2_chunks_1d(
            &mut reader,
            &info,
            &chunk_ctx,
            &chunks,
            &mut output,
            &mut handled,
        )
        .unwrap();

        assert_eq!(
            super::super::support::parallel_deflate_chunks_handled(),
            FULL_CHUNKS
        );
        assert!(handled.iter().all(|handled| *handled));
        assert_eq!(&output[..expected_prefix.len()], expected_prefix);
        assert!(output[expected_prefix.len()..]
            .iter()
            .all(|byte| *byte == 0));
        super::super::support::set_parallel_deflate_worker_override(0);
    }

    #[test]
    fn parallel_deflate_btree_v2_helper_skips_masked_chunk() {
        const CHUNK_ELEMS: usize = 2048;
        const FULL_CHUNKS: usize = 8;
        let data_dims = [(CHUNK_ELEMS * FULL_CHUNKS) as u64];
        let chunk_dims = [CHUNK_ELEMS as u64];
        let chunk_bytes = CHUNK_ELEMS * 4;
        let total_bytes = CHUNK_ELEMS * FULL_CHUNKS * 4;
        let (info, chunk_ctx) =
            deflate_info_and_context(&data_dims, &chunk_dims, chunk_bytes, total_bytes);
        let (image, mut chunks, _) = synthetic_btree_v2_deflate_payloads(CHUNK_ELEMS, FULL_CHUNKS);
        chunks[3].filter_mask = 1;
        let mut reader = HdfReader::new(Cursor::new(image));
        let mut output = vec![0u8; total_bytes];
        let mut handled = vec![false; chunks.len()];

        super::super::support::set_parallel_deflate_worker_override(2);
        super::super::support::reset_parallel_deflate_chunks_handled();
        Dataset::try_read_parallel_deflate_btree_v2_chunks_1d(
            &mut reader,
            &info,
            &chunk_ctx,
            &chunks,
            &mut output,
            &mut handled,
        )
        .unwrap();

        assert_eq!(super::super::support::parallel_deflate_chunks_handled(), 0);
        assert!(handled.iter().all(|handled| !*handled));
        assert!(output.iter().all(|byte| *byte == 0));
        super::super::support::set_parallel_deflate_worker_override(0);
    }
}
