diff --git a/rust/lance-encoding/src/encodings/logical/primitive.rs b/rust/lance-encoding/src/encodings/logical/primitive.rs index a1bb195117f..c13ef2f3034 100644 --- a/rust/lance-encoding/src/encodings/logical/primitive.rs +++ b/rust/lance-encoding/src/encodings/logical/primitive.rs @@ -89,6 +89,7 @@ use crate::{ }; pub mod blob; +mod chunk_index; pub mod constant; pub mod dict; pub mod fullzip; @@ -96,6 +97,8 @@ mod layout; pub mod miniblock; pub(crate) mod sparse; +use chunk_index::{ItemCounts, MiniBlockChunkIndex, PrefixSums, RowMapping, parse_nested_rep}; + const FILL_BYTE: u8 = 0xFE; const DEFAULT_DICT_DIVISOR: u64 = 2; const DEFAULT_DICT_MAX_CARDINALITY: u64 = 100_000; @@ -1209,121 +1212,18 @@ struct MiniBlockSchedulerDictionary { num_dictionary_items: u64, } -/// Individual block metadata within a MiniBlock repetition index. -#[derive(Debug)] -struct MiniBlockRepIndexBlock { - // The index of the first row that starts after the beginning of this block. If the block - // has a preamble this will be the row after the preamble. If the block is entirely preamble - // then this will be a row that starts in some future block. - first_row: u64, - // The number of rows in the block, including the trailer but not the preamble. - // Can be 0 if the block is entirely preamble - starts_including_trailer: u64, - // Whether the block has a preamble - has_preamble: bool, - // Whether the block has a trailer - has_trailer: bool, -} - -impl DeepSizeOf for MiniBlockRepIndexBlock { - fn deep_size_of_children(&self, _context: &mut Context) -> usize { - 0 - } -} - -/// Repetition index for MiniBlock encoding. -/// -/// Stores block-level offset information to enable efficient random -/// access to nested data structures within mini-blocks. -#[derive(Debug)] -struct MiniBlockRepIndex { - blocks: Vec, -} - -impl DeepSizeOf for MiniBlockRepIndex { - fn deep_size_of_children(&self, context: &mut Context) -> usize { - self.blocks.deep_size_of_children(context) - } -} - -impl MiniBlockRepIndex { - /// Decode repetition index from chunk metadata using default values. - /// - /// This creates a repetition index where each chunk has no partial values - /// and no trailers, suitable for simple sequential data layouts. - pub fn default_from_chunks(chunks: &[ChunkMeta]) -> Self { - let mut blocks = Vec::with_capacity(chunks.len()); - let mut offset: u64 = 0; - - for c in chunks { - blocks.push(MiniBlockRepIndexBlock { - first_row: offset, - starts_including_trailer: c.num_values, - has_preamble: false, - has_trailer: false, - }); - - offset += c.num_values; - } - - Self { blocks } - } - - /// Decode repetition index from raw bytes in little-endian format. - /// - /// The bytes should contain u64 values arranged in groups of `stride` elements, - /// where the first two values of each group represent ends_count and partial_count. - /// Returns an empty index if no bytes are provided. - pub fn decode_from_bytes(rep_bytes: &[u8], stride: usize) -> Self { - // Convert bytes to u64 slice, handling alignment automatically - let buffer = crate::buffer::LanceBuffer::from(rep_bytes.to_vec()); - let u64_slice = buffer.borrow_to_typed_slice::(); - let n = u64_slice.len() / stride; - - let mut blocks = Vec::with_capacity(n); - let mut chunk_has_preamble = false; - let mut offset: u64 = 0; - - // Extract first two values from each block: ends_count and partial_count - for i in 0..n { - let base_idx = i * stride; - let ends = u64_slice[base_idx]; - let partial = u64_slice[base_idx + 1]; - - let has_trailer = partial > 0; - // Convert branches to arithmetic for better compiler optimization - let starts_including_trailer = - ends + (has_trailer as u64) - (chunk_has_preamble as u64); - - blocks.push(MiniBlockRepIndexBlock { - first_row: offset, - starts_including_trailer, - has_preamble: chunk_has_preamble, - has_trailer, - }); - - chunk_has_preamble = has_trailer; - offset += starts_including_trailer; - } - - Self { blocks } - } -} - /// State that is loaded once and cached for future lookups #[derive(Debug)] struct MiniBlockCacheableState { - /// Metadata that describes each chunk in the page - chunk_meta: Vec, - /// The decoded repetition index - rep_index: MiniBlockRepIndex, + /// Compact per-chunk index (byte ranges + row/item mapping) for the page + chunk_index: MiniBlockChunkIndex, /// The dictionary for the page, if any dictionary: Option>, } impl DeepSizeOf for MiniBlockCacheableState { fn deep_size_of_children(&self, context: &mut Context) -> usize { - self.rep_index.deep_size_of_children(context) + self.chunk_index.deep_size_of_children(context) + self .dictionary .as_ref() @@ -1468,19 +1368,14 @@ impl MiniBlockScheduler { } fn lookup_chunks(&self, chunk_indices: &[usize]) -> Vec { - let page_meta = self.page_meta.as_ref().unwrap(); + let chunk_index = &self.page_meta.as_ref().unwrap().chunk_index; chunk_indices .iter() - .map(|&chunk_idx| { - let chunk_meta = &page_meta.chunk_meta[chunk_idx]; - let bytes_start = chunk_meta.offset_bytes; - let bytes_end = bytes_start + chunk_meta.chunk_size_bytes; - LoadedChunk { - byte_range: bytes_start..bytes_end, - items_in_chunk: chunk_meta.num_values, - chunk_idx, - data: LanceBuffer::empty(), - } + .map(|&chunk_idx| LoadedChunk { + byte_range: chunk_index.byte_range(chunk_idx), + items_in_chunk: chunk_index.items_in_chunk(chunk_idx), + chunk_idx, + data: LanceBuffer::empty(), }) .collect() } @@ -1585,9 +1480,12 @@ impl ChunkInstructions { // // The output will be a set of `ChunkInstructions` which tell us how to read from the chunks fn schedule_instructions( - rep_index: &MiniBlockRepIndex, + chunk_index: &MiniBlockChunkIndex, user_ranges: &[Range], ) -> Vec { + // Bind the per-page chunk count once; re-deriving it each iteration + // costs a width match plus a length read. + let num_chunks = chunk_index.num_chunks(); // This is an in-exact capacity guess but pretty good. The actual capacity can be // smaller if instructions are merged. It can be larger if there are multiple instructions // per row which can happen with lists. @@ -1599,43 +1497,30 @@ impl ChunkInstructions { // Need to find the first chunk with a first row >= user_range.start. If there are // multiple chunks with the same first row we need to take the first one. - let mut block_index = match rep_index - .blocks - .binary_search_by_key(&user_range.start, |block| block.first_row) - { - Ok(idx) => { - // Slightly tricky case, we may need to walk backwards a bit to make sure we - // are grabbing first eligible chunk - let mut idx = idx; - while idx > 0 && rep_index.blocks[idx - 1].first_row == user_range.start { - idx -= 1; - } - idx - } - // Easy case. idx is greater, and idx - 1 is smaller, so idx - 1 contains the start - Err(idx) => idx - 1, - }; + let mut block_index = chunk_index.find_chunk(user_range.start); - let mut to_skip = user_range.start - rep_index.blocks[block_index].first_row; + let mut to_skip = user_range.start - chunk_index.first_row(block_index); while rows_needed > 0 || need_preamble { // Check if we've gone past the last block (should not happen) - if block_index >= rep_index.blocks.len() { + if block_index >= num_chunks { log::warn!( - "schedule_instructions inconsistency: block_index >= rep_index.blocks.len(), exiting early" + "schedule_instructions inconsistency: block_index >= num_chunks, exiting early" ); break; } - let chunk = &rep_index.blocks[block_index]; - let rows_avail = chunk.starts_including_trailer.saturating_sub(to_skip); + let starts_including_trailer = chunk_index.rows_in_chunk(block_index); + let has_preamble = chunk_index.has_preamble(block_index); + let has_trailer = chunk_index.has_trailer(block_index); + let rows_avail = starts_including_trailer.saturating_sub(to_skip); // Handle blocks that are entirely preamble (rows_avail = 0) // These blocks have no rows to take but may have a preamble we need // We only look for preamble if to_skip == 0 (we're not skipping rows) if rows_avail == 0 && to_skip == 0 { // Only process if this chunk has a preamble we need - if chunk.has_preamble && need_preamble { + if has_preamble && need_preamble { chunk_instructions.push(Self { chunk_idx: block_index, preamble: PreambleAction::Take, @@ -1644,14 +1529,12 @@ impl ChunkInstructions { // We still need to look at has_trailer to distinguish between "all preamble // and row ends at end of chunk" and "all preamble and row bleeds into next // chunk". Both cases will have 0 rows available. - take_trailer: chunk.has_trailer, + take_trailer: has_trailer, }); // Only set need_preamble = false if the chunk has at least one row, // Or we are reaching the last block, // Otherwise, the chunk is entirely preamble and we need the next chunk's preamble too - if chunk.starts_including_trailer > 0 - || block_index == rep_index.blocks.len() - 1 - { + if starts_including_trailer > 0 || block_index == num_chunks - 1 { need_preamble = false; } } @@ -1666,7 +1549,7 @@ impl ChunkInstructions { if rows_avail == 0 && to_skip > 0 { // This block doesn't have enough rows to skip, move to next block // Adjust to_skip by the number of rows in this block - to_skip -= chunk.starts_including_trailer; + to_skip -= starts_including_trailer; block_index += 1; continue; } @@ -1675,7 +1558,7 @@ impl ChunkInstructions { rows_needed -= rows_to_take; let mut take_trailer = false; - let preamble = if chunk.has_preamble { + let preamble = if has_preamble { if need_preamble { PreambleAction::Take } else { @@ -1686,7 +1569,7 @@ impl ChunkInstructions { }; // Are we taking the trailer? If so, make sure we mark that we need the preamble - if rows_to_take == rows_avail && chunk.has_trailer { + if rows_to_take == rows_avail && has_trailer { take_trailer = true; need_preamble = true; } else { @@ -1710,25 +1593,33 @@ impl ChunkInstructions { // are _adjacent_ (i.e. don't merge "take first row of chunk 0" and "take third row of chunk 0" into "take 2 // rows of chunk 0 starting at 0") if user_ranges.len() > 1 { - // TODO: Could probably optimize this allocation away - let mut merged_instructions = Vec::with_capacity(chunk_instructions.len()); - let mut instructions_iter = chunk_instructions.into_iter(); - merged_instructions.push(instructions_iter.next().unwrap()); - for instruction in instructions_iter { - let last = merged_instructions.last_mut().unwrap(); - if last.chunk_idx == instruction.chunk_idx - && last.rows_to_take + last.rows_to_skip == instruction.rows_to_skip - { - last.rows_to_take += instruction.rows_to_take; - last.take_trailer |= instruction.take_trailer; + // Merge adjacent instructions in place. `write` indexes the last + // retained instruction; each following instruction is either folded + // into it (contiguous within the same chunk) or compacted forward. + let mut write = 0; + for read in 1..chunk_instructions.len() { + let merges = { + let last = &chunk_instructions[write]; + let candidate = &chunk_instructions[read]; + last.chunk_idx == candidate.chunk_idx + && last.rows_to_take + last.rows_to_skip == candidate.rows_to_skip + }; + if merges { + let rows_to_take = chunk_instructions[read].rows_to_take; + let take_trailer = chunk_instructions[read].take_trailer; + let last = &mut chunk_instructions[write]; + last.rows_to_take += rows_to_take; + last.take_trailer |= take_trailer; } else { - merged_instructions.push(instruction); + write += 1; + if write != read { + chunk_instructions.swap(write, read); + } } } - merged_instructions - } else { - chunk_instructions + chunk_instructions.truncate(write + 1); } + chunk_instructions } fn drain_from_instruction( @@ -1839,6 +1730,129 @@ impl<'a> Iterator for WordsIter<'a> { } } +/// Per-chunk leaf value-count analysis derived from the metadata words. +/// +/// `values_per_chunk` is the count shared by every non-last chunk (meaningful +/// when `uniform`), and `last_chunk_values` is the final chunk's count. +struct FlatValueCounts { + logs: Vec, + uniform: bool, + values_per_chunk: u64, + last_chunk_values: u64, +} + +fn analyze_value_counts(words: &Words, items_in_page: u64) -> FlatValueCounts { + let num_chunks = words.len(); + let logs = words.iter().map(|w| (w & 0x0F) as u8).collect::>(); + let mut counted = 0u64; + for &log in logs.iter().take(num_chunks.saturating_sub(1)) { + // Non-last chunks always encode a positive log value count. + debug_assert!(log > 0); + counted += 1u64 << log; + } + let last_chunk_values = items_in_page - counted; + if let Some(&last_log) = logs.last() { + debug_assert!(last_log == 0 || (1u64 << last_log) == last_chunk_values); + } + let uniform = num_chunks <= 1 || logs[..num_chunks - 1].iter().all(|&log| log == logs[0]); + // A single-chunk page has no "non-last" chunk to derive a stride from; use the + // page item count (min 1 so it stays a valid divisor in `find_chunk`). + let values_per_chunk = if num_chunks <= 1 { + items_in_page.max(1) + } else { + 1u64 << logs[0] + }; + FlatValueCounts { + logs, + uniform, + values_per_chunk, + last_chunk_values, + } +} + +/// Iterator over per-chunk value counts for a non-uniform flat page. Non-last +/// chunks yield `1 << log`; the last yields the remaining items in the page. +fn flat_value_counts_iter(logs: &[u8], items_in_page: u64) -> impl Iterator + '_ { + let num_chunks = logs.len(); + let mut counted = 0u64; + (0..num_chunks).map(move |i| { + let count = if i + 1 < num_chunks { + 1u64 << logs[i] + } else { + items_in_page - counted + }; + counted += count; + count + }) +} + +/// Builds the compact per-chunk index from the metadata words and, for nested +/// pages, the raw repetition-index bytes. The row axis is picked by page shape: +/// `UniformFlat` when all non-last chunks share a value count (fixed-width / +/// bitpacking), `Flat` for non-uniform flat pages (RLE / FSST), else `Nested`. +fn build_chunk_index( + words: &Words, + items_in_page: u64, + base: u64, + data_buf_size: u64, + rep_index_bytes: Option<&[u8]>, + repetition_index_depth: u16, +) -> MiniBlockChunkIndex { + let num_chunks = words.len(); + // Each chunk stores `(divided_bytes + 1) * MINIBLOCK_ALIGNMENT` bytes, so the + // deltas are the chunk sizes and their grand total is the data buffer size. + let byte_starts = PrefixSums::from_deltas( + words + .iter() + .map(|word| ((word >> 4) as u64 + 1) * MINIBLOCK_ALIGNMENT as u64), + num_chunks, + data_buf_size, + ); + + // Nested pages track rows via the repetition index and keep leaf item counts + // separately; flat pages have row == value index, so value counts are rows. + let rows = if let Some(rep_index_data) = rep_index_bytes { + assert!(rep_index_data.len() % 8 == 0); + let stride = repetition_index_depth as usize + 1; + let (row_starts, has_trailer) = parse_nested_rep(rep_index_data, stride); + let value_counts = analyze_value_counts(words, items_in_page); + let item_counts = if value_counts.uniform { + ItemCounts::Uniform { + values_per_chunk: value_counts.values_per_chunk, + last_chunk_values: value_counts.last_chunk_values, + } + } else { + ItemCounts::PerChunkLog { + logs: value_counts.logs, + last_chunk_values: value_counts.last_chunk_values, + } + }; + RowMapping::Nested { + row_starts, + has_trailer, + item_counts, + } + } else { + let value_counts = analyze_value_counts(words, items_in_page); + if value_counts.uniform { + RowMapping::UniformFlat { + values_per_chunk: value_counts.values_per_chunk, + last_chunk_values: value_counts.last_chunk_values, + num_chunks, + } + } else { + let value_starts = PrefixSums::from_deltas( + flat_value_counts_iter(&value_counts.logs, items_in_page), + num_chunks, + items_in_page, + ); + RowMapping::Flat { value_starts } + } + }; + + MiniBlockChunkIndex::new(base, byte_starts, rows) +} + impl StructuralPageScheduler for MiniBlockScheduler { fn initialize<'a>( &'a mut self, @@ -1848,7 +1862,8 @@ impl StructuralPageScheduler for MiniBlockScheduler { // we may also need to fetch the repetition index. Here, we gather what buffers we // need. let (meta_buf_position, meta_buf_size) = self.buffer_offsets_and_sizes[0]; - let value_buf_position = self.buffer_offsets_and_sizes[1].0; + let base = self.buffer_offsets_and_sizes[1].0; + let data_buf_size = self.buffer_offsets_and_sizes[1].1; let mut bufs_needed = 1; if self.dictionary.is_some() { bufs_needed += 1; @@ -1877,65 +1892,31 @@ impl StructuralPageScheduler for MiniBlockScheduler { let dictionary_bytes = self.dictionary.as_ref().and_then(|_| buffers.next()); let rep_index_bytes = buffers.next(); - // Parse the metadata and build the chunk meta let words = Words::from_bytes(meta_bytes, self.has_large_chunk)?; - let mut chunk_meta = Vec::with_capacity(words.len()); - - let mut rows_counter = 0; - let mut offset_bytes = value_buf_position; - for (word_idx, word) in words.iter().enumerate() { - let log_num_values = word & 0x0F; - let divided_bytes = word >> 4; - let num_bytes = (divided_bytes as usize + 1) * MINIBLOCK_ALIGNMENT; - debug_assert!(num_bytes > 0); - let num_values = if word_idx < words.len() - 1 { - debug_assert!(log_num_values > 0); - 1 << log_num_values - } else { - debug_assert!( - log_num_values == 0 - || (1 << log_num_values) == (self.items_in_page - rows_counter) - ); - self.items_in_page - rows_counter - }; - rows_counter += num_values; - - chunk_meta.push(ChunkMeta { - num_values, - chunk_size_bytes: num_bytes as u64, - offset_bytes, - }); - offset_bytes += num_bytes as u64; - } - - // Build the repetition index - let rep_index = if let Some(rep_index_data) = rep_index_bytes { - assert!(rep_index_data.len() % 8 == 0); - let stride = self.repetition_index_depth as usize + 1; - MiniBlockRepIndex::decode_from_bytes(&rep_index_data, stride) - } else { - MiniBlockRepIndex::default_from_chunks(&chunk_meta) - }; - - let mut page_meta = MiniBlockCacheableState { - chunk_meta, - rep_index, - dictionary: None, - }; + let chunk_index = build_chunk_index( + &words, + self.items_in_page, + base, + data_buf_size, + rep_index_bytes.as_deref(), + self.repetition_index_depth, + ); // decode dictionary - if let Some(ref mut dictionary) = self.dictionary { + let dictionary = if let Some(ref mut dictionary) = self.dictionary { let dictionary_data = dictionary_bytes.unwrap(); - page_meta.dictionary = - Some(Arc::new(dictionary.dictionary_decompressor.decompress( - LanceBuffer::from_bytes( - dictionary_data, - dictionary.dictionary_data_alignment, - ), - dictionary.num_dictionary_items, - )?)); + Some(Arc::new(dictionary.dictionary_decompressor.decompress( + LanceBuffer::from_bytes(dictionary_data, dictionary.dictionary_data_alignment), + dictionary.num_dictionary_items, + )?)) + } else { + None }; - let page_meta = Arc::new(page_meta); + + let page_meta = Arc::new(MiniBlockCacheableState { + chunk_index, + dictionary, + }); self.page_meta = Some(page_meta.clone()); Ok(page_meta as Arc) } @@ -1961,7 +1942,7 @@ impl StructuralPageScheduler for MiniBlockScheduler { let page_meta = self.page_meta.as_ref().unwrap(); let chunk_instructions = - ChunkInstructions::schedule_instructions(&page_meta.rep_index, ranges); + ChunkInstructions::schedule_instructions(&page_meta.chunk_index, ranges); debug_assert_eq!( num_rows, @@ -5940,8 +5921,8 @@ mod tests { use super::{ ChunkInstructions, DataBlock, DecodeMiniBlockTask, FixedPerValueDecompressor, FixedWidthDataBlock, FullZipCacheableState, FullZipDecodeDetails, FullZipReadSource, - FullZipRepIndexDetails, FullZipScheduler, MiniBlockChunk, MiniBlockCompressed, - MiniBlockRepIndex, PerValueDecompressor, PreambleAction, StructuralPageScheduler, + FullZipRepIndexDetails, FullZipScheduler, MiniBlockChunk, MiniBlockChunkIndex, + MiniBlockCompressed, PerValueDecompressor, PreambleAction, StructuralPageScheduler, VariableFullZipDecoder, }; use crate::buffer::LanceBuffer; @@ -6436,11 +6417,10 @@ mod tests { // Convert repetition index to bytes for testing let rep_data: Vec = vec![5, 2, 3, 0, 4, 7, 2, 0]; let rep_bytes: Vec = rep_data.iter().flat_map(|v| v.to_le_bytes()).collect(); - let repetition_index = MiniBlockRepIndex::decode_from_bytes(&rep_bytes, 2); + let chunk_index = MiniBlockChunkIndex::new_nested_for_test(&rep_bytes, 2); let check = |user_ranges, expected_instructions| { - let instructions = - ChunkInstructions::schedule_instructions(&repetition_index, user_ranges); + let instructions = ChunkInstructions::schedule_instructions(&chunk_index, user_ranges); assert_eq!(instructions, expected_instructions); }; @@ -6592,11 +6572,11 @@ mod tests { // Convert repetition index to bytes for testing let rep_data: Vec = vec![5, 2, 3, 0, 4, 7, 2, 0]; let rep_bytes: Vec = rep_data.iter().flat_map(|v| v.to_le_bytes()).collect(); - let repetition_index = MiniBlockRepIndex::decode_from_bytes(&rep_bytes, 2); + let chunk_index = MiniBlockChunkIndex::new_nested_for_test(&rep_bytes, 2); let user_ranges = vec![1..7, 10..14]; // First, schedule the ranges - let scheduled = ChunkInstructions::schedule_instructions(&repetition_index, &user_ranges); + let scheduled = ChunkInstructions::schedule_instructions(&chunk_index, &user_ranges); let mut to_drain = VecDeque::from(scheduled.clone()); @@ -6677,11 +6657,11 @@ mod tests { // Regression case. Need a chunk with preamble, rows, and trailer (the middle chunk here) let rep_data: Vec = vec![5, 2, 3, 3, 20, 0]; let rep_bytes: Vec = rep_data.iter().flat_map(|v| v.to_le_bytes()).collect(); - let repetition_index = MiniBlockRepIndex::decode_from_bytes(&rep_bytes, 2); + let chunk_index = MiniBlockChunkIndex::new_nested_for_test(&rep_bytes, 2); let user_ranges = vec![0..28]; // First, schedule the ranges - let scheduled = ChunkInstructions::schedule_instructions(&repetition_index, &user_ranges); + let scheduled = ChunkInstructions::schedule_instructions(&chunk_index, &user_ranges); let mut to_drain = VecDeque::from(scheduled.clone()); @@ -6741,6 +6721,181 @@ mod tests { assert_eq!(skip_in_chunk, 0); } + use super::chunk_index::{PrefixSums, RowMapping}; + use super::{MINIBLOCK_ALIGNMENT, Words, build_chunk_index}; + use bytes::Bytes; + use lance_core::cache::{Context, DeepSizeOf}; + use rstest::rstest; + + /// Builds a `Words` metadata buffer (u16 words) from `(log_num_values, num_bytes)` + /// pairs, returning the words and the total data-buffer size. + fn words_from(entries: &[(u32, u32)]) -> (Words, u64) { + let mut raw = Vec::with_capacity(entries.len() * 2); + let mut total = 0u64; + for &(log, num_bytes) in entries { + assert!(num_bytes > 0 && num_bytes % MINIBLOCK_ALIGNMENT as u32 == 0); + let divided = num_bytes / MINIBLOCK_ALIGNMENT as u32 - 1; + let word = (divided << 4) | log; + assert!(word <= u16::MAX as u32, "test word {word} exceeds u16"); + raw.extend_from_slice(&(word as u16).to_le_bytes()); + total += num_bytes as u64; + } + (Words::from_bytes(Bytes::from(raw), false).unwrap(), total) + } + + fn rep_bytes_from(values: &[u64]) -> Vec { + values.iter().flat_map(|v| v.to_le_bytes()).collect() + } + + #[rstest] + // Two full chunks of 8 values (log 3) plus a partial last chunk; byte sizes vary + // independently of value counts. + #[case::uniform_partial_last(&[(3, 16), (3, 24), (0, 8)], 19, "uniform_flat", 8, 3)] + // Single chunk covers the whole page. + #[case::single_chunk(&[(0, 24)], 5, "uniform_flat", 5, 5)] + // Last chunk is also full (exact multiple). + #[case::exact_multiple(&[(3, 16), (3, 16)], 16, "uniform_flat", 8, 8)] + // Non-last chunks differ in size, so this is a non-uniform flat page. + #[case::non_uniform(&[(4, 16), (2, 16), (0, 8)], 21, "flat", 16, 1)] + fn test_flat_detection( + #[case] entries: &[(u32, u32)], + #[case] items_in_page: u64, + #[case] expected_kind: &str, + #[case] expected_first_items: u64, + #[case] expected_last_items: u64, + ) { + let base = 100u64; + let (words, data_buf_size) = words_from(entries); + let index = build_chunk_index(&words, items_in_page, base, data_buf_size, None, 0); + + assert_eq!(index.row_mapping_debug(), expected_kind); + assert_eq!(index.num_chunks(), entries.len()); + assert_eq!(index.items_in_chunk(0), expected_first_items); + assert_eq!(index.items_in_chunk(entries.len() - 1), expected_last_items); + + // Byte ranges are absolute, contiguous, and exactly cover the data buffer. + let mut expected_start = base; + for (i, &(_, num_bytes)) in entries.iter().enumerate() { + let range = index.byte_range(i); + assert_eq!(range.start, expected_start); + assert_eq!(range.end - range.start, num_bytes as u64); + expected_start = range.end; + } + assert_eq!(expected_start, base + data_buf_size); + + // For flat pages rows == items, so the per-chunk items sum to the page total. + let total_items: u64 = (0..index.num_chunks()) + .map(|i| index.items_in_chunk(i)) + .sum(); + assert_eq!(total_items, items_in_page); + } + + #[test] + fn test_nested_detection_and_axes() { + // Repetition index (stride 2): three chunks holding 5, 4, 3 rows, no trailers. + let rep = rep_bytes_from(&[5, 0, 4, 0, 3, 0]); + + // Uniform leaf chunking: value counts 4, 4, 2. + let (words, data_buf_size) = words_from(&[(2, 8), (2, 8), (0, 8)]); + let index = build_chunk_index(&words, 10, 0, data_buf_size, Some(&rep), 1); + assert_eq!(index.row_mapping_debug(), "nested"); + assert_eq!(index.num_chunks(), 3); + // Rows come from the repetition index, not the value counts. + assert_eq!(index.first_row(0), 0); + assert_eq!(index.rows_in_chunk(0), 5); + assert_eq!(index.first_row(1), 5); + assert_eq!(index.rows_in_chunk(1), 4); + assert_eq!(index.first_row(2), 9); + assert_eq!(index.rows_in_chunk(2), 3); + // Items come from the value words. + assert_eq!(index.items_in_chunk(0), 4); + assert_eq!(index.items_in_chunk(1), 4); + assert_eq!(index.items_in_chunk(2), 2); + + // Non-uniform leaf chunking: value counts 8, 2, 5. + let (words_nu, dbs_nu) = words_from(&[(3, 8), (1, 8), (0, 8)]); + let index_nu = build_chunk_index(&words_nu, 15, 0, dbs_nu, Some(&rep), 1); + assert_eq!(index_nu.row_mapping_debug(), "nested"); + assert_eq!(index_nu.items_in_chunk(0), 8); + assert_eq!(index_nu.items_in_chunk(1), 2); + assert_eq!(index_nu.items_in_chunk(2), 5); + // The row axis is unchanged by the leaf chunking. + assert_eq!(index_nu.rows_in_chunk(0), 5); + } + + #[test] + fn test_uniform_flat_matches_prefix_sum_flat() { + // Distribution: 4 chunks of 4 values, last chunk 3 (15 items total). + let (words, data_buf_size) = words_from(&[(2, 8), (2, 8), (2, 8), (0, 8)]); + let uniform = build_chunk_index(&words, 15, 0, data_buf_size, None, 0); + assert_eq!(uniform.row_mapping_debug(), "uniform_flat"); + + // The same distribution expressed as a non-uniform Flat prefix-sum index. + let byte_starts = PrefixSums::from_deltas([8u64, 8, 8, 8].into_iter(), 4, 32); + let value_starts = PrefixSums::from_deltas([4u64, 4, 4, 3].into_iter(), 4, 15); + let flat = MiniBlockChunkIndex::new(0, byte_starts, RowMapping::Flat { value_starts }); + assert_eq!(flat.row_mapping_debug(), "flat"); + + // Lookup parity: identical byte ranges and item counts. + for i in 0..4 { + assert_eq!(uniform.byte_range(i), flat.byte_range(i)); + assert_eq!(uniform.items_in_chunk(i), flat.items_in_chunk(i)); + } + + // Scheduler parity across scan / single-row / partial / scattered multi-range. + let range_sets: Vec>> = vec![ + vec![0..15], + vec![0..1], + vec![7..8], + vec![14..15], + vec![3..10], + vec![0..2, 5..6, 12..15], + ]; + for ranges in &range_sets { + let from_uniform = ChunkInstructions::schedule_instructions(&uniform, ranges); + let from_flat = ChunkInstructions::schedule_instructions(&flat, ranges); + assert_eq!(from_uniform, from_flat, "mismatch for ranges {ranges:?}"); + } + + // A full scan yields one Absent, no-trailer instruction per chunk. + let full = ChunkInstructions::schedule_instructions(&uniform, &[0..15]); + assert_eq!(full.len(), 4); + for (i, inst) in full.iter().enumerate() { + assert_eq!(inst.chunk_idx, i); + assert_eq!(inst.preamble, PreambleAction::Absent); + assert_eq!(inst.rows_to_skip, 0); + assert!(!inst.take_trailer); + } + assert_eq!(full.iter().map(|i| i.rows_to_take).sum::(), 15); + } + + #[test] + fn test_deep_size_per_variant_below_legacy() { + // The previous representation cached 48 bytes per chunk (24 for ChunkMeta plus + // 24 for a rep-index block); every variant's heap must be well below that. + const LEGACY_PER_CHUNK: usize = 48; + let num_chunks = 3; + let heap = |index: &MiniBlockChunkIndex| index.deep_size_of_children(&mut Context::new()); + + let (uniform_words, uniform_dbs) = words_from(&[(2, 8), (2, 8), (0, 8)]); + let uniform = build_chunk_index(&uniform_words, 10, 0, uniform_dbs, None, 0); + assert_eq!(uniform.row_mapping_debug(), "uniform_flat"); + assert!(heap(&uniform) < LEGACY_PER_CHUNK * num_chunks); + + let (flat_words, flat_dbs) = words_from(&[(3, 8), (1, 8), (0, 8)]); + let flat = build_chunk_index(&flat_words, 11, 0, flat_dbs, None, 0); + assert_eq!(flat.row_mapping_debug(), "flat"); + assert!(heap(&flat) < LEGACY_PER_CHUNK * num_chunks); + // Flat carries a value-starts array that UniformFlat derives arithmetically. + assert!(heap(&flat) > heap(&uniform)); + + let rep = rep_bytes_from(&[4, 0, 3, 0, 3, 0]); + let (nested_words, nested_dbs) = words_from(&[(2, 8), (2, 8), (0, 8)]); + let nested = build_chunk_index(&nested_words, 10, 0, nested_dbs, Some(&rep), 1); + assert_eq!(nested.row_mapping_debug(), "nested"); + assert!(heap(&nested) < LEGACY_PER_CHUNK * num_chunks); + } + #[tokio::test] async fn test_fullzip_initialize_is_lazy() { use futures::{FutureExt, future::BoxFuture}; diff --git a/rust/lance-encoding/src/encodings/logical/primitive/chunk_index.rs b/rust/lance-encoding/src/encodings/logical/primitive/chunk_index.rs new file mode 100644 index 00000000000..19fb0fdb041 --- /dev/null +++ b/rust/lance-encoding/src/encodings/logical/primitive/chunk_index.rs @@ -0,0 +1,536 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The Lance Authors + +//! Compact per-page chunk index for the mini-block structural encoding. +//! +//! The chunk index is stored on disk in an extremely compressed form that +//! requires a lot of CPU to work with. However, extracting it out to its full +//! width can be RAM-intensive. As a compromise we extract into a prefix-sum +//! array that we fit into `u32` if possible and we avoid storing per-block row +//! counts when those are redundant. The scheduler looks chunks up by index +//! (byte range, leaf value count) and by row (which chunk holds a row). +//! +//! ```text +//! MiniBlockChunkIndex +//! |- base: u64 absolute file position of the value buffer +//! |- byte_starts: PrefixSums cumulative chunk byte sizes (all pages) +//! `- rows: RowMapping +//! |- UniformFlat { .. } flat page, uniform leaf chunking (arithmetic) +//! |- Flat { value_starts } flat page, non-uniform leaf chunking +//! `- Nested { row_starts, .. } repetition present; rows tracked as prefix sums +//! ``` + +use std::ops::Range; + +use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder}; +use lance_core::cache::{Context, DeepSizeOf}; + +/// Cumulative (prefix-sum) array of length `num_chunks + 1` (entry `0` is `0`, +/// the last entry is the grand total). Stored as `u32` when the total fits, +/// else `u64`. +#[derive(Debug, DeepSizeOf)] +pub enum PrefixSums { + U32(Vec), + U64(Vec), +} + +impl PrefixSums { + /// Builds the cumulative array from per-chunk `deltas`. `total` selects the + /// storage width (callers must pass the true sum); `num_chunks` only pre-sizes. + pub fn from_deltas(deltas: impl Iterator, num_chunks: usize, total: u64) -> Self { + if total <= u32::MAX as u64 { + let mut values = Vec::with_capacity(num_chunks + 1); + let mut acc = 0u32; + values.push(0); + for delta in deltas { + acc += delta as u32; + values.push(acc); + } + debug_assert_eq!(values.len(), num_chunks + 1); + debug_assert_eq!(acc as u64, total); + Self::U32(values) + } else { + let mut values = Vec::with_capacity(num_chunks + 1); + let mut acc = 0u64; + values.push(0); + for delta in deltas { + acc += delta; + values.push(acc); + } + debug_assert_eq!(values.len(), num_chunks + 1); + debug_assert_eq!(acc, total); + Self::U64(values) + } + } + + /// Builds a `PrefixSums` from an already-cumulative array (`[0, .., total]`), + /// narrowing to `u32` when the total fits. Avoids the deltas buffer + /// [`Self::from_deltas`] would need. + fn from_prefix(prefix: Vec) -> Self { + debug_assert!(!prefix.is_empty()); + debug_assert_eq!(prefix[0], 0); + let total = prefix.last().copied().unwrap_or(0); + if total <= u32::MAX as u64 { + Self::U32(prefix.into_iter().map(|v| v as u32).collect()) + } else { + Self::U64(prefix) + } + } + + /// Cumulative value at position `i` (i.e. the start of chunk `i`). + pub fn get(&self, i: usize) -> u64 { + match self { + Self::U32(values) => values[i] as u64, + Self::U64(values) => values[i], + } + } + + /// Start and end of chunk `i` (positions `i`, `i + 1`) behind one width + /// match -- halves the branching of two `get` calls on the hot per-chunk path. + pub fn get_pair(&self, i: usize) -> (u64, u64) { + match self { + Self::U32(values) => (values[i] as u64, values[i + 1] as u64), + Self::U64(values) => (values[i], values[i + 1]), + } + } + + /// Number of chunks (array length minus the trailing total). + pub fn num_chunks(&self) -> usize { + match self { + Self::U32(values) => values.len() - 1, + Self::U64(values) => values.len() - 1, + } + } + + /// Size of chunk `i` (the delta between consecutive cumulative values). + pub fn delta(&self, i: usize) -> u64 { + let (start, end) = self.get_pair(i); + end - start + } + + /// Index of the chunk whose half-open span `[get(i), get(i+1))` contains + /// `value`. On an exact hit against a chunk start, returns the *first* chunk + /// with that start (chunks can share a start row). + pub fn find(&self, value: u64) -> usize { + // Match the width once, then binary-search only the starts (not the + // trailing total). `partition_point` already yields the first of any + // duplicated starts; the `idx - 1` fallback is safe since `get(0) == 0`. + match self { + Self::U32(values) => { + let starts = &values[..values.len() - 1]; + let idx = starts.partition_point(|&start| (start as u64) < value); + if idx < starts.len() && starts[idx] as u64 == value { + idx + } else { + idx - 1 + } + } + Self::U64(values) => { + let starts = &values[..values.len() - 1]; + let idx = starts.partition_point(|&start| start < value); + if idx < starts.len() && starts[idx] == value { + idx + } else { + idx - 1 + } + } + } + } +} + +/// Leaf value counts per chunk, needed to decode. Tracked only for nested +/// pages; flat pages read items off the row mapping (rows == items). +#[derive(Debug, DeepSizeOf)] +pub enum ItemCounts { + /// Every non-last chunk holds the same number of values. + Uniform { + values_per_chunk: u64, + last_chunk_values: u64, + }, + /// `log2` of each chunk's value count, stored as one byte per chunk rather + /// than the full count because this index stays cached in RAM; the last + /// chunk is handled via `last_chunk_values`. + PerChunkLog { + logs: Vec, + last_chunk_values: u64, + }, +} + +impl ItemCounts { + fn get(&self, i: usize, num_chunks: usize) -> u64 { + match self { + Self::Uniform { + values_per_chunk, + last_chunk_values, + } => { + if i == num_chunks - 1 { + *last_chunk_values + } else { + *values_per_chunk + } + } + Self::PerChunkLog { + logs, + last_chunk_values, + } => { + if i == num_chunks - 1 { + *last_chunk_values + } else { + 1u64 << logs[i] + } + } + } + } +} + +/// How row ranges map onto chunks. +/// +/// Flat pages have row == value index and no preamble/trailer. Nested pages +/// track rows separately from leaf items and store a trailer bit per chunk; a +/// chunk's preamble is the previous chunk's trailer. +#[derive(Debug)] +pub enum RowMapping { + /// Flat page whose non-last chunks all hold `values_per_chunk` values, so + /// row->chunk is pure arithmetic. + UniformFlat { + values_per_chunk: u64, + last_chunk_values: u64, + num_chunks: usize, + }, + /// Flat page with non-uniform chunk sizes; `value_starts` are the cumulative + /// value counts (final entry == number of items in the page). + Flat { value_starts: PrefixSums }, + /// Nested page. `row_starts` are cumulative row counts (final entry == + /// number of rows in the page); `has_trailer[i]` is set when chunk `i` ends + /// with a partial list. + Nested { + row_starts: PrefixSums, + has_trailer: BooleanBuffer, + item_counts: ItemCounts, + }, +} + +impl DeepSizeOf for RowMapping { + fn deep_size_of_children(&self, context: &mut Context) -> usize { + match self { + Self::UniformFlat { .. } => 0, + Self::Flat { value_starts } => value_starts.deep_size_of_children(context), + Self::Nested { + row_starts, + has_trailer, + item_counts, + } => { + row_starts.deep_size_of_children(context) + + has_trailer.len().div_ceil(8) + + item_counts.deep_size_of_children(context) + } + } + } +} + +/// Compact per-page chunk index that avoids fully materializing the repetition +/// index into u64's to save RAM. See the module docs for the layout. +#[derive(Debug)] +pub struct MiniBlockChunkIndex { + base: u64, + byte_starts: PrefixSums, + rows: RowMapping, +} + +impl MiniBlockChunkIndex { + pub fn new(base: u64, byte_starts: PrefixSums, rows: RowMapping) -> Self { + Self { + base, + byte_starts, + rows, + } + } + + /// Number of chunks in the page. + pub fn num_chunks(&self) -> usize { + self.byte_starts.num_chunks() + } + + /// Absolute byte range of chunk `i` within the file. + pub fn byte_range(&self, i: usize) -> Range { + let (start, end) = self.byte_starts.get_pair(i); + (self.base + start)..(self.base + end) + } + + /// Number of leaf values in chunk `i` (passed to the value decompressor). + pub fn items_in_chunk(&self, i: usize) -> u64 { + let num_chunks = self.num_chunks(); + match &self.rows { + RowMapping::UniformFlat { + values_per_chunk, + last_chunk_values, + .. + } => { + if i == num_chunks - 1 { + *last_chunk_values + } else { + *values_per_chunk + } + } + RowMapping::Flat { value_starts } => value_starts.delta(i), + RowMapping::Nested { item_counts, .. } => item_counts.get(i, num_chunks), + } + } + + /// Index of the chunk that contains `row`. + pub fn find_chunk(&self, row: u64) -> usize { + match &self.rows { + RowMapping::UniformFlat { + values_per_chunk, + num_chunks, + .. + } => ((row / values_per_chunk) as usize).min(num_chunks - 1), + RowMapping::Flat { value_starts } => value_starts.find(row), + RowMapping::Nested { row_starts, .. } => row_starts.find(row), + } + } + + /// First row (relative to the page) that begins in chunk `i`. + pub fn first_row(&self, i: usize) -> u64 { + match &self.rows { + RowMapping::UniformFlat { + values_per_chunk, .. + } => i as u64 * values_per_chunk, + RowMapping::Flat { value_starts } => value_starts.get(i), + RowMapping::Nested { row_starts, .. } => row_starts.get(i), + } + } + + /// Number of rows that start in chunk `i`, including a trailer but not a + /// preamble (the previous `starts_including_trailer`). + pub fn rows_in_chunk(&self, i: usize) -> u64 { + let num_chunks = self.num_chunks(); + match &self.rows { + RowMapping::UniformFlat { + values_per_chunk, + last_chunk_values, + .. + } => { + if i == num_chunks - 1 { + *last_chunk_values + } else { + *values_per_chunk + } + } + RowMapping::Flat { value_starts } => value_starts.delta(i), + RowMapping::Nested { row_starts, .. } => row_starts.delta(i), + } + } + + /// Whether chunk `i` begins with a preamble (a continuation of the previous + /// chunk's list). Always false for flat pages; for nested pages this is the + /// previous chunk's trailer. + pub fn has_preamble(&self, i: usize) -> bool { + match &self.rows { + RowMapping::Nested { has_trailer, .. } => i > 0 && has_trailer.value(i - 1), + _ => false, + } + } + + /// Whether chunk `i` ends with a trailer (a partial list continued in the + /// next chunk). Always false for flat pages. + pub fn has_trailer(&self, i: usize) -> bool { + match &self.rows { + RowMapping::Nested { has_trailer, .. } => has_trailer.value(i), + _ => false, + } + } + + /// Name of the active row-mapping variant, used to assert detection in tests. + #[cfg(test)] + pub fn row_mapping_debug(&self) -> &'static str { + match &self.rows { + RowMapping::UniformFlat { .. } => "uniform_flat", + RowMapping::Flat { .. } => "flat", + RowMapping::Nested { .. } => "nested", + } + } + + /// Builds a nested index from raw repetition-index bytes, using placeholder + /// byte offsets and item counts. Only the row axis is populated, which is + /// all the scheduler exercises. + #[cfg(test)] + pub fn new_nested_for_test(rep_bytes: &[u8], stride: usize) -> Self { + let (row_starts, has_trailer) = parse_nested_rep(rep_bytes, stride); + let num_chunks = row_starts.num_chunks(); + let byte_starts = PrefixSums::from_deltas( + std::iter::repeat_n(8u64, num_chunks), + num_chunks, + 8 * num_chunks as u64, + ); + Self { + base: 0, + byte_starts, + rows: RowMapping::Nested { + row_starts, + has_trailer, + item_counts: ItemCounts::Uniform { + values_per_chunk: 1, + last_chunk_values: 1, + }, + }, + } + } +} + +impl DeepSizeOf for MiniBlockChunkIndex { + fn deep_size_of_children(&self, context: &mut Context) -> usize { + self.byte_starts.deep_size_of_children(context) + self.rows.deep_size_of_children(context) + } +} + +/// Parses a mini-block repetition index into the compact nested row mapping. +/// +/// Bytes are `u64`s in groups of `stride`; the first two are `ends` (lists +/// finishing in the chunk) and `partial` (leftover items). Only cumulative row +/// starts and a trailer bit are kept: `has_preamble[i] = has_trailer[i-1]` and +/// `starts_including_trailer = ends + has_trailer - has_preamble`. +pub fn parse_nested_rep(rep_bytes: &[u8], stride: usize) -> (PrefixSums, BooleanBuffer) { + // Read the two `u64`s per group straight from the little-endian bytes rather + // than copying the buffer to reinterpret it. The caller guarantees + // `rep_bytes.len() % 8 == 0`, so the 8-byte windows stay in bounds. + const WORD: usize = std::mem::size_of::(); + let read_word = |word_idx: usize| -> u64 { + let byte = word_idx * WORD; + u64::from_le_bytes(rep_bytes[byte..byte + WORD].try_into().unwrap()) + }; + let num_chunks = (rep_bytes.len() / WORD) / stride; + + let mut has_trailer_builder = BooleanBufferBuilder::new(num_chunks); + // Accumulate the cumulative row starts in a single pass (entry 0 is 0, the + // trailing entry is the total) so there is no separate deltas buffer. + let mut row_starts = Vec::with_capacity(num_chunks + 1); + row_starts.push(0u64); + let mut acc = 0u64; + let mut chunk_has_preamble = false; + + for i in 0..num_chunks { + let base_idx = i * stride; + let ends = read_word(base_idx); + let partial = read_word(base_idx + 1); + + let has_trailer = partial > 0; + let starts_including_trailer = ends + (has_trailer as u64) - (chunk_has_preamble as u64); + + has_trailer_builder.append(has_trailer); + acc += starts_including_trailer; + row_starts.push(acc); + + chunk_has_preamble = has_trailer; + } + + ( + PrefixSums::from_prefix(row_starts), + has_trailer_builder.finish(), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::buffer::LanceBuffer; + + /// Reference decode: the previous per-block repetition index, used as an + /// oracle for the compact `parse_nested_rep`. + struct RefBlock { + first_row: u64, + starts_including_trailer: u64, + has_preamble: bool, + has_trailer: bool, + } + + fn reference_decode(rep_bytes: &[u8], stride: usize) -> Vec { + let buffer = LanceBuffer::from(rep_bytes.to_vec()); + let u64_slice = buffer.borrow_to_typed_slice::(); + let n = u64_slice.len() / stride; + let mut blocks = Vec::with_capacity(n); + let mut chunk_has_preamble = false; + let mut offset = 0u64; + for i in 0..n { + let base_idx = i * stride; + let ends = u64_slice[base_idx]; + let partial = u64_slice[base_idx + 1]; + let has_trailer = partial > 0; + let starts_including_trailer = + ends + (has_trailer as u64) - (chunk_has_preamble as u64); + blocks.push(RefBlock { + first_row: offset, + starts_including_trailer, + has_preamble: chunk_has_preamble, + has_trailer, + }); + chunk_has_preamble = has_trailer; + offset += starts_including_trailer; + } + blocks + } + + fn rep_bytes(values: &[u64]) -> Vec { + values.iter().flat_map(|v| v.to_le_bytes()).collect() + } + + #[test] + fn test_prefix_sums_u32() { + let sums = PrefixSums::from_deltas([2u64, 3, 5].into_iter(), 3, 10); + assert!(matches!(sums, PrefixSums::U32(_))); + assert_eq!(sums.num_chunks(), 3); + assert_eq!(sums.get(0), 0); + assert_eq!(sums.get(1), 2); + assert_eq!(sums.get(3), 10); + assert_eq!(sums.delta(1), 3); + } + + #[test] + fn test_prefix_sums_u64_selected_by_total() { + let big = u32::MAX as u64 + 1; + let sums = PrefixSums::from_deltas([big].into_iter(), 1, big); + assert!(matches!(sums, PrefixSums::U64(_))); + assert_eq!(sums.get(1), big); + } + + #[test] + fn test_prefix_sums_find() { + // Chunk starts: 0, 5, 5, 12 (a zero-width chunk creates a duplicate start) + let sums = PrefixSums::from_deltas([5u64, 0, 7].into_iter(), 3, 12); + // Inside the first chunk + assert_eq!(sums.find(3), 0); + // Exact match on a duplicated start returns the first such chunk + assert_eq!(sums.find(5), 1); + // Inside the last chunk + assert_eq!(sums.find(11), 2); + // Start of the first chunk + assert_eq!(sums.find(0), 0); + } + + #[test] + fn test_parse_nested_rep_matches_reference() { + let cases: Vec> = vec![ + vec![5, 2, 3, 0, 4, 7, 2, 0], + vec![5, 2, 3, 3, 20, 0], + vec![0, 5, 0, 3, 10, 0], + vec![1, 0], + ]; + for values in cases { + let bytes = rep_bytes(&values); + let reference = reference_decode(&bytes, 2); + let (row_starts, has_trailer) = parse_nested_rep(&bytes, 2); + assert_eq!(row_starts.num_chunks(), reference.len()); + for (i, block) in reference.iter().enumerate() { + assert_eq!(row_starts.get(i), block.first_row, "first_row[{i}]"); + assert_eq!( + row_starts.delta(i), + block.starts_including_trailer, + "starts_including_trailer[{i}]" + ); + assert_eq!(has_trailer.value(i), block.has_trailer, "has_trailer[{i}]"); + let derived_preamble = i > 0 && has_trailer.value(i - 1); + assert_eq!(derived_preamble, block.has_preamble, "has_preamble[{i}]"); + } + } + } +}