diff --git a/rust/lance-encoding/benches/common/mod.rs b/rust/lance-encoding/benches/common/mod.rs index 3539fdd90b4..3fc77f4f03b 100644 --- a/rust/lance-encoding/benches/common/mod.rs +++ b/rust/lance-encoding/benches/common/mod.rs @@ -26,7 +26,6 @@ use lance_encoding::{ }, }, encodings::logical::primitive::{fullzip::PerValueCompressor, miniblock::MiniBlockCompressor}, - format::pb21::CompressiveEncoding, }; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -141,7 +140,7 @@ impl CompressionStrategy for BenchCompressionStrategy { &self, field: &Field, data: &DataBlock, - ) -> Result<(Box, CompressiveEncoding)> { + ) -> Result> { let params = self.field_params(field); if self.encoding == BenchEncoding::StructuralU32 && let Some(compressor) = try_fixed_u8_rle_block(data, ¶ms)? diff --git a/rust/lance-encoding/benches/decoder.rs b/rust/lance-encoding/benches/decoder.rs index 98d7caaf9e2..fa831b69b25 100644 --- a/rust/lance-encoding/benches/decoder.rs +++ b/rust/lance-encoding/benches/decoder.rs @@ -651,7 +651,7 @@ where #[cfg(feature = "bitpacking")] fn typed_view_unchunk(buffer: LanceBuffer, uncompressed_bits: u64, num_values: u64) -> DataBlock { InlineBitpacking::new(uncompressed_bits) - .decompress(buffer, num_values) + .decompress(Some(buffer), num_values) .unwrap() } diff --git a/rust/lance-encoding/src/compression.rs b/rust/lance-encoding/src/compression.rs index f20c8eb04ae..9c201375c9c 100644 --- a/rust/lance-encoding/src/compression.rs +++ b/rust/lance-encoding/src/compression.rs @@ -61,10 +61,7 @@ use crate::{ value::{ValueDecompressor, ValueEncoder}, }, }, - format::{ - ProtobufUtils21, - pb21::{CompressiveEncoding, compressive_encoding::Compression}, - }, + format::pb21::{CompressiveEncoding, compressive_encoding::Compression}, statistics::{GetStat, Stat}, }; @@ -98,11 +95,11 @@ const RLE_BLOCK_HEADER_BYTES: u128 = std::mem::size_of::() as u128; /// required (e.g. when encoding metadata buffers like a dictionary or for encoding rep/def /// mini-block chunks) pub trait BlockCompressor: std::fmt::Debug + Send + Sync { - /// Compress the data into a single buffer + /// Compress the data into zero or one buffers and describe the codec used. /// - /// Also returns a description of the compression that can be used to decompress - /// when reading the data back - fn compress(&self, data: DataBlock) -> Result; + /// `None` represents a metadata-only codec. `Some` represents a physical + /// payload, including a zero-byte payload. + fn compress(&self, data: DataBlock) -> Result<(Option, CompressiveEncoding)>; } /// A trait to pick which compression to use for given data @@ -118,12 +115,12 @@ pub trait BlockCompressor: std::fmt::Debug + Send + Sync { /// used for narrow data types (both fixed and variable length) where we can /// fit many values into an 16KiB block. pub trait CompressionStrategy: Send + Sync + std::fmt::Debug { - /// Create a block compressor for the given data + /// Create a block compressor for the given data. fn create_block_compressor( &self, field: &Field, data: &DataBlock, - ) -> Result<(Box, CompressiveEncoding)>; + ) -> Result>; /// Create a per-value compressor for the given data fn create_per_value( @@ -140,6 +137,19 @@ pub trait CompressionStrategy: Send + Sync + std::fmt::Debug { ) -> Result>; } +pub(crate) fn compress_required_block( + strategy: &dyn CompressionStrategy, + field: &Field, + data: DataBlock, +) -> Result<(LanceBuffer, CompressiveEncoding)> { + let compressor = strategy.create_block_compressor(field, &data)?; + let (payload, encoding) = compressor.compress(data)?; + let payload = payload.ok_or_else(|| { + Error::internal("Required block compressor selected a metadata-only codec".to_string()) + })?; + Ok((payload, encoding)) +} + fn try_bss_for_mini_block( data: &FixedWidthDataBlock, params: &CompressionFieldParams, @@ -269,7 +279,7 @@ fn try_rle_for_block_with_width( params: &CompressionFieldParams, run_length_width: RunLengthWidth, rle_payload_bytes: u128, -) -> Result, CompressiveEncoding)>> { +) -> Result>> { let bits = data.bits_per_value; if !matches!(bits, 8 | 16 | 32 | 64) { return Ok(None); @@ -305,18 +315,15 @@ fn try_rle_for_block_with_width( } } - let compressor = Box::new(RleEncoder::with_run_length_width(run_length_width)); - let encoding = ProtobufUtils21::rle( - ProtobufUtils21::flat(bits, None), - ProtobufUtils21::flat(run_length_width.bits_per_value(), None), - ); - Ok(Some((compressor, encoding))) + Ok(Some(Box::new(RleEncoder::with_run_length_width( + run_length_width, + )))) } fn try_fixed_u8_rle_for_block( data: &FixedWidthDataBlock, params: &CompressionFieldParams, -) -> Result, CompressiveEncoding)>> { +) -> Result>> { if !matches!(data.bits_per_value, 8 | 16 | 32 | 64) { return Ok(None); } @@ -327,7 +334,7 @@ fn try_fixed_u8_rle_for_block( fn try_variable_rle_for_block( data: &FixedWidthDataBlock, params: &CompressionFieldParams, -) -> Result, CompressiveEncoding)>> { +) -> Result>> { if !matches!(data.bits_per_value, 8 | 16 | 32 | 64) { return Ok(None); } @@ -410,9 +417,7 @@ fn estimate_inline_bitpacking_bytes(data: &FixedWidthDataBlock) -> Option { u64::try_from(estimated_bytes).ok() } -fn try_bitpack_for_block( - data: &FixedWidthDataBlock, -) -> Option<(Box, CompressiveEncoding)> { +fn try_bitpack_for_block(data: &FixedWidthDataBlock) -> Option> { let bits = data.bits_per_value; if !matches!(bits, 8 | 16 | 32 | 64) { return None; @@ -430,16 +435,9 @@ fn try_bitpack_for_block( } if data.num_values <= 1024 { - let compressor = Box::new(InlineBitpacking::new(bits)); - let encoding = ProtobufUtils21::inline_bitpacking(bits, None); - Some((compressor, encoding)) + Some(Box::new(InlineBitpacking::new(bits))) } else { - let compressor = Box::new(OutOfLineBitpacking::new(max_bit_width, bits)); - let encoding = ProtobufUtils21::out_of_line_bitpacking( - bits, - ProtobufUtils21::flat(max_bit_width, None), - ); - Some((compressor, encoding)) + Some(Box::new(OutOfLineBitpacking::new(max_bit_width, bits))) } } @@ -818,7 +816,7 @@ pub fn try_variable_width_per_value( pub fn try_fixed_u8_rle_block( data: &DataBlock, params: &CompressionFieldParams, -) -> Result, CompressiveEncoding)>> { +) -> Result>> { let DataBlock::FixedWidth(data) = data else { return Ok(None); }; @@ -829,7 +827,7 @@ pub fn try_fixed_u8_rle_block( pub fn try_variable_rle_block( data: &DataBlock, params: &CompressionFieldParams, -) -> Result, CompressiveEncoding)>> { +) -> Result>> { let DataBlock::FixedWidth(data) = data else { return Ok(None); }; @@ -837,9 +835,7 @@ pub fn try_variable_rle_block( } /// Select block bitpacking for applicable fixed-width values. -pub fn try_bitpacking_block( - data: &DataBlock, -) -> Option<(Box, CompressiveEncoding)> { +pub fn try_bitpacking_block(data: &DataBlock) -> Option> { let DataBlock::FixedWidth(data) = data else { return None; }; @@ -850,35 +846,22 @@ pub fn try_bitpacking_block( pub fn try_general_block( data: &DataBlock, params: &CompressionFieldParams, -) -> Result, CompressiveEncoding)>> { - let Some((compressor, config)) = try_general_compression(params, data)? else { +) -> Result>> { + let Some((compressor, _config)) = try_general_compression(params, data)? else { return Ok(None); }; - let inner = match data { - DataBlock::FixedWidth(data) => ProtobufUtils21::flat(data.bits_per_value, None), - DataBlock::VariableWidth(data) => ProtobufUtils21::variable( - ProtobufUtils21::flat(data.bits_per_offset as u64, None), - None, - ), - _ => return Ok(None), - }; - Ok(Some((compressor, ProtobufUtils21::wrapped(config, inner)?))) + Ok(Some(compressor)) } /// Store fixed- and variable-width block values without block compression. -pub fn try_raw_block(data: &DataBlock) -> Option<(Box, CompressiveEncoding)> { +pub fn try_raw_block(data: &DataBlock) -> Option> { match data { - DataBlock::FixedWidth(data) => Some(( - Box::new(ValueEncoder::default()) as Box, - ProtobufUtils21::flat(data.bits_per_value, None), - )), - DataBlock::VariableWidth(data) => Some(( - Box::new(VariableEncoder::default()) as Box, - ProtobufUtils21::variable( - ProtobufUtils21::flat(data.bits_per_offset as u64, None), - None, - ), - )), + DataBlock::FixedWidth(_) => { + Some(Box::new(ValueEncoder::default()) as Box) + } + DataBlock::VariableWidth(_) => { + Some(Box::new(VariableEncoder::default()) as Box) + } _ => None, } } @@ -922,7 +905,23 @@ pub trait VariablePerValueDecompressor: std::fmt::Debug + Send + Sync { } pub trait BlockDecompressor: std::fmt::Debug + Send + Sync { - fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result; + fn decompress(&self, data: Option, num_values: u64) -> Result; + + /// Whether this codec consumes one payload buffer. + fn requires_payload(&self) -> bool { + true + } +} + +pub(crate) fn require_block_payload(data: Option, codec: &str) -> Result { + data.ok_or_else(|| Error::invalid_input(format!("{codec} requires one payload"))) +} + +pub(crate) fn require_no_block_payload(data: Option, codec: &str) -> Result<()> { + if data.is_some() { + return Err(Error::invalid_input(format!("{codec} expects no payload"))); + } + Ok(()) } pub trait DecompressionStrategy: std::fmt::Debug + Send + Sync { @@ -1389,6 +1388,16 @@ mod tests { strategy(TestEncoding::StructuralU16, params) } + fn selected_block_codec( + strategy: &Arc, + field: &Field, + data: &DataBlock, + ) -> (Box, CompressiveEncoding) { + let compressor = strategy.create_block_compressor(field, data).unwrap(); + let (_, encoding) = compressor.compress(data.clone()).unwrap(); + (compressor, encoding) + } + fn miniblock_context() -> MiniBlockCompressionContext { MiniBlockCompressionContext::new(0, true, true) } @@ -1643,7 +1652,7 @@ mod tests { block.compute_stat(); let data = DataBlock::FixedWidth(block); - let (compressor, _encoding) = strategy.create_block_compressor(&field, &data).unwrap(); + let compressor = strategy.create_block_compressor(&field, &data).unwrap(); let debug_str = format!("{:?}", compressor); assert!( debug_str.contains("OutOfLineBitpacking"), @@ -1665,7 +1674,7 @@ mod tests { block.compute_stat(); let data = DataBlock::FixedWidth(block); - let (compressor, encoding) = strategy.create_block_compressor(&field, &data).unwrap(); + let (compressor, encoding) = selected_block_codec(&strategy, &field, &data); assert!(format!("{compressor:?}").contains("ValueEncoder")); assert!(matches!( @@ -1693,7 +1702,7 @@ mod tests { block.compute_stat(); let data = DataBlock::FixedWidth(block); - let (compressor, encoding) = strategy.create_block_compressor(&field, &data).unwrap(); + let (compressor, encoding) = selected_block_codec(&strategy, &field, &data); let debug_str = format!("{compressor:?}"); assert!( debug_str.contains("OutOfLineBitpacking"), @@ -2492,18 +2501,17 @@ mod tests { let expected_num_values = expected_block.num_values; let num_values = expected_num_values; - let (compressor, encoding) = strategy + let compressor = strategy .create_block_compressor(&field, &data) .expect("general compression should be selected"); + let (compressed_buffer, encoding) = compressor + .compress(data.clone()) + .expect("write path general compression should succeed"); match encoding.compression.as_ref() { Some(Compression::General(_)) => {} other => panic!("expected general compression, got {:?}", other), } - let compressed_buffer = compressor - .compress(data.clone()) - .expect("write path general compression should succeed"); - let decompressor = DefaultDecompressionStrategy::default() .create_block_decompressor(&encoding) .expect("general block decompressor should be created"); @@ -2538,9 +2546,10 @@ mod tests { let field = create_test_field("dict_values", DataType::FixedSizeBinary(3)); let data = create_fixed_width_block(24, 1024); - let (_compressor, encoding) = strategy + let compressor = strategy .create_block_compressor(&field, &data) .expect("block compressor selection should succeed"); + let (_, encoding) = compressor.compress(data).unwrap(); assert!( !matches!(encoding.compression.as_ref(), Some(Compression::General(_))), @@ -2568,9 +2577,10 @@ mod tests { "test requires block size above automatic general compression threshold" ); - let (_compressor, encoding) = strategy + let compressor = strategy .create_block_compressor(&field, &data) .expect("block compressor selection should succeed"); + let (_, encoding) = compressor.compress(data).unwrap(); assert!( !matches!(encoding.compression.as_ref(), Some(Compression::General(_))), @@ -2592,10 +2602,9 @@ mod tests { let data = DataBlock::FixedWidth(block); let strategy = strategy(TestEncoding::StructuralSparse, CompressionParams::new()); - let (compressor, encoding) = strategy.create_block_compressor(&field, &data).unwrap(); + let compressor = strategy.create_block_compressor(&field, &data).unwrap(); + let (compressed, encoding) = compressor.compress(data).unwrap(); assert_eq!(rle_run_length_bits(&encoding), 32); - - let compressed = compressor.compress(data).unwrap(); let decompressor = DefaultDecompressionStrategy::default() .create_block_decompressor(&encoding) .unwrap(); @@ -2626,7 +2635,8 @@ mod tests { let data = DataBlock::FixedWidth(block); let strategy = strategy(TestEncoding::StructuralU32, CompressionParams::new()); - let (_compressor, encoding) = strategy.create_block_compressor(&field, &data).unwrap(); + let compressor = strategy.create_block_compressor(&field, &data).unwrap(); + let (_, encoding) = compressor.compress(data).unwrap(); assert_eq!(rle_run_length_bits(&encoding), 8); } @@ -2656,7 +2666,7 @@ mod tests { let strategy = strategy(TestEncoding::StructuralU32, CompressionParams::new()); - let (compressor, _) = strategy + let compressor = strategy .create_block_compressor(&field, &data_block) .unwrap(); @@ -2690,7 +2700,7 @@ mod tests { let strategy = strategy(TestEncoding::StructuralU16, CompressionParams::new()); - let (compressor, _) = strategy + let compressor = strategy .create_block_compressor(&field, &data_block) .unwrap(); diff --git a/rust/lance-encoding/src/encodings/logical/primitive.rs b/rust/lance-encoding/src/encodings/logical/primitive.rs index 15b39d62b26..c26d9af6787 100644 --- a/rust/lance-encoding/src/encodings/logical/primitive.rs +++ b/rust/lance-encoding/src/encodings/logical/primitive.rs @@ -45,7 +45,7 @@ use crate::utils::bytepack::ByteUnpacker; use crate::{ compression::{ BlockDecompressor, CompressionStrategy, DecompressionStrategy, MiniBlockDecompressor, - create_rle_decompressor, + compress_required_block, create_rle_decompressor, }, data::{AllNullDataBlock, DataBlock, VariableWidthBlock}, utils::bytepack::BytepackedIntegerEncoder, @@ -192,7 +192,7 @@ impl DecodeMiniBlockTask { levels: LanceBuffer, num_levels: u16, ) -> Result> { - let rep = rep_decompressor.decompress(levels, num_levels as u64)?; + let rep = rep_decompressor.decompress(Some(levels), num_levels as u64)?; let rep = rep.as_fixed_width().unwrap(); debug_assert_eq!(rep.num_values, num_levels as u64); debug_assert_eq!(rep.bits_per_value, 16); @@ -1569,7 +1569,7 @@ impl StructuralPageScheduler for ComplexAllNullScheduler { } LevelCodec::Block(decompressor) => { let frame = LanceBuffer::from_bytes(compressed_bytes, 1); - let decompressed = decompressor.decompress(frame, num_values)?; + let decompressed = decompressor.decompress(Some(frame), num_values)?; dense_levels_from_block(decompressed, num_values, level_type) } } @@ -2624,7 +2624,10 @@ impl StructuralPageScheduler for MiniBlockScheduler { let dictionary = if let Some(ref mut dictionary) = self.dictionary { let dictionary_data = dictionary_bytes.unwrap(); Some(Arc::new(dictionary.dictionary_decompressor.decompress( - LanceBuffer::from_bytes(dictionary_data, dictionary.dictionary_data_alignment), + Some(LanceBuffer::from_bytes( + dictionary_data, + dictionary.dictionary_data_alignment, + )), dictionary.num_dictionary_items, )?)) } else { @@ -5020,8 +5023,9 @@ impl PrimitiveStructuralEncoder { let levels_block = DataBlock::FixedWidth(fixed_width_block); let levels_field = Field::new_arrow("", DataType::UInt16, false)?; // Pick a block compressor - let (compressor, compressor_desc) = + let compressor = compression_strategy.create_block_compressor(&levels_field, &levels_block)?; + let mut compressor_desc = None; // Compress blocks of levels (sized according to the chunks) let mut level_chunks = Vec::with_capacity(chunks.len()); let mut values_counter = 0; @@ -5091,7 +5095,22 @@ impl PrimitiveStructuralEncoder { }; chunk_fixed_width.compute_stat(); let chunk_levels_block = DataBlock::FixedWidth(chunk_fixed_width); - let compressed_levels = compressor.compress(chunk_levels_block)?; + let (compressed_levels, chunk_compressor_desc) = + compressor.compress(chunk_levels_block)?; + if let Some(compressor_desc) = compressor_desc.as_ref() { + if compressor_desc != &chunk_compressor_desc { + return Err(Error::internal( + "Rep/def block compressor changed encoding between chunks".to_string(), + )); + } + } else { + compressor_desc = Some(chunk_compressor_desc); + } + let compressed_levels = compressed_levels.ok_or_else(|| { + Error::internal( + "Rep/def block compressor selected a metadata-only codec".to_string(), + ) + })?; let num_levels = u16::try_from(num_chunk_levels).map_err(|_| { Error::invalid_input_source( format!( @@ -5116,7 +5135,9 @@ impl PrimitiveStructuralEncoder { }; Ok(CompressedLevels { data: level_chunks, - compression: compressor_desc, + compression: compressor_desc.ok_or_else(|| { + Error::internal("Rep/def compression produced no chunks".to_string()) + })?, rep_index, }) } @@ -5152,9 +5173,8 @@ impl PrimitiveStructuralEncoder { let levels_block = DataBlock::FixedWidth(fixed_width_block); let levels_field = Field::new_arrow("", DataType::UInt16, false)?; - let (compressor, encoding) = - compression_strategy.create_block_compressor(&levels_field, &levels_block)?; - let compressed_buffer = compressor.compress(levels_block)?; + let (compressed_buffer, encoding) = + compress_required_block(compression_strategy, &levels_field, levels_block)?; Ok((compressed_buffer, encoding)) } @@ -5540,9 +5560,8 @@ impl PrimitiveStructuralEncoder { let num_dictionary_items = dictionary_data.num_values(); let dict_values_field = Self::build_dict_values_compressor_field(field)?; - let (compressor, dictionary_encoding) = compression_strategy - .create_block_compressor(&dict_values_field, &dictionary_data)?; - let dictionary_buffer = compressor.compress(dictionary_data)?; + let (dictionary_buffer, dictionary_encoding) = + compress_required_block(compression_strategy, &dict_values_field, dictionary_data)?; data.push(dictionary_buffer); if let Some(rep_index) = rep_index { @@ -9648,7 +9667,7 @@ mod tests { .create_block_decompressor(&encoding) .unwrap(); let decompressed = decompressor - .decompress(compressed_buf, values.len() as u64) + .decompress(Some(compressed_buf), values.len() as u64) .unwrap(); let decompressed_fixed_width = decompressed.as_fixed_width().unwrap(); assert_eq!(decompressed_fixed_width.num_values, values.len() as u64); @@ -9737,6 +9756,8 @@ mod tests { }); BlockCompressor::compress(&RleEncoder::with_run_length_width(run_length_width), block) .unwrap() + .0 + .unwrap() } fn encoded_u16_runs(levels: &[u16], run_length_width: RunLengthWidth) -> RleRuns { diff --git a/rust/lance-encoding/src/encodings/logical/primitive/sparse.rs b/rust/lance-encoding/src/encodings/logical/primitive/sparse.rs index 25cb85f49ed..46bb36c808c 100644 --- a/rust/lance-encoding/src/encodings/logical/primitive/sparse.rs +++ b/rust/lance-encoding/src/encodings/logical/primitive/sparse.rs @@ -2645,7 +2645,7 @@ impl SparseStructuralScheduler { ) -> Result> { Self::validate_structural_buffer_headers(encoding, &data, label)?; let decoded = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - decompressor.decompress(LanceBuffer::from_bytes(data, 1), num_values) + decompressor.decompress(Some(LanceBuffer::from_bytes(data, 1)), num_values) })) .map_err(|_| { Error::invalid_input_source( diff --git a/rust/lance-encoding/src/encodings/logical/primitive/sparse/writer.rs b/rust/lance-encoding/src/encodings/logical/primitive/sparse/writer.rs index ad910c6d8f9..5ec1d32b5e1 100644 --- a/rust/lance-encoding/src/encodings/logical/primitive/sparse/writer.rs +++ b/rust/lance-encoding/src/encodings/logical/primitive/sparse/writer.rs @@ -10,7 +10,7 @@ use lance_core::{Error, Result, datatypes::Field, utils::bit::pad_bytes}; use crate::{ buffer::LanceBuffer, - compression::CompressionStrategy, + compression::{CompressionStrategy, compress_required_block}, data::{BlockInfo, DataBlock, FixedWidthDataBlock}, decoder::PageEncoding, encoder::EncodedPage, @@ -593,8 +593,7 @@ fn encode_u64_values( }); block.compute_stat(); let field = Field::new_arrow("", arrow_schema::DataType::UInt64, false)?; - let (compressor, encoding) = compression_strategy.create_block_compressor(&field, &block)?; - Ok((compressor.compress(block)?, encoding)) + compress_required_block(compression_strategy, &field, block) } fn positions_to_deltas(positions: &[u64], label: &str) -> Result> { diff --git a/rust/lance-encoding/src/encodings/physical/binary.rs b/rust/lance-encoding/src/encodings/physical/binary.rs index 06df60434e9..a3e8487b221 100644 --- a/rust/lance-encoding/src/encodings/physical/binary.rs +++ b/rust/lance-encoding/src/encodings/physical/binary.rs @@ -15,6 +15,7 @@ use core::panic; use crate::compression::{ BlockCompressor, BlockDecompressor, MiniBlockDecompressor, VariablePerValueDecompressor, + require_block_payload, }; use crate::buffer::LanceBuffer; @@ -453,7 +454,15 @@ impl MiniBlockDecompressor for BinaryMiniBlockDecompressor { pub struct VariableEncoder {} impl BlockCompressor for VariableEncoder { - fn compress(&self, mut data: DataBlock) -> Result { + fn compress(&self, mut data: DataBlock) -> Result<(Option, CompressiveEncoding)> { + let bits_per_offset = match &data { + DataBlock::VariableWidth(data) => data.bits_per_offset, + _ => { + return Err(Error::invalid_input( + "BinaryBlockEncoder requires a variable-width block", + )); + } + }; match data { DataBlock::VariableWidth(ref mut variable_width_data) => { match variable_width_data.bits_per_offset { @@ -505,18 +514,23 @@ impl BlockCompressor for VariableEncoder { output.extend_from_slice(&variable_width_data.data); Ok(LanceBuffer::from(output)) } - _ => { - panic!( - "BinaryBlockEncoder does not work with {} bits per offset VariableWidth DataBlock.", - variable_width_data.bits_per_offset - ); - } + _ => Err(Error::invalid_input(format!( + "BinaryBlockEncoder does not support {}-bit offsets", + variable_width_data.bits_per_offset + ))), } } - _ => { - panic!("BinaryBlockEncoder can only work with Variable Width DataBlock."); - } + _ => unreachable!("variable-width input was validated above"), } + .map(|payload| { + ( + Some(payload), + ProtobufUtils21::variable( + ProtobufUtils21::flat(bits_per_offset as u64, None), + None, + ), + ) + }) } } @@ -547,7 +561,8 @@ impl VariablePerValueDecompressor for VariableDecoder { pub struct BinaryBlockDecompressor {} impl BlockDecompressor for BinaryBlockDecompressor { - fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result { + fn decompress(&self, data: Option, num_values: u64) -> Result { + let data = require_block_payload(data, "Binary block")?; // In older (not quite stable) versions we stored the bits per offset as a single byte and then the num_values // as four bytes. However, this led to alignment problems and was wasteful since we already store the num_values // in higher layers. @@ -1251,7 +1266,9 @@ mod tests { }); BlockCompressor::compress(&super::VariableEncoder::default(), block) .unwrap() + .0 .as_ref() + .unwrap() .to_vec() } @@ -1280,7 +1297,7 @@ mod tests { .copy_from_slice(&mutated_offset_value.to_le_bytes()[..bytes_per_offset]); let block = super::BinaryBlockDecompressor::default() - .decompress(LanceBuffer::from(encoded), 3) + .decompress(Some(LanceBuffer::from(encoded)), 3) .unwrap(); let data_type = match bits_per_offset { 32 => DataType::Binary, @@ -1302,7 +1319,7 @@ mod tests { let mut encoded = encoded_binary_block(32); encoded[8..12].copy_from_slice(&5_u32.to_le_bytes()); let err = decompressor - .decompress(LanceBuffer::from(encoded), 3) + .decompress(Some(LanceBuffer::from(encoded)), 3) .unwrap_err(); assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}"); assert!(err.to_string().contains("first offset"), "{err}"); @@ -1310,14 +1327,14 @@ mod tests { // The offsets region must hold exactly num_values + 1 offsets. let encoded = encoded_binary_block(32); let err = decompressor - .decompress(LanceBuffer::from(encoded), 4) + .decompress(Some(LanceBuffer::from(encoded)), 4) .unwrap_err(); assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}"); assert!(err.to_string().contains("offset bytes"), "{err}"); // A block too small to hold its header is rejected, not a panic. let err = decompressor - .decompress(LanceBuffer::from(vec![0_u8; 2]), 1) + .decompress(Some(LanceBuffer::from(vec![0_u8; 2])), 1) .unwrap_err(); assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}"); assert!(err.to_string().contains("too small"), "{err}"); diff --git a/rust/lance-encoding/src/encodings/physical/bitpacking.rs b/rust/lance-encoding/src/encodings/physical/bitpacking.rs index 9b775911005..72c7ba6c7b4 100644 --- a/rust/lance-encoding/src/encodings/physical/bitpacking.rs +++ b/rust/lance-encoding/src/encodings/physical/bitpacking.rs @@ -23,7 +23,9 @@ use lance_bitpacking::BitPacking; use lance_core::{Error, Result}; use crate::buffer::LanceBuffer; -use crate::compression::{BlockCompressor, BlockDecompressor, MiniBlockDecompressor}; +use crate::compression::{ + BlockCompressor, BlockDecompressor, MiniBlockDecompressor, require_block_payload, +}; use crate::data::BlockInfo; use crate::data::{DataBlock, FixedWidthDataBlock}; use crate::encodings::logical::primitive::miniblock::{ @@ -303,10 +305,27 @@ impl MiniBlockCompressor for InlineBitpacking { } impl BlockCompressor for InlineBitpacking { - fn compress(&self, data: DataBlock) -> Result { - let fixed_width = data.as_fixed_width().unwrap(); + fn compress(&self, data: DataBlock) -> Result<(Option, CompressiveEncoding)> { + let DataBlock::FixedWidth(fixed_width) = data else { + return Err(Error::invalid_input( + "Inline bitpacking requires fixed-width data", + )); + }; + if fixed_width.bits_per_value != self.uncompressed_bit_width { + return Err(Error::invalid_input(format!( + "Inline bitpacking expects {}-bit values, got {}", + self.uncompressed_bit_width, fixed_width.bits_per_value + ))); + } let (chunked, _) = self.chunk_data(fixed_width); - Ok(chunked.data.into_iter().next().unwrap()) + let payload = + chunked.data.into_iter().next().ok_or_else(|| { + Error::internal("Inline bitpacking produced no payload".to_string()) + })?; + Ok(( + Some(payload), + ProtobufUtils21::inline_bitpacking(self.uncompressed_bit_width, None), + )) } } @@ -335,7 +354,8 @@ impl MiniBlockDecompressor for InlineBitpacking { } impl BlockDecompressor for InlineBitpacking { - fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result { + fn decompress(&self, data: Option, num_values: u64) -> Result { + let data = require_block_payload(data, "Inline bitpacking")?; if num_values == 0 { // Empty blocks carry no inline bit-width header to decode; avoid // spurious "too small for header" corrupt-file errors and mirror @@ -553,21 +573,43 @@ impl OutOfLineBitpacking { } impl BlockCompressor for OutOfLineBitpacking { - fn compress(&self, data: DataBlock) -> Result { - let fixed_width = data.as_fixed_width().unwrap(); + fn compress(&self, data: DataBlock) -> Result<(Option, CompressiveEncoding)> { + let DataBlock::FixedWidth(fixed_width) = data else { + return Err(Error::invalid_input( + "Out-of-line bitpacking requires fixed-width data", + )); + }; + if fixed_width.bits_per_value != self.uncompressed_bit_width { + return Err(Error::invalid_input(format!( + "Out-of-line bitpacking expects {}-bit values, got {}", + self.uncompressed_bit_width, fixed_width.bits_per_value + ))); + } let compressed = match fixed_width.bits_per_value { 8 => bitpack_out_of_line::(fixed_width, self.compressed_bit_width as usize), 16 => bitpack_out_of_line::(fixed_width, self.compressed_bit_width as usize), 32 => bitpack_out_of_line::(fixed_width, self.compressed_bit_width as usize), 64 => bitpack_out_of_line::(fixed_width, self.compressed_bit_width as usize), - _ => panic!("Bitpacking word size must be 8,16,32,64"), + _ => { + return Err(Error::invalid_input(format!( + "Bitpacking word size must be 8, 16, 32, or 64, got {}", + fixed_width.bits_per_value + ))); + } }; - Ok(compressed) + Ok(( + Some(compressed), + ProtobufUtils21::out_of_line_bitpacking( + self.uncompressed_bit_width, + ProtobufUtils21::flat(self.compressed_bit_width, None), + ), + )) } } impl BlockDecompressor for OutOfLineBitpacking { - fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result { + fn decompress(&self, data: Option, num_values: u64) -> Result { + let data = require_block_payload(data, "Out-of-line bitpacking")?; let word_size = match self.uncompressed_bit_width { 8 => std::mem::size_of::(), 16 => std::mem::size_of::(), @@ -660,7 +702,7 @@ mod test { fn test_inline_bitpacking_decompress_empty_block(#[case] bit_width: u64) { let decompressor = InlineBitpacking::new(bit_width); let decompressed = - BlockDecompressor::decompress(&decompressor, LanceBuffer::empty(), 0).unwrap(); + BlockDecompressor::decompress(&decompressor, Some(LanceBuffer::empty()), 0).unwrap(); let DataBlock::FixedWidth(block) = decompressed else { panic!("Expected FixedWidth block"); diff --git a/rust/lance-encoding/src/encodings/physical/block.rs b/rust/lance-encoding/src/encodings/physical/block.rs index 188af50c8a0..a645988a147 100644 --- a/rust/lance-encoding/src/encodings/physical/block.rs +++ b/rust/lance-encoding/src/encodings/physical/block.rs @@ -26,7 +26,7 @@ use lance_core::{Error, Result}; use std::str::FromStr; -use crate::compression::{BlockCompressor, BlockDecompressor}; +use crate::compression::{BlockCompressor, BlockDecompressor, require_block_payload}; use crate::encodings::physical::binary::{BinaryBlockDecompressor, VariableEncoder}; use crate::format::{ ProtobufUtils21, @@ -450,11 +450,12 @@ impl GeneralBlockDecompressor { } impl BlockDecompressor for GeneralBlockDecompressor { - fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result { + fn decompress(&self, data: Option, num_values: u64) -> Result { + let data = require_block_payload(data, "General block compression")?; let mut decompressed = Vec::new(); self.compressor.decompress(&data, &mut decompressed)?; self.inner - .decompress(LanceBuffer::from(decompressed), num_values) + .decompress(Some(LanceBuffer::from(decompressed)), num_values) } } @@ -614,13 +615,26 @@ impl VariablePerValueDecompressor for CompressedBufferEncoder { } impl BlockCompressor for CompressedBufferEncoder { - fn compress(&self, data: DataBlock) -> Result { - let encoded = match data { - DataBlock::FixedWidth(fixed_width) => fixed_width.data, + fn compress(&self, data: DataBlock) -> Result<(Option, CompressiveEncoding)> { + let (encoded, inner_encoding) = match data { + DataBlock::FixedWidth(fixed_width) => ( + fixed_width.data, + ProtobufUtils21::flat(fixed_width.bits_per_value, None), + ), DataBlock::VariableWidth(variable_width) => { // Wrap VariableEncoder to handle the encoding let encoder = VariableEncoder::default(); - BlockCompressor::compress(&encoder, DataBlock::VariableWidth(variable_width))? + let (payload, encoding) = + BlockCompressor::compress(&encoder, DataBlock::VariableWidth(variable_width))?; + ( + payload.ok_or_else(|| { + Error::internal( + "VariableEncoder returned no payload for general compression" + .to_string(), + ) + })?, + encoding, + ) } _ => { return Err(Error::invalid_input_source( @@ -631,18 +645,22 @@ impl BlockCompressor for CompressedBufferEncoder { let mut compressed = Vec::new(); self.compressor.compress(&encoded, &mut compressed)?; - Ok(LanceBuffer::from(compressed)) + Ok(( + Some(LanceBuffer::from(compressed)), + ProtobufUtils21::wrapped(self.compressor.config(), inner_encoding)?, + )) } } impl BlockDecompressor for CompressedBufferEncoder { - fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result { + fn decompress(&self, data: Option, num_values: u64) -> Result { + let data = require_block_payload(data, "Compressed variable block")?; let mut decompressed = Vec::new(); self.compressor.decompress(&data, &mut decompressed)?; // Delegate to BinaryBlockDecompressor which handles the inline metadata let inner_decoder = BinaryBlockDecompressor::default(); - inner_decoder.decompress(LanceBuffer::from(decompressed), num_values) + inner_decoder.decompress(Some(LanceBuffer::from(decompressed)), num_values) } } diff --git a/rust/lance-encoding/src/encodings/physical/constant.rs b/rust/lance-encoding/src/encodings/physical/constant.rs index c3fa16863f4..dd153f789d5 100644 --- a/rust/lance-encoding/src/encodings/physical/constant.rs +++ b/rust/lance-encoding/src/encodings/physical/constant.rs @@ -5,7 +5,7 @@ use crate::{ buffer::LanceBuffer, - compression::{BlockDecompressor, FixedPerValueDecompressor}, + compression::{BlockDecompressor, FixedPerValueDecompressor, require_no_block_payload}, data::{AllNullDataBlock, ConstantDataBlock, DataBlock, FixedWidthDataBlock}, }; @@ -24,7 +24,8 @@ impl ConstantDecompressor { } impl BlockDecompressor for ConstantDecompressor { - fn decompress(&self, _data: LanceBuffer, num_values: u64) -> Result { + fn decompress(&self, data: Option, num_values: u64) -> Result { + require_no_block_payload(data, "Constant")?; if let Some(scalar) = self.scalar.clone() { Ok(DataBlock::Constant(ConstantDataBlock { data: scalar, @@ -34,6 +35,10 @@ impl BlockDecompressor for ConstantDecompressor { Ok(DataBlock::AllNull(AllNullDataBlock { num_values })) } } + + fn requires_payload(&self) -> bool { + false + } } impl FixedPerValueDecompressor for ConstantDecompressor { @@ -55,3 +60,22 @@ impl FixedPerValueDecompressor for ConstantDecompressor { .unwrap_or(0) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn block_constant_requires_no_payload() { + let decompressor = ConstantDecompressor::new(None); + + assert!(!decompressor.requires_payload()); + assert!(matches!( + BlockDecompressor::decompress(&decompressor, None, 3).unwrap(), + DataBlock::AllNull(AllNullDataBlock { num_values: 3 }) + )); + assert!( + BlockDecompressor::decompress(&decompressor, Some(LanceBuffer::empty()), 3).is_err() + ); + } +} diff --git a/rust/lance-encoding/src/encodings/physical/rle.rs b/rust/lance-encoding/src/encodings/physical/rle.rs index 0f3e330a3ba..0215e392584 100644 --- a/rust/lance-encoding/src/encodings/physical/rle.rs +++ b/rust/lance-encoding/src/encodings/physical/rle.rs @@ -58,7 +58,9 @@ use arrow_buffer::{ArrowNativeType, ScalarBuffer}; use log::trace; use crate::buffer::LanceBuffer; -use crate::compression::{BlockCompressor, BlockDecompressor, MiniBlockDecompressor}; +use crate::compression::{ + BlockCompressor, BlockDecompressor, MiniBlockDecompressor, require_block_payload, +}; use crate::data::DataBlock; use crate::data::{BlockInfo, FixedWidthDataBlock}; use crate::encodings::logical::primitive::miniblock::{ @@ -872,7 +874,10 @@ impl RleEncoder { num_values: child_values, block_info: BlockInfo::default(), }); - let chunk_packed = BlockCompressor::compress(&compressor, block)?; + let (chunk_packed, _) = BlockCompressor::compress(&compressor, block)?; + let chunk_packed = chunk_packed.ok_or_else(|| { + Error::internal("RLE bitpacking child returned no payload".to_string()) + })?; let packed_size = u32::try_from(chunk_packed.len()).map_err(|_| { Error::invalid_input_source( format!( @@ -1107,7 +1112,7 @@ impl MiniBlockCompressor for RleEncoder { impl BlockCompressor for RleEncoder { // Block format: [8-byte header: values buffer size][values buffer][run_lengths buffer] - fn compress(&self, data: DataBlock) -> Result { + fn compress(&self, data: DataBlock) -> Result<(Option, CompressiveEncoding)> { match data { DataBlock::FixedWidth(fixed_width) => { let num_values = fixed_width.num_values; @@ -1122,7 +1127,13 @@ impl BlockCompressor for RleEncoder { combined.extend_from_slice(&values_size.to_le_bytes()); combined.extend_from_slice(&all_buffers[0]); combined.extend_from_slice(&all_buffers[1]); - Ok(LanceBuffer::from(combined)) + Ok(( + Some(LanceBuffer::from(combined)), + ProtobufUtils21::rle( + ProtobufUtils21::flat(bits_per_value, None), + ProtobufUtils21::flat(self.run_length_width.bits_per_value(), None), + ), + )) } _ => Err(Error::invalid_input_source( "RLE encoding only supports FixedWidth data blocks".into(), @@ -1216,7 +1227,7 @@ impl RleChildDecompressor { } else { num_values.unwrap_or(0) }; - let decoded = decompressor.decompress(data, num_values)?; + let decoded = decompressor.decompress(Some(data), num_values)?; self.extract_fixed_width(decoded, num_values, label) } } @@ -1565,7 +1576,8 @@ impl MiniBlockDecompressor for RleDecompressor { } impl BlockDecompressor for RleDecompressor { - fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result { + fn decompress(&self, data: Option, num_values: u64) -> Result { + let data = require_block_payload(data, "RLE")?; let (values_buffer, lengths_buffer) = parse_rle_block_frame(&data)?; self.decode_data(vec![values_buffer, lengths_buffer], num_values, false) } @@ -1892,11 +1904,17 @@ mod tests { num_values, block_info: BlockInfo::new(), }); - let frame = BlockCompressor::compress(&RleEncoder::new(), block).unwrap(); + let frame = BlockCompressor::compress(&RleEncoder::new(), block) + .unwrap() + .0 + .unwrap(); - let eager = - BlockDecompressor::decompress(&RleDecompressor::new(16), frame.clone(), num_values) - .unwrap(); + let eager = BlockDecompressor::decompress( + &RleDecompressor::new(16), + Some(frame.clone()), + num_values, + ) + .unwrap(); let DataBlock::FixedWidth(eager) = eager else { panic!("expected fixed-width block"); }; @@ -1969,7 +1987,10 @@ mod tests { num_values, block_info: BlockInfo::new(), }); - let frame = BlockCompressor::compress(&RleEncoder::new(), block).unwrap(); + let frame = BlockCompressor::compress(&RleEncoder::new(), block) + .unwrap() + .0 + .unwrap(); let (values, lengths) = parse_rle_block_frame(&frame).unwrap(); let compression = test_general_compression(); @@ -2011,7 +2032,10 @@ mod tests { num_values, block_info: BlockInfo::new(), }); - let frame = BlockCompressor::compress(&RleEncoder::new(), block).unwrap(); + let frame = BlockCompressor::compress(&RleEncoder::new(), block) + .unwrap() + .0 + .unwrap(); let runs = RleDecompressor::new(16) .decode_u16_runs(frame, num_values) .unwrap(); @@ -2032,7 +2056,10 @@ mod tests { num_values: n, block_info: BlockInfo::new(), }); - let frame = BlockCompressor::compress(&RleEncoder::new(), block).unwrap(); + let frame = BlockCompressor::compress(&RleEncoder::new(), block) + .unwrap() + .0 + .unwrap(); let runs = RleDecompressor::new(16).decode_u16_runs(frame, n).unwrap(); assert_eq!( runs.coalesced_runs() as u64, @@ -2211,7 +2238,10 @@ mod tests { block_info: BlockInfo::default(), }); let bitpacked_run_lengths = - BlockCompressor::compress(&OutOfLineBitpacking::new(3, 8), run_lengths_block).unwrap(); + BlockCompressor::compress(&OutOfLineBitpacking::new(3, 8), run_lengths_block) + .unwrap() + .0 + .unwrap(); let encoding = ProtobufUtils21::rle( ProtobufUtils21::flat(32, None), ProtobufUtils21::out_of_line_bitpacking(8, ProtobufUtils21::flat(3, None)), @@ -2699,8 +2729,9 @@ mod tests { payload.extend_from_slice(&values); payload.extend_from_slice(&lengths); - let error = BlockDecompressor::decompress(&decompressor, LanceBuffer::from(payload), 5) - .unwrap_err(); + let error = + BlockDecompressor::decompress(&decompressor, Some(LanceBuffer::from(payload)), 5) + .unwrap_err(); assert!(matches!(&error, Error::InvalidInput { .. })); assert!( error @@ -3151,7 +3182,7 @@ mod tests { let mut data = Vec::new(); data.extend_from_slice(&u64::MAX.to_le_bytes()); - let result = BlockDecompressor::decompress(&decompressor, LanceBuffer::from(data), 1); + let result = BlockDecompressor::decompress(&decompressor, Some(LanceBuffer::from(data)), 1); assert!(result.is_err()); assert!( result @@ -3164,8 +3195,11 @@ mod tests { #[test] fn test_block_decompressor_too_small() { let decompressor = RleDecompressor::new(32); - let result = - BlockDecompressor::decompress(&decompressor, LanceBuffer::from(vec![1, 2, 3]), 10); + let result = BlockDecompressor::decompress( + &decompressor, + Some(LanceBuffer::from(vec![1, 2, 3])), + 10, + ); assert!(result.is_err()); assert!( result @@ -3181,7 +3215,10 @@ mod tests { let data = vec![1i32, 1, 1]; let array = Int32Array::from(data); - let compressed = BlockCompressor::compress(&encoder, DataBlock::from_array(array)).unwrap(); + let compressed = BlockCompressor::compress(&encoder, DataBlock::from_array(array)) + .unwrap() + .0 + .unwrap(); // Verify header format: first 8 bytes should be values_size as u64 assert!(compressed.len() >= 8); @@ -3205,7 +3242,7 @@ mod tests { let array = Int32Array::from(data.clone()); let data_block = DataBlock::from_array(array); - let compressed = BlockCompressor::compress(&encoder, data_block).unwrap(); + let compressed = BlockCompressor::compress(&encoder, data_block).unwrap().0; let decompressed = BlockDecompressor::decompress(&decompressor, compressed, data.len() as u64).unwrap(); @@ -3234,7 +3271,9 @@ mod tests { assert_eq!(total_values, 10000); let array = Int32Array::from(data.clone()); - let compressed = BlockCompressor::compress(&encoder, DataBlock::from_array(array)).unwrap(); + let compressed = BlockCompressor::compress(&encoder, DataBlock::from_array(array)) + .unwrap() + .0; let decompressed = BlockDecompressor::decompress(&decompressor, compressed, total_values as u64).unwrap(); diff --git a/rust/lance-encoding/src/encodings/physical/value.rs b/rust/lance-encoding/src/encodings/physical/value.rs index 0284e9784c6..e9794c067c6 100644 --- a/rust/lance-encoding/src/encodings/physical/value.rs +++ b/rust/lance-encoding/src/encodings/physical/value.rs @@ -6,6 +6,7 @@ use arrow_buffer::{BooleanBufferBuilder, bit_util}; use crate::buffer::LanceBuffer; use crate::compression::{ BlockCompressor, BlockDecompressor, FixedPerValueDecompressor, MiniBlockDecompressor, + require_block_payload, }; use crate::data::{ BlockInfo, DataBlock, FixedSizeListBlock, FixedWidthDataBlock, NullableDataBlock, @@ -458,15 +459,17 @@ impl ValueEncoder { } impl BlockCompressor for ValueEncoder { - fn compress(&self, data: DataBlock) -> Result { - let data = match data { - DataBlock::FixedWidth(fixed_width) => fixed_width.data, - _ => unimplemented!( - "Cannot compress block of type {} with ValueEncoder", + fn compress(&self, data: DataBlock) -> Result<(Option, CompressiveEncoding)> { + let DataBlock::FixedWidth(fixed_width) = data else { + return Err(Error::invalid_input(format!( + "ValueEncoder cannot compress a {} block", data.name() - ), + ))); }; - Ok(data) + Ok(( + Some(fixed_width.data), + ProtobufUtils21::flat(fixed_width.bits_per_value, None), + )) } } @@ -575,7 +578,8 @@ impl ValueDecompressor { } impl BlockDecompressor for ValueDecompressor { - fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result { + fn decompress(&self, data: Option, num_values: u64) -> Result { + let data = require_block_payload(data, "Flat block")?; let block = self.buffer_to_block(data, num_values); assert_eq!(block.num_values(), num_values); Ok(block) diff --git a/rust/lance-encoding/src/testing.rs b/rust/lance-encoding/src/testing.rs index 90c5c0bb61b..0cdf85d0300 100644 --- a/rust/lance-encoding/src/testing.rs +++ b/rust/lance-encoding/src/testing.rs @@ -195,7 +195,7 @@ impl CompressionStrategy for TestCompressionStrategy { &self, field: &LanceField, data: &DataBlock, - ) -> Result<(Box, CompressiveEncoding)> { + ) -> Result> { let params = self.field_params(field); let rle = match self.encoding { TestEncoding::Array | TestEncoding::StructuralU16 => None, diff --git a/rust/lance-file/src/versions/v2_1/compression.rs b/rust/lance-file/src/versions/v2_1/compression.rs index 5be59bf8e53..2ef40ed0241 100644 --- a/rust/lance-file/src/versions/v2_1/compression.rs +++ b/rust/lance-file/src/versions/v2_1/compression.rs @@ -14,7 +14,6 @@ use lance_encoding::{ compression_config::{CompressionFieldParams, CompressionParams}, data::DataBlock, encodings::logical::primitive::{fullzip::PerValueCompressor, miniblock::MiniBlockCompressor}, - format::pb21::CompressiveEncoding, }; #[derive(Debug, Clone)] @@ -111,7 +110,7 @@ impl CompressionStrategy for Strategy { &self, field: &Field, data: &DataBlock, - ) -> Result<(Box, CompressiveEncoding)> { + ) -> Result> { let _params = self.field_params(field); if let Some(compressor) = try_bitpacking_block(data) { return Ok(compressor); diff --git a/rust/lance-file/src/versions/v2_2/compression.rs b/rust/lance-file/src/versions/v2_2/compression.rs index a965d5afc64..aa8e3b64958 100644 --- a/rust/lance-file/src/versions/v2_2/compression.rs +++ b/rust/lance-file/src/versions/v2_2/compression.rs @@ -17,7 +17,6 @@ use lance_encoding::{ compression_config::{CompressionFieldParams, CompressionParams}, data::DataBlock, encodings::logical::primitive::{fullzip::PerValueCompressor, miniblock::MiniBlockCompressor}, - format::pb21::CompressiveEncoding, }; #[derive(Debug, Clone)] @@ -105,7 +104,7 @@ impl CompressionStrategy for Strategy { &self, field: &Field, data: &DataBlock, - ) -> Result<(Box, CompressiveEncoding)> { + ) -> Result> { let params = self.field_params(field); if let Some(compressor) = try_fixed_u8_rle_block(data, ¶ms)? { return Ok(compressor); diff --git a/rust/lance-file/src/versions/v2_3/compression.rs b/rust/lance-file/src/versions/v2_3/compression.rs index 731e4a8c084..f671877a44d 100644 --- a/rust/lance-file/src/versions/v2_3/compression.rs +++ b/rust/lance-file/src/versions/v2_3/compression.rs @@ -17,7 +17,6 @@ use lance_encoding::{ compression_config::{CompressionFieldParams, CompressionParams}, data::DataBlock, encodings::logical::primitive::{fullzip::PerValueCompressor, miniblock::MiniBlockCompressor}, - format::pb21::CompressiveEncoding, }; #[derive(Debug, Clone)] @@ -105,7 +104,7 @@ impl CompressionStrategy for Strategy { &self, field: &Field, data: &DataBlock, - ) -> Result<(Box, CompressiveEncoding)> { + ) -> Result> { let params = self.field_params(field); if let Some(compressor) = try_variable_rle_block(data, ¶ms)? { return Ok(compressor);