diff --git a/docs/src/format/file/encoding.md b/docs/src/format/file/encoding.md index f1d14a3123b..71b293ce06f 100644 --- a/docs/src/format/file/encoding.md +++ b/docs/src/format/file/encoding.md @@ -584,10 +584,10 @@ on a per-value basis. We use ☑️ to mark a technique that is applied on a per | --------------- | --------------------- | ------------------------ | -------------------------- | | Flat | ✅ (2.1) | ✅ (2.1) | ✅ (2.1) | | Variable | ✅ (2.1) | ✅ (2.1) | ✅ (2.1) | -| Constant | ✅ (2.1) | ❓ | ❓ | -| Range | ✅ (2.3) | ❌ | ❓ | -| Delta | ✅ (2.3) | ❌ | ❓ | -| Dictionary | ✅ (2.3) | ❌ | ❓ | +| Constant | ✅ (2.1) | ❓ | ✅ (2.3, variable offsets) | +| Range | ✅ (2.3) | ❌ | ✅ (2.3, variable offsets) | +| Delta | ✅ (2.3) | ❌ | ✅ (2.3, variable offsets) | +| Dictionary | ✅ (2.3) | ❌ | ✅ (2.3, variable offsets) | | Bitpacking | ✅ (2.1) | ❓ | ✅ (2.1) | | Fsst | ❓ | ✅ (2.1) | ✅ (2.1) | | Rle | ✅ (2.2) | ❌ | ✅ (2.1) | @@ -599,28 +599,49 @@ in various contexts. ### Generic Block Sequences -Starting in Lance 2.3, block compression can describe unsigned `u32` and `u64` sequences with a bounded, -zero-or-one-payload codec tree. The containing layout supplies the value type and cardinality. +Starting in Lance 2.3, block compression can encode unsigned `u32` and `u64` sequences with a shared descriptor +contract. Direct codecs such as Flat, bitpacking, RLE, and Dictionary also support non-monotonic values; Range and +Delta require non-decreasing input. The containing layout supplies the value type and cardinality. A descriptor +constructs a concrete decoder tree whose nodes own their child codecs and framing validation. -`Range` is metadata-only. It stores the unsigned width, first value, and positive step. The value at index `i` -is `start + step * i`; readers reject cardinalities below two, overflow, and widths other than 32 or 64 bits. +Writers select a concrete compressor from a bounded set: `Constant`, `Range`, `Flat`, bitpacking, RLE, +`Dictionary`, `Delta`, and general compression. Candidate costs include the protobuf descriptor, codec framing, +buffer entries, and alignment. Payload estimates are exact except for `General`, which extrapolates from a bounded +sample; only the selected compressor is invoked to materialize payloads. Writers use the following canonical +metadata-only codecs: + +- An empty sequence uses `Constant` with no scalar. +- A non-empty constant sequence uses `Constant` with one little-endian scalar. +- An arithmetic progression with a positive step uses `Range`. + +The first block-dictionary writer only emits Dictionary for `u64` sequences. +The first block-sequence grammar permits `General` only as the outer root around a `Flat` child; readers reject +inner, sibling, repeated, or non-`Flat` `General` transforms. + +`Range` stores the unsigned width, first value, and positive step. The value at index `i` is +`start + step * i`; readers reject overflow and widths other than 32 or 64 bits. ```protobuf %%% proto.message.Range %%% ``` -`Delta` stores the first value inline. Its child describes the `n - 1` non-negative adjacent differences and -determines whether the Delta tree has a payload. Zero differences are valid. Readers reconstruct the sequence -with checked prefix sums. +`Delta` stores the first value inline. Its child represents the `n - 1` non-negative adjacent differences. Delta +has a payload exactly when its child has one. Zero differences are valid. Readers reconstruct the sequence with +checked prefix sums. ```protobuf %%% proto.message.Delta %%% ``` -Block Dictionary stores `u32` indices and typed dictionary items as child codec trees. When at least one child -has a payload, both children are combined into one outer payload: +RLE and block Dictionary expose no outer payload when both children are metadata-only. Otherwise, they +combine their children into one framed outer payload: ```text +RLE payload: + u64 values_payload_bytes + values payload + run-lengths payload + Dictionary payload: u64 indices_payload_bytes u64 items_payload_bytes @@ -628,8 +649,11 @@ Dictionary payload: dictionary-items payload ``` -The framed length for a metadata-only child must be zero. Readers validate the item count, frame boundaries, -child cardinalities, and every index. Lance 2.0 through 2.2 writers do not emit Range, Delta, or block Dictionary. +The framed length for a metadata-only child must be zero. Readers validate child cardinalities, run-length sums, +dictionary index bounds, and all frame boundaries. + +Lance 2.0 through 2.2 writers do not emit `Range`, `Delta`, or block Dictionary. Their block selector order +and payload shapes remain unchanged. ### Flat @@ -648,6 +672,25 @@ When applied in a mini-block context each block may have a different number of v until we find the point that would exceed 4,096 bytes and then use the most recent power of 2 number of values that we have passed. +Lance 2.0 through 2.2 store each mini-block as one legacy buffer containing chunk-local Flat offsets followed by +the value bytes. Starting in Lance 2.3, the writer also evaluates a generic-offset container. Generic offsets are +zero-based and independently encoded in each chunk with one page-wide concrete offset codec: + +```text +Legacy chunk: + [adjusted Flat offsets][value bytes] + +Generic chunk: + [optional offset payload][value bytes] +``` + +The generic form has one buffer size per chunk for metadata-only offset codecs and two for payload-bearing offset +codecs. It is selected only when its complete serialized size is strictly smaller than the legacy form. A Flat +offset descriptor always denotes the legacy form, which keeps the two wire shapes unambiguous. Both forms keep +offsets and values in the same mini-block chunk, so a random read still fetches only the selected chunk. Fields that +explicitly request an outer General compressor retain the legacy form because that transform can change the +complete-container winner after offset selection. + ### Constant Constant compression is currently only utilized in a few specialized scenarios such as all-null arrays. diff --git a/rust/lance-encoding/src/compression.rs b/rust/lance-encoding/src/compression.rs index 133aa3f5df8..65d349b9d1d 100644 --- a/rust/lance-encoding/src/compression.rs +++ b/rust/lance-encoding/src/compression.rs @@ -684,9 +684,21 @@ impl DefaultCompressionStrategy { let data_size = data.expect_single_stat::(Stat::DataSize); let max_len = data.expect_single_stat::(Stat::MaxLength); - // Explicitly disable all compression. + let build_binary = || { + let generic_offsets_own_final_payload = + compression.is_none() || compression == Some("none"); + if self.version.resolve() >= LanceFileVersion::V2_3 && generic_offsets_own_final_payload + { + BinaryMiniBlockEncoder::with_generic_offsets(params.minichunk_size, params.clone()) + } else { + BinaryMiniBlockEncoder::new(params.minichunk_size) + } + }; + + // "none" disables general compression but still permits structural + // offset codecs in 2.3. if compression == Some("none") { - return Ok(Box::new(BinaryMiniBlockEncoder::new(params.minichunk_size))); + return Ok(Box::new(build_binary())); } let use_fsst = compression == Some("fsst") @@ -699,7 +711,7 @@ impl DefaultCompressionStrategy { let mut base_encoder: Box = if use_fsst { Box::new(FsstMiniBlockEncoder::new(params.minichunk_size)) } else { - Box::new(BinaryMiniBlockEncoder::new(params.minichunk_size)) + Box::new(build_binary()) }; // Wrap with general compression when configured (except FSST / none). @@ -1887,6 +1899,109 @@ mod tests { check_uncompressed_encoding(&encoding, true); } + #[test] + fn test_variable_offset_codec_is_version_gated() { + let num_values = 2_048_u64; + let offsets = (0..=num_values) + .map(|index| (index * 3) as i32) + .collect::>(); + let mut variable = VariableWidthBlock { + data: LanceBuffer::from(vec![7_u8; num_values as usize * 3]), + offsets: LanceBuffer::reinterpret_vec(offsets), + bits_per_offset: 32, + num_values, + block_info: BlockInfo::default(), + }; + variable.compute_stat(); + let data = DataBlock::VariableWidth(variable); + let field = create_test_field("bytes", DataType::Binary); + + for version in [ + LanceFileVersion::V2_0, + LanceFileVersion::V2_1, + LanceFileVersion::V2_2, + ] { + let strategy = DefaultCompressionStrategy::new().with_version(version); + let compressor = strategy.create_miniblock_compressor(&field, &data).unwrap(); + let (compressed, encoding) = compressor + .compress(data.clone(), miniblock_context()) + .unwrap(); + let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else { + panic!("expected Variable encoding for {version}"); + }; + assert!(matches!( + variable + .offsets + .as_deref() + .and_then(|offsets| offsets.compression.as_ref()), + Some(Compression::Flat(_)) + )); + assert_eq!(compressed.data.len(), 1); + assert!( + compressed + .chunks + .iter() + .all(|chunk| chunk.buffer_sizes.len() == 1) + ); + } + + let strategy = DefaultCompressionStrategy::new().with_version(LanceFileVersion::V2_3); + let compressor = strategy.create_miniblock_compressor(&field, &data).unwrap(); + let (compressed, encoding) = compressor.compress(data, miniblock_context()).unwrap(); + let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else { + panic!("expected Variable encoding for 2.3"); + }; + assert!(matches!( + variable + .offsets + .as_deref() + .and_then(|offsets| offsets.compression.as_ref()), + Some(Compression::Range(_)) + )); + assert_eq!(compressed.data.len(), 1); + } + + #[test] + #[cfg(any(feature = "lz4", feature = "zstd"))] + fn test_v2_3_explicit_general_compression_keeps_legacy_offsets() { + let num_values = 2_048_u64; + let offsets = (0..=num_values) + .map(|index| (index * 3) as i32) + .collect::>(); + let mut variable = VariableWidthBlock { + data: LanceBuffer::from(vec![7_u8; num_values as usize * 3]), + offsets: LanceBuffer::reinterpret_vec(offsets), + bits_per_offset: 32, + num_values, + block_info: BlockInfo::default(), + }; + variable.compute_stat(); + let data = DataBlock::VariableWidth(variable); + let mut field = create_test_field("bytes", DataType::Binary); + field.metadata.insert( + COMPRESSION_META_KEY.to_string(), + if cfg!(feature = "lz4") { "lz4" } else { "zstd" }.to_string(), + ); + + let strategy = DefaultCompressionStrategy::new().with_version(LanceFileVersion::V2_3); + let compressor = strategy.create_miniblock_compressor(&field, &data).unwrap(); + let (_, encoding) = compressor.compress(data, miniblock_context()).unwrap(); + let value_encoding = match encoding.compression.as_ref().unwrap() { + Compression::General(general) => general.values.as_deref().unwrap(), + _ => &encoding, + }; + let Some(Compression::Variable(variable)) = value_encoding.compression.as_ref() else { + panic!("expected Variable encoding"); + }; + assert!(matches!( + variable + .offsets + .as_deref() + .and_then(|offsets| offsets.compression.as_ref()), + Some(Compression::Flat(_)) + )); + } + #[test] fn test_field_metadata_none_compression() { // Prepare field with metadata for none compression diff --git a/rust/lance-encoding/src/encodings/logical/primitive.rs b/rust/lance-encoding/src/encodings/logical/primitive.rs index cad9ade7d54..64a302ae148 100644 --- a/rust/lance-encoding/src/encodings/logical/primitive.rs +++ b/rust/lance-encoding/src/encodings/logical/primitive.rs @@ -6602,16 +6602,18 @@ mod tests { ChunkInstructions, DataBlock, DecodeMiniBlockTask, FixedPerValueDecompressor, FixedWidthDataBlock, FullZipCacheableState, FullZipDecodeDetails, FullZipReadSource, FullZipRepIndexDetails, FullZipScheduler, LazyLevels, LevelCodec, LevelCursor, LevelPlan, - MiniBlockChunk, MiniBlockChunkIndex, MiniBlockCompressed, PerValueDecompressor, - PreambleAction, RunEndsBuilder, RunPosition, RunStorage, StructuralPageScheduler, - VariableFullZipDecoder, dense_levels_from_block, validate_complex_all_null_levels, + MiniBlockChunk, MiniBlockChunkIndex, MiniBlockCompressed, MiniBlockScheduler, + PerValueDecompressor, PreambleAction, RunEndsBuilder, RunPosition, RunStorage, + StructuralPageScheduler, VariableFullZipDecoder, dense_levels_from_block, + validate_complex_all_null_levels, }; + use crate::EncodingsIo; use crate::buffer::LanceBuffer; use crate::compression::{BlockCompressor, DefaultDecompressionStrategy}; use crate::constants::{ - COMPRESSION_LEVEL_META_KEY, COMPRESSION_META_KEY, DICT_VALUES_COMPRESSION_LEVEL_META_KEY, - DICT_VALUES_COMPRESSION_META_KEY, STRUCTURAL_ENCODING_META_KEY, - STRUCTURAL_ENCODING_MINIBLOCK, + COMPRESSION_LEVEL_META_KEY, COMPRESSION_META_KEY, DICT_DIVISOR_META_KEY, + DICT_VALUES_COMPRESSION_LEVEL_META_KEY, DICT_VALUES_COMPRESSION_META_KEY, + STRUCTURAL_ENCODING_META_KEY, STRUCTURAL_ENCODING_MINIBLOCK, }; use crate::data::BlockInfo; use crate::decoder::{PageEncoding, StructuralFieldDecoder}; @@ -6628,8 +6630,14 @@ mod tests { use arrow_array::{ArrayRef, Int8Array, StringArray}; use arrow_buffer::ScalarBuffer; use arrow_schema::{DataType, Field as ArrowField}; + use futures::{FutureExt, future::BoxFuture}; + use prost::Message; use std::collections::HashMap; - use std::{collections::VecDeque, sync::Arc}; + use std::{ + collections::VecDeque, + ops::Range, + sync::{Arc, Mutex}, + }; #[test] fn test_is_narrow() { @@ -8681,6 +8689,268 @@ mod tests { pages.into_iter().next().unwrap() } + fn variable_offset_test_field() -> arrow_schema::Field { + arrow_schema::Field::new("c", DataType::Utf8, false).with_metadata(HashMap::from([ + ( + STRUCTURAL_ENCODING_META_KEY.to_string(), + STRUCTURAL_ENCODING_MINIBLOCK.to_string(), + ), + (COMPRESSION_META_KEY.to_string(), "none".to_string()), + (DICT_DIVISOR_META_KEY.to_string(), "100000".to_string()), + ])) + } + + fn variable_offset_test_array(lengths: &[usize], num_rows: usize) -> ArrayRef { + Arc::new(StringArray::from_iter_values((0..num_rows).map(|index| { + let length = lengths[index % lengths.len()]; + assert!(length >= 4); + format!("{index:04x}{}", "x".repeat(length - 4)) + }))) + } + + fn miniblock_layout(page: &crate::encoder::EncodedPage) -> &pb21::MiniBlockLayout { + let PageEncoding::Structural(layout) = &page.description else { + panic!("Expected structural page encoding"); + }; + let pb21::page_layout::Layout::MiniBlockLayout(layout) = layout.layout.as_ref().unwrap() + else { + panic!("Expected mini-block page layout"); + }; + layout + } + + fn variable_offset_compression( + page: &crate::encoder::EncodedPage, + ) -> &pb21::compressive_encoding::Compression { + let value_encoding = miniblock_layout(page).value_compression.as_ref().unwrap(); + let Compression::Variable(variable) = value_encoding.compression.as_ref().unwrap() else { + panic!("Expected Variable value compression"); + }; + variable + .offsets + .as_deref() + .and_then(|offsets| offsets.compression.as_ref()) + .unwrap() + } + + fn variable_value_wire_bytes(page: &crate::encoder::EncodedPage) -> usize { + let descriptor_bytes = miniblock_layout(page) + .value_compression + .as_ref() + .unwrap() + .encoded_len(); + descriptor_bytes + page.data.iter().map(LanceBuffer::len).sum::() + } + + #[derive(Debug)] + struct RecordingScheduler { + data: Bytes, + requests: Mutex>>>, + } + + impl RecordingScheduler { + fn new(data: Bytes) -> Self { + Self { + data, + requests: Mutex::new(Vec::new()), + } + } + + fn take_requests(&self) -> Vec>> { + std::mem::take(&mut *self.requests.lock().unwrap()) + } + } + + impl EncodingsIo for RecordingScheduler { + fn submit_request( + &self, + ranges: Vec>, + _priority: u64, + ) -> BoxFuture<'static, lance_core::Result>> { + self.requests.lock().unwrap().push(ranges.clone()); + let data = ranges + .into_iter() + .map(|range| self.data.slice(range.start as usize..range.end as usize)) + .collect(); + std::future::ready(Ok(data)).boxed() + } + } + + async fn miniblock_take_request_shape(page: &crate::encoder::EncodedPage) -> [usize; 6] { + let mut position = 0_u64; + let buffer_offsets_and_sizes = page + .data + .iter() + .map(|buffer| { + let size = buffer.len() as u64; + let descriptor = (position, size); + position += size; + descriptor + }) + .collect::>(); + let mut bytes = Vec::with_capacity(position as usize); + for buffer in &page.data { + bytes.extend_from_slice(buffer); + } + + let recorder = Arc::new(RecordingScheduler::new(Bytes::from(bytes))); + let io: Arc = recorder.clone(); + let decompression = DefaultDecompressionStrategy::default(); + let mut cold = MiniBlockScheduler::try_new( + &buffer_offsets_and_sizes, + 0, + page.num_rows, + miniblock_layout(page), + &decompression, + ) + .unwrap(); + let cached = cold.initialize(&io).await.unwrap(); + let initialize_requests = recorder.take_requests(); + + let cold_tasks = cold.schedule_ranges(&[123..124], &io).unwrap(); + let cold_requests = recorder.take_requests(); + for task in cold_tasks { + task.decoder_fut.await.unwrap(); + } + + let mut warm = MiniBlockScheduler::try_new( + &buffer_offsets_and_sizes, + 0, + page.num_rows, + miniblock_layout(page), + &decompression, + ) + .unwrap(); + warm.load(&cached); + let warm_tasks = warm.schedule_ranges(&[123..124], &io).unwrap(); + let warm_requests = recorder.take_requests(); + for task in warm_tasks { + task.decoder_fut.await.unwrap(); + } + + [ + initialize_requests.len(), + initialize_requests.iter().map(Vec::len).sum(), + cold_requests.len(), + cold_requests.iter().map(Vec::len).sum(), + warm_requests.len(), + warm_requests.iter().map(Vec::len).sum(), + ] + } + + #[tokio::test] + async fn test_v2_3_variable_offsets_use_complete_serialized_cost() { + let cases = [ + (&[16_usize, 22, 17, 20][..], "legacy"), + (&[4_usize, 10, 5, 8][..], "delta"), + (&[8_usize][..], "range"), + ]; + for (lengths, expected) in cases { + let array = variable_offset_test_array(lengths, 10_000); + let legacy = encode_first_page( + variable_offset_test_field(), + array.clone(), + LanceFileVersion::V2_2, + ) + .await; + let selected = + encode_first_page(variable_offset_test_field(), array, LanceFileVersion::V2_3) + .await; + + match expected { + "legacy" => { + assert!(matches!( + variable_offset_compression(&selected), + Compression::Flat(_) + )); + assert_eq!( + variable_value_wire_bytes(&selected), + variable_value_wire_bytes(&legacy) + ); + } + "delta" => assert!(matches!( + variable_offset_compression(&selected), + Compression::Delta(_) + )), + "range" => assert!(matches!( + variable_offset_compression(&selected), + Compression::Range(_) + )), + _ => unreachable!(), + } + if expected != "legacy" { + assert!( + variable_value_wire_bytes(&selected) < variable_value_wire_bytes(&legacy), + "{expected} generic container must be strictly smaller than legacy" + ); + } + } + } + + #[tokio::test] + async fn test_v2_3_variable_offsets_use_delta_range_for_increasing_lengths() { + let array = Arc::new(StringArray::from_iter_values( + (0..262_144).map(|index| "x".repeat(4 + index % 64)), + )); + let legacy = encode_first_page( + variable_offset_test_field(), + array.clone(), + LanceFileVersion::V2_2, + ) + .await; + let selected = + encode_first_page(variable_offset_test_field(), array, LanceFileVersion::V2_3).await; + + let Compression::Delta(delta) = variable_offset_compression(&selected) else { + panic!( + "expected Delta offsets, got {:?}", + variable_offset_compression(&selected) + ); + }; + assert!(matches!( + delta + .deltas + .as_deref() + .and_then(|deltas| deltas.compression.as_ref()), + Some(Compression::Range(_)) + )); + assert!( + variable_value_wire_bytes(&selected) < variable_value_wire_bytes(&legacy), + "Delta(Range) generic container must be strictly smaller than legacy" + ); + } + + #[tokio::test] + async fn test_v2_3_generic_offsets_do_not_add_take_requests() { + let array = variable_offset_test_array(&[4, 10, 5, 8], 10_000); + let legacy = encode_first_page( + variable_offset_test_field(), + array.clone(), + LanceFileVersion::V2_2, + ) + .await; + let generic = + encode_first_page(variable_offset_test_field(), array, LanceFileVersion::V2_3).await; + + assert!(matches!( + variable_offset_compression(&legacy), + Compression::Flat(_) + )); + assert!(matches!( + variable_offset_compression(&generic), + Compression::Delta(_) + )); + assert_eq!(miniblock_layout(&legacy).num_buffers, 1); + assert_eq!(miniblock_layout(&generic).num_buffers, 2); + assert_eq!(legacy.data.len(), 2); + assert_eq!(generic.data.len(), 2); + + let legacy_requests = miniblock_take_request_shape(&legacy).await; + let generic_requests = miniblock_take_request_shape(&generic).await; + assert_eq!(legacy_requests, [1, 1, 1, 1, 1, 1]); + assert_eq!(generic_requests, legacy_requests); + } + #[tokio::test] async fn test_constant_layout_out_of_line_fixed_size_binary_v2_2() { use crate::format::pb21::page_layout::Layout;