Add the path for scheduling variable-length full-zip data

Author: westonpace

protos/encodings.proto

@@ -230,15 +230,8 @@ message Binary {
   uint64 null_adjustment = 3;
 }
 
-message BinaryMiniBlock {
-}
-
-message BinaryBlock {
-}
-
-message FsstMiniBlock {
-  ArrayEncoding BinaryMiniBlock = 1;
-  bytes symbol_table = 2;
+message Variable {
+  uint32 bits_per_offset = 1;
 }
 
 message Fsst {
@@ -285,10 +278,8 @@ message ArrayEncoding {
         BitpackedForNonNeg bitpacked_for_non_neg = 12;
         Constant constant = 13;
         Bitpack2 bitpack2 = 14;
-        BinaryMiniBlock binary_mini_block = 15;
-        FsstMiniBlock fsst_mini_block = 16;
-        BinaryBlock binary_block = 17;
-        PackedStructFixedWidthMiniBlock packed_struct_fixed_width_mini_block = 18;
+        Variable variable = 15;
+        PackedStructFixedWidthMiniBlock packed_struct_fixed_width_mini_block = 16;
     }
 }
 
@@ -316,6 +307,34 @@ message ColumnEncoding {
   }
 }
 
+// # Standardized Interpretation of Counting Terms
+//
+// When working with 2.1 encodings we have a number of different "counting terms" and it can be
+// difficult to understand what we mean when we are talking about a "number of values".  Here is
+// a standard interpretation of these terms:
+//
+// TODO: This is a newly added standardization and hasn't yet been applied to all code.
+//
+// To understand these definitions consider a data type FIXED_SIZE_LIST<LIST<INT32>>.
+//
+// A "value" is an abstract term when we aren't being specific.
+//
+// - num_rows: This is the highest level counting term.  A single row includes everything in the
+//             fixed size list.  This is what the user asks for when they asks for a range of rows.
+// - num_elements: The number of elements is the number of rows multiplied by the dimension of any
+//             fixed size list wrappers.  This is what you get when you flatten the FSL layer and
+//             is the starting point for structural encoding.  Note that an element can be a list
+//             value or a single primitive value.
+// - num_items: The number of items is the number of values in the repetition and definition vectors
+//             after everything has been flattened.
+// - num_visible_items: The number of visible items is the number of items after invisible items
+//             have been removed.  Invisible items are rep/def levels that don't correspond to an
+//             actual value.
+//
+// Note that we haven't exactly defined LIST<FIXED_SIZE_LIST<..>> yet.  Both FIXED_SIZE_LIST<LIST<..>>
+// and LIST<FIXED_SIZE_LIST<..>> haven't been fully implemented and tested.
+
+/// Describes the meaning of each repdef layer in a mini-block layout
 enum RepDefLayer {
   // Should never be used, included for debugging purporses and general protobuf best practice
   REPDEF_UNSPECIFIED = 0;
@@ -375,10 +394,25 @@ message FullZipLayout {
   uint32 bits_rep = 1;
   // The number of bits of definition info (0 if there is no definition)
   uint32 bits_def = 2;
+  // The number of bits of value info
+  //
+  // Note: we use bits here (and not bytes) for consistency with other encodings.  However, in practice,
+  // there is never a reason to use a bits per value that is not a multiple of 8.  The complexity is not
+  // worth the small savings in space since this encoding is typically used with large values already.
+  oneof details {
+    // If this is a fixed width block then we need to have a fixed number of bits per value
+    uint32 bits_per_value = 3;
+    // If this is a variable width block then we need to have a fixed number of bits per offset
+    uint32 bits_per_offset = 4;
+  }
+  // The number of items in the page
+  uint32 num_items = 5;
+  // The number of visible items in the page
+  uint32 num_visible_items = 6;
   // Description of the compression of values
-  ArrayEncoding value_compression = 3;
+  ArrayEncoding value_compression = 7;
   // The meaning of each repdef layer, used to interpret repdef buffers correctly
-  repeated RepDefLayer layers = 4;
+  repeated RepDefLayer layers = 8;
 }
 
 /// A layout used for pages where all values are null

rust/lance-encoding/src/data.rs

@@ -631,6 +631,16 @@ impl VariableWidthBlock {
         })
     }
 
+    pub fn offsets_as_block(&mut self) -> DataBlock {
+        let offsets = self.offsets.borrow_and_clone();
+        DataBlock::FixedWidth(FixedWidthDataBlock {
+            data: offsets,
+            bits_per_value: self.bits_per_offset as u64,
+            num_values: self.num_values + 1,
+            block_info: BlockInfo::new(),
+        })
+    }
+
     pub fn data_size(&self) -> u64 {
         (self.data.len() + self.offsets.len()) as u64
     }

rust/lance-encoding/src/decoder.rs

@@ -235,7 +235,7 @@ use lance_core::{Error, Result};
 use tracing::instrument;
 
 use crate::buffer::LanceBuffer;
-use crate::data::DataBlock;
+use crate::data::{DataBlock, FixedWidthDataBlock, VariableWidthBlock};
 use crate::encoder::{values_column_encoding, EncodedBatch};
 use crate::encodings::logical::binary::BinaryFieldScheduler;
 use crate::encodings::logical::blob::BlobFieldScheduler;
@@ -248,7 +248,9 @@ use crate::encodings::logical::primitive::{
 use crate::encodings::logical::r#struct::{
     SimpleStructDecoder, SimpleStructScheduler, StructuralStructDecoder, StructuralStructScheduler,
 };
-use crate::encodings::physical::binary::{BinaryBlockDecompressor, BinaryMiniBlockDecompressor};
+use crate::encodings::physical::binary::{
+    BinaryBlockDecompressor, BinaryMiniBlockDecompressor, VariableDecoder,
+};
 use crate::encodings::physical::bitpack_fastlanes::BitpackMiniBlockDecompressor;
 use crate::encodings::physical::fsst::FsstMiniBlockDecompressor;
 use crate::encodings::physical::struct_encoding::PackedStructFixedWidthMiniBlockDecompressor;
@@ -459,17 +461,20 @@ pub trait MiniBlockDecompressor: std::fmt::Debug + Send + Sync {
     fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result<DataBlock>;
 }
 
-pub trait PerValueDecompressor: std::fmt::Debug + Send + Sync {
+pub trait FixedPerValueDecompressor: std::fmt::Debug + Send + Sync {
     /// Decompress one or more values
-    fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result<DataBlock>;
+    fn decompress(&self, data: FixedWidthDataBlock) -> Result<DataBlock>;
     /// The number of bits in each value
     ///
-    /// Returns 0 if the data type is variable-width
-    ///
     /// Currently (and probably long term) this must be a multiple of 8
     fn bits_per_value(&self) -> u64;
 }
 
+pub trait VariablePerValueDecompressor: std::fmt::Debug + Send + Sync {
+    /// Decompress one or more values
+    fn decompress(&self, data: VariableWidthBlock) -> Result<DataBlock>;
+}
+
 pub trait BlockDecompressor: std::fmt::Debug + Send + Sync {
     fn decompress(&self, data: LanceBuffer) -> Result<DataBlock>;
 }
@@ -480,10 +485,15 @@ pub trait DecompressorStrategy: std::fmt::Debug + Send + Sync {
         description: &pb::ArrayEncoding,
     ) -> Result<Box<dyn MiniBlockDecompressor>>;
 
-    fn create_per_value_decompressor(
+    fn create_fixed_per_value_decompressor(
+        &self,
+        description: &pb::ArrayEncoding,
+    ) -> Result<Box<dyn FixedPerValueDecompressor>>;
+
+    fn create_variable_per_value_decompressor(
         &self,
         description: &pb::ArrayEncoding,
-    ) -> Result<Box<dyn PerValueDecompressor>>;
+    ) -> Result<Box<dyn VariablePerValueDecompressor>>;
 
     fn create_block_decompressor(
         &self,
@@ -506,10 +516,10 @@ impl DecompressorStrategy for CoreDecompressorStrategy {
             pb::array_encoding::ArrayEncoding::Bitpack2(description) => {
                 Ok(Box::new(BitpackMiniBlockDecompressor::new(description)))
             }
-            pb::array_encoding::ArrayEncoding::BinaryMiniBlock(_) => {
+            pb::array_encoding::ArrayEncoding::Variable(_) => {
                 Ok(Box::new(BinaryMiniBlockDecompressor::default()))
             }
-            pb::array_encoding::ArrayEncoding::FsstMiniBlock(description) => {
+            pb::array_encoding::ArrayEncoding::Fsst(description) => {
                 Ok(Box::new(FsstMiniBlockDecompressor::new(description)))
             }
             pb::array_encoding::ArrayEncoding::PackedStructFixedWidthMiniBlock(description) => {
@@ -521,15 +531,28 @@ impl DecompressorStrategy for CoreDecompressorStrategy {
         }
     }
 
-    fn create_per_value_decompressor(
+    fn create_fixed_per_value_decompressor(
         &self,
         description: &pb::ArrayEncoding,
-    ) -> Result<Box<dyn PerValueDecompressor>> {
+    ) -> Result<Box<dyn FixedPerValueDecompressor>> {
         match description.array_encoding.as_ref().unwrap() {
             pb::array_encoding::ArrayEncoding::Flat(flat) => {
                 Ok(Box::new(ValueDecompressor::new(flat)))
             }
-            _ => todo!(),
+            _ => todo!("fixed-per-value decompressor for {:?}", description),
+        }
+    }
+
+    fn create_variable_per_value_decompressor(
+        &self,
+        description: &pb::ArrayEncoding,
+    ) -> Result<Box<dyn VariablePerValueDecompressor>> {
+        match description.array_encoding.as_ref().unwrap() {
+            &pb::array_encoding::ArrayEncoding::Variable(variable) => {
+                assert!(variable.bits_per_offset < u8::MAX as u32);
+                Ok(Box::new(VariableDecoder::default()))
+            }
+            _ => todo!("variable-per-value decompressor for {:?}", description),
         }
     }
 
@@ -548,7 +571,7 @@ impl DecompressorStrategy for CoreDecompressorStrategy {
                     constant.num_values,
                 )))
             }
-            pb::array_encoding::ArrayEncoding::BinaryBlock(_) => {
+            pb::array_encoding::ArrayEncoding::Variable(_) => {
                 Ok(Box::new(BinaryBlockDecompressor::default()))
             }
             _ => todo!(),

rust/lance-encoding/src/encoder.rs

@@ -24,14 +24,16 @@ use crate::encodings::logical::list::ListStructuralEncoder;
 use crate::encodings::logical::primitive::PrimitiveStructuralEncoder;
 use crate::encodings::logical::r#struct::StructFieldEncoder;
 use crate::encodings::logical::r#struct::StructStructuralEncoder;
-use crate::encodings::physical::binary::{BinaryBlockEncoder, BinaryMiniBlockEncoder};
+use crate::encodings::physical::binary::{BinaryMiniBlockEncoder, VariableEncoder};
 use crate::encodings::physical::bitpack_fastlanes::BitpackedForNonNegArrayEncoder;
 use crate::encodings::physical::bitpack_fastlanes::{
     compute_compressed_bit_width_for_non_neg, BitpackMiniBlockEncoder,
 };
 use crate::encodings::physical::block_compress::{CompressionConfig, CompressionScheme};
 use crate::encodings::physical::dictionary::AlreadyDictionaryEncoder;
-use crate::encodings::physical::fsst::{FsstArrayEncoder, FsstMiniBlockEncoder};
+use crate::encodings::physical::fsst::{
+    FsstArrayEncoder, FsstMiniBlockEncoder, FsstPerValueEncoder,
+};
 use crate::encodings::physical::packed_struct::PackedStructEncoder;
 use crate::encodings::physical::struct_encoding::PackedStructFixedWidthMiniBlockEncoder;
 use crate::format::ProtobufUtils;
@@ -217,11 +219,21 @@ pub trait MiniBlockCompressor: std::fmt::Debug + Send + Sync {
 /// A single buffer of value data and a buffer of offsets
 ///
 /// TODO: In the future we may allow metadata buffers
+#[derive(Debug)]
 pub enum PerValueDataBlock {
     Fixed(FixedWidthDataBlock),
     Variable(VariableWidthBlock),
 }
 
+impl PerValueDataBlock {
+    pub fn data_size(&self) -> u64 {
+        match self {
+            Self::Fixed(fixed) => fixed.data_size(),
+            Self::Variable(variable) => variable.data_size(),
+        }
+    }
+}
+
 /// Trait for compression algorithms that are suitable for use in the zipped structural encoding
 ///
 /// This compression must return either a FixedWidthDataBlock or a VariableWidthBlock.  This is because
@@ -884,8 +896,23 @@ impl CompressionStrategy for CoreArrayEncodingStrategy {
                 let encoder = Box::new(ValueEncoder::default());
                 Ok(encoder)
             }
-            DataBlock::VariableWidth(_variable_width) => {
-                todo!()
+            DataBlock::VariableWidth(variable_width) => {
+                if variable_width.bits_per_offset == 32 {
+                    let data_size = variable_width.expect_single_stat::<UInt64Type>(Stat::DataSize);
+                    let max_len = variable_width.expect_single_stat::<UInt64Type>(Stat::MaxLength);
+
+                    let variable_compression = Box::new(VariableEncoder::default());
+
+                    if max_len >= FSST_LEAST_INPUT_MAX_LENGTH
+                        && data_size >= FSST_LEAST_INPUT_SIZE as u64
+                    {
+                        Ok(Box::new(FsstPerValueEncoder::new(variable_compression)))
+                    } else {
+                        Ok(variable_compression)
+                    }
+                } else {
+                    todo!("Implement MiniBlockCompression for VariableWidth DataBlock with 64 bits offsets.")
+                }
             }
             _ => unreachable!(),
         }
@@ -905,13 +932,9 @@ impl CompressionStrategy for CoreArrayEncodingStrategy {
                 Ok((encoder, encoding))
             }
             DataBlock::VariableWidth(variable_width) => {
-                if variable_width.bits_per_offset == 32 {
-                    let encoder = Box::new(BinaryBlockEncoder::default());
-                    let encoding = ProtobufUtils::binary_block();
-                    Ok((encoder, encoding))
-                } else {
-                    todo!("Implement BlockCompression for VariableWidth DataBlock with 64 bits offsets.")
-                }
+                let encoder = Box::new(VariableEncoder::default());
+                let encoding = ProtobufUtils::variable(variable_width.bits_per_offset);
+                Ok((encoder, encoding))
             }
             _ => unreachable!(),
         }

rust/lance-encoding/src/encodings/logical/list.rs

@@ -1583,8 +1583,43 @@ mod tests {
             .await;
     }
 
+    #[rstest]
     #[test_log::test(tokio::test)]
-    async fn test_simple_sliced_list() {
+    async fn test_simple_string_list(
+        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
+        structural_encoding: &str,
+    ) {
+        let items_builder = StringBuilder::new();
+        let mut list_builder = ListBuilder::new(items_builder);
+        list_builder.append_value([Some("a"), Some("bc"), Some("def")]);
+        list_builder.append_value([Some("gh"), None]);
+        list_builder.append_null();
+        list_builder.append_value([Some("ijk"), Some("lmnop"), Some("qrs")]);
+        let list_array = list_builder.finish();
+
+        let mut field_metadata = HashMap::new();
+        field_metadata.insert(
+            STRUCTURAL_ENCODING_META_KEY.to_string(),
+            structural_encoding.into(),
+        );
+
+        let test_cases = TestCases::default()
+            .with_range(0..2)
+            .with_range(0..3)
+            .with_range(1..3)
+            .with_indices(vec![1, 3])
+            .with_indices(vec![2])
+            .with_file_version(LanceFileVersion::V2_1);
+        check_round_trip_encoding_of_data(vec![Arc::new(list_array)], &test_cases, field_metadata)
+            .await;
+    }
+
+    #[rstest]
+    #[test_log::test(tokio::test)]
+    async fn test_simple_sliced_list(
+        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
+        structural_encoding: &str,
+    ) {
         let items_builder = Int32Builder::new();
         let mut list_builder = ListBuilder::new(items_builder);
         list_builder.append_value([Some(1), Some(2), Some(3)]);
@@ -1595,18 +1630,28 @@ mod tests {
 
         let list_array = list_array.slice(1, 2);
 
+        let mut field_metadata = HashMap::new();
+        field_metadata.insert(
+            STRUCTURAL_ENCODING_META_KEY.to_string(),
+            structural_encoding.into(),
+        );
+
         let test_cases = TestCases::default()
             .with_range(0..2)
             .with_range(1..2)
             .with_indices(vec![0])
             .with_indices(vec![1])
             .with_file_version(LanceFileVersion::V2_1);
-        check_round_trip_encoding_of_data(vec![Arc::new(list_array)], &test_cases, HashMap::new())
+        check_round_trip_encoding_of_data(vec![Arc::new(list_array)], &test_cases, field_metadata)
             .await;
     }
 
+    #[rstest]
     #[test_log::test(tokio::test)]
-    async fn test_list_with_garbage_nulls() {
+    async fn test_list_with_garbage_nulls(
+        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
+        structural_encoding: &str,
+    ) {
         // In Arrow, list nulls are allowed to be non-empty, with masked garbage values
         // Here we make a list with a null row in the middle with 3 garbage values
         let items = UInt64Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
@@ -1620,13 +1665,19 @@ mod tests {
             Some(list_validity),
         );
 
+        let mut field_metadata = HashMap::new();
+        field_metadata.insert(
+            STRUCTURAL_ENCODING_META_KEY.to_string(),
+            structural_encoding.into(),
+        );
+
         let test_cases = TestCases::default()
             .with_range(0..3)
             .with_range(1..2)
             .with_indices(vec![1])
             .with_indices(vec![2])
             .with_file_version(LanceFileVersion::V2_1);
-        check_round_trip_encoding_of_data(vec![Arc::new(list_arr)], &test_cases, HashMap::new())
+        check_round_trip_encoding_of_data(vec![Arc::new(list_arr)], &test_cases, field_metadata)
             .await;
     }
 

rust/lance-encoding/src/encodings/logical/primitive.rs

@@ -285,9 +285,7 @@ pub fn decoder_from_array_encoding(
         // 2.1 only
         pb::array_encoding::ArrayEncoding::Constant(_) => unreachable!(),
         pb::array_encoding::ArrayEncoding::Bitpack2(_) => unreachable!(),
-        pb::array_encoding::ArrayEncoding::BinaryMiniBlock(_) => unreachable!(),
-        pb::array_encoding::ArrayEncoding::FsstMiniBlock(_) => unreachable!(),
-        pb::array_encoding::ArrayEncoding::BinaryBlock(_) => unreachable!(),
+        pb::array_encoding::ArrayEncoding::Variable(_) => unreachable!(),
         pb::array_encoding::ArrayEncoding::PackedStructFixedWidthMiniBlock(_) => unreachable!(),
     }
 }

rust/lance-encoding/src/encodings/physical.rs

@@ -16,15 +16,20 @@ use snafu::{location, Location};
 
 use futures::{future::BoxFuture, FutureExt};
 
-use crate::decoder::{BlockDecompressor, LogicalPageDecoder, MiniBlockDecompressor};
-use crate::encoder::{BlockCompressor, MiniBlockChunk, MiniBlockCompressed, MiniBlockCompressor};
+use crate::decoder::{
+    BlockDecompressor, LogicalPageDecoder, MiniBlockDecompressor, VariablePerValueDecompressor,
+};
+use crate::encoder::{
+    BlockCompressor, MiniBlockChunk, MiniBlockCompressed, MiniBlockCompressor, PerValueCompressor,
+    PerValueDataBlock,
+};
 use crate::encodings::logical::primitive::PrimitiveFieldDecoder;
 
 use crate::buffer::LanceBuffer;
 use crate::data::{
     BlockInfo, DataBlock, FixedWidthDataBlock, NullableDataBlock, VariableWidthBlock,
 };
-use crate::format::ProtobufUtils;
+use crate::format::{pb, ProtobufUtils};
 use crate::{
     decoder::{PageScheduler, PrimitivePageDecoder},
     encoder::{ArrayEncoder, EncodedArray},
@@ -687,16 +692,13 @@ impl BinaryMiniBlockEncoder {
                 chunks,
                 num_values: data.num_values,
             },
-            ProtobufUtils::binary_miniblock(),
+            ProtobufUtils::variable(/*bits_per_value=*/ 32),
         )
     }
 }
 
 impl MiniBlockCompressor for BinaryMiniBlockEncoder {
-    fn compress(
-        &self,
-        data: DataBlock,
-    ) -> Result<(MiniBlockCompressed, crate::format::pb::ArrayEncoding)> {
+    fn compress(&self, data: DataBlock) -> Result<(MiniBlockCompressed, pb::ArrayEncoding)> {
         match data {
             DataBlock::VariableWidth(variable_width) => Ok(self.chunk_data(variable_width)),
             _ => Err(Error::InvalidInput {
@@ -740,9 +742,11 @@ impl MiniBlockDecompressor for BinaryMiniBlockDecompressor {
     }
 }
 
+/// Most basic encoding for variable-width data which does no compression at all
 #[derive(Debug, Default)]
-pub struct BinaryBlockEncoder {}
-impl BlockCompressor for BinaryBlockEncoder {
+pub struct VariableEncoder {}
+
+impl BlockCompressor for VariableEncoder {
     fn compress(&self, data: DataBlock) -> Result<LanceBuffer> {
         let num_values: u32 = data
             .num_values()
@@ -785,6 +789,26 @@ impl BlockCompressor for BinaryBlockEncoder {
     }
 }
 
+impl PerValueCompressor for VariableEncoder {
+    fn compress(&self, data: DataBlock) -> Result<(PerValueDataBlock, pb::ArrayEncoding)> {
+        let DataBlock::VariableWidth(variable) = data else {
+            panic!("BinaryPerValueCompressor can only work with Variable Width DataBlock.");
+        };
+
+        let encoding = ProtobufUtils::variable(variable.bits_per_offset);
+        Ok((PerValueDataBlock::Variable(variable), encoding))
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct VariableDecoder {}
+
+impl VariablePerValueDecompressor for VariableDecoder {
+    fn decompress(&self, data: VariableWidthBlock) -> Result<DataBlock> {
+        Ok(DataBlock::VariableWidth(data))
+    }
+}
+
 #[derive(Debug, Default)]
 pub struct BinaryBlockDecompressor {}
 

rust/lance-encoding/src/encodings/physical/binary.rs

@@ -14,10 +14,14 @@ use crate::{
     buffer::LanceBuffer,
     data::{BlockInfo, DataBlock, NullableDataBlock, VariableWidthBlock},
     decoder::{MiniBlockDecompressor, PageScheduler, PrimitivePageDecoder},
-    encoder::{ArrayEncoder, EncodedArray},
-    encoder::{MiniBlockCompressed, MiniBlockCompressor},
-    format::pb::{self},
-    format::ProtobufUtils,
+    encoder::{
+        ArrayEncoder, EncodedArray, MiniBlockCompressed, MiniBlockCompressor, PerValueCompressor,
+        PerValueDataBlock,
+    },
+    format::{
+        pb::{self},
+        ProtobufUtils,
+    },
     EncodingsIo,
 };
 
@@ -202,14 +206,13 @@ impl ArrayEncoder for FsstArrayEncoder {
     }
 }
 
-#[derive(Debug, Default)]
-pub struct FsstMiniBlockEncoder {}
+struct FsstCompressed {
+    data: VariableWidthBlock,
+    symbol_table: Vec<u8>,
+}
 
-impl MiniBlockCompressor for FsstMiniBlockEncoder {
-    fn compress(
-        &self,
-        data: DataBlock,
-    ) -> Result<(MiniBlockCompressed, crate::format::pb::ArrayEncoding)> {
+impl FsstCompressed {
+    fn fsst_compress(data: DataBlock) -> Result<Self> {
         match data {
             DataBlock::VariableWidth(mut variable_width) => {
                 let offsets = variable_width.offsets.borrow_to_typed_slice::<i32>();
@@ -231,29 +234,22 @@ impl MiniBlockCompressor for FsstMiniBlockEncoder {
                 )?;
 
                 // construct `DataBlock` for BinaryMiniBlockEncoder, we may want some `DataBlock` construct methods later
-                let data_block = DataBlock::VariableWidth(VariableWidthBlock {
+                let compressed = VariableWidthBlock {
                     data: LanceBuffer::reinterpret_vec(dest_values),
                     bits_per_offset: 32,
                     offsets: LanceBuffer::reinterpret_vec(dest_offsets),
                     num_values: variable_width.num_values,
                     block_info: BlockInfo::new(),
-                });
-
-                // compress the fsst compressed data using `BinaryMiniBlockEncoder`
-                let binary_compressor =
-                    Box::new(BinaryMiniBlockEncoder::default()) as Box<dyn MiniBlockCompressor>;
+                };
 
-                let (binary_miniblock_compressed, binary_array_encoding) =
-                    binary_compressor.compress(data_block)?;
-
-                Ok((
-                    binary_miniblock_compressed,
-                    ProtobufUtils::fsst_mini_block(binary_array_encoding, symbol_table),
-                ))
+                Ok(Self {
+                    data: compressed,
+                    symbol_table,
+                })
             }
             _ => Err(Error::InvalidInput {
                 source: format!(
-                    "Cannot compress a data block of type {} with BinaryMiniBlockEncoder",
+                    "Cannot compress a data block of type {} with FsstEncoder",
                     data.name()
                 )
                 .into(),
@@ -263,13 +259,65 @@ impl MiniBlockCompressor for FsstMiniBlockEncoder {
     }
 }
 
+#[derive(Debug, Default)]
+pub struct FsstMiniBlockEncoder {}
+
+impl MiniBlockCompressor for FsstMiniBlockEncoder {
+    fn compress(
+        &self,
+        data: DataBlock,
+    ) -> Result<(MiniBlockCompressed, crate::format::pb::ArrayEncoding)> {
+        let compressed = FsstCompressed::fsst_compress(data)?;
+
+        let data_block = DataBlock::VariableWidth(compressed.data);
+
+        // compress the fsst compressed data using `BinaryMiniBlockEncoder`
+        let binary_compressor =
+            Box::new(BinaryMiniBlockEncoder::default()) as Box<dyn MiniBlockCompressor>;
+
+        let (binary_miniblock_compressed, binary_array_encoding) =
+            binary_compressor.compress(data_block)?;
+
+        Ok((
+            binary_miniblock_compressed,
+            ProtobufUtils::fsst(binary_array_encoding, compressed.symbol_table),
+        ))
+    }
+}
+
+#[derive(Debug)]
+pub struct FsstPerValueEncoder {
+    inner: Box<dyn PerValueCompressor>,
+}
+
+impl FsstPerValueEncoder {
+    pub fn new(inner: Box<dyn PerValueCompressor>) -> Self {
+        Self { inner }
+    }
+}
+
+impl PerValueCompressor for FsstPerValueEncoder {
+    fn compress(&self, data: DataBlock) -> Result<(PerValueDataBlock, pb::ArrayEncoding)> {
+        let compressed = FsstCompressed::fsst_compress(data)?;
+
+        let data_block = DataBlock::VariableWidth(compressed.data);
+
+        let (binary_compressed, binary_array_encoding) = self.inner.compress(data_block)?;
+
+        Ok((
+            binary_compressed,
+            ProtobufUtils::fsst(binary_array_encoding, compressed.symbol_table),
+        ))
+    }
+}
+
 #[derive(Debug)]
 pub struct FsstMiniBlockDecompressor {
     symbol_table: Vec<u8>,
 }
 
 impl FsstMiniBlockDecompressor {
-    pub fn new(description: &pb::FsstMiniBlock) -> Self {
+    pub fn new(description: &pb::Fsst) -> Self {
         Self {
             symbol_table: description.symbol_table.clone(),
         }

rust/lance-encoding/src/encodings/physical/fsst.rs

@@ -14,8 +14,7 @@ use crate::buffer::LanceBuffer;
 use crate::data::{
     BlockInfo, ConstantDataBlock, DataBlock, FixedSizeListBlock, FixedWidthDataBlock,
 };
-use crate::decoder::PerValueDecompressor;
-use crate::decoder::{BlockDecompressor, MiniBlockDecompressor};
+use crate::decoder::{BlockDecompressor, FixedPerValueDecompressor, MiniBlockDecompressor};
 use crate::encoder::{
     BlockCompressor, MiniBlockChunk, MiniBlockCompressed, MiniBlockCompressor, PerValueCompressor,
     PerValueDataBlock, MAX_MINIBLOCK_BYTES, MAX_MINIBLOCK_VALUES,
@@ -413,37 +412,33 @@ impl ValueDecompressor {
             bytes_per_value: description.bits_per_value / 8,
         }
     }
-}
 
-impl BlockDecompressor for ValueDecompressor {
-    fn decompress(&self, data: LanceBuffer) -> Result<DataBlock> {
-        let num_values = data.len() as u64 / self.bytes_per_value;
-        assert_eq!(data.len() as u64 % self.bytes_per_value, 0);
-        Ok(DataBlock::FixedWidth(FixedWidthDataBlock {
+    fn buffer_to_block(&self, data: LanceBuffer) -> DataBlock {
+        DataBlock::FixedWidth(FixedWidthDataBlock {
             bits_per_value: self.bytes_per_value * 8,
+            num_values: data.len() as u64 / self.bytes_per_value,
             data,
-            num_values,
             block_info: BlockInfo::new(),
-        }))
+        })
+    }
+}
+
+impl BlockDecompressor for ValueDecompressor {
+    fn decompress(&self, data: LanceBuffer) -> Result<DataBlock> {
+        Ok(self.buffer_to_block(data))
     }
 }
 
 impl MiniBlockDecompressor for ValueDecompressor {
     fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result<DataBlock> {
-        debug_assert!(data.len() as u64 >= num_values * self.bytes_per_value);
-
-        Ok(DataBlock::FixedWidth(FixedWidthDataBlock {
-            data,
-            bits_per_value: self.bytes_per_value * 8,
-            num_values,
-            block_info: BlockInfo::new(),
-        }))
+        assert_eq!(data.len() as u64, num_values * self.bytes_per_value);
+        Ok(self.buffer_to_block(data))
     }
 }
 
-impl PerValueDecompressor for ValueDecompressor {
-    fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result<DataBlock> {
-        MiniBlockDecompressor::decompress(self, data, num_values)
+impl FixedPerValueDecompressor for ValueDecompressor {
+    fn decompress(&self, data: FixedWidthDataBlock) -> Result<DataBlock> {
+        Ok(DataBlock::FixedWidth(data))
     }
 
     fn bits_per_value(&self) -> u64 {

rust/lance-encoding/src/encodings/physical/value.rs

@@ -17,12 +17,12 @@ pub mod pb {
 use pb::{
     array_encoding::ArrayEncoding as ArrayEncodingEnum,
     buffer::BufferType,
+    full_zip_layout,
     nullable::{AllNull, NoNull, Nullability, SomeNull},
     page_layout::Layout,
-    AllNullLayout, ArrayEncoding, Binary, BinaryBlock, BinaryMiniBlock, Bitpack2, Bitpacked,
-    BitpackedForNonNeg, Dictionary, FixedSizeBinary, FixedSizeList, Flat, Fsst, FsstMiniBlock,
-    MiniBlockLayout, Nullable, PackedStruct, PackedStructFixedWidthMiniBlock, PageLayout,
-    RepDefLayer,
+    AllNullLayout, ArrayEncoding, Binary, Bitpack2, Bitpacked, BitpackedForNonNeg, Dictionary,
+    FixedSizeBinary, FixedSizeList, Flat, Fsst, MiniBlockLayout, Nullable, PackedStruct,
+    PackedStructFixedWidthMiniBlock, PageLayout, RepDefLayer, Variable,
 };
 
 use crate::{
@@ -145,25 +145,21 @@ impl ProtobufUtils {
         }
     }
 
-    pub fn binary_miniblock() -> ArrayEncoding {
+    pub fn variable(bits_per_offset: u8) -> ArrayEncoding {
         ArrayEncoding {
-            array_encoding: Some(ArrayEncodingEnum::BinaryMiniBlock(BinaryMiniBlock {})),
-        }
-    }
-
-    pub fn binary_block() -> ArrayEncoding {
-        ArrayEncoding {
-            array_encoding: Some(ArrayEncodingEnum::BinaryBlock(BinaryBlock {})),
+            array_encoding: Some(ArrayEncodingEnum::Variable(Variable {
+                bits_per_offset: bits_per_offset as u32,
+            })),
         }
     }
 
     // Construct a `FsstMiniBlock` ArrayEncoding, the inner `binary_mini_block` encoding is actually
     // not used and `FsstMiniBlockDecompressor` constructs a `binary_mini_block` in a `hard-coded` fashion.
     // This can be an optimization later.
-    pub fn fsst_mini_block(data: ArrayEncoding, symbol_table: Vec<u8>) -> ArrayEncoding {
+    pub fn fsst(data: ArrayEncoding, symbol_table: Vec<u8>) -> ArrayEncoding {
         ArrayEncoding {
-            array_encoding: Some(ArrayEncodingEnum::FsstMiniBlock(Box::new(FsstMiniBlock {
-                binary_mini_block: Some(Box::new(data)),
+            array_encoding: Some(ArrayEncodingEnum::Fsst(Box::new(Fsst {
+                binary: Some(Box::new(data)),
                 symbol_table,
             }))),
         }
@@ -246,15 +242,6 @@ impl ProtobufUtils {
         }
     }
 
-    pub fn fsst(data: ArrayEncoding, symbol_table: Vec<u8>) -> ArrayEncoding {
-        ArrayEncoding {
-            array_encoding: Some(ArrayEncodingEnum::Fsst(Box::new(Fsst {
-                binary: Some(Box::new(data)),
-                symbol_table,
-            }))),
-        }
-    }
-
     fn def_inter_to_repdef_layer(def: DefinitionInterpretation) -> i32 {
         match def {
             DefinitionInterpretation::AllValidItem => RepDefLayer::RepdefAllValidItem as i32,
@@ -309,17 +296,23 @@ impl ProtobufUtils {
         }
     }
 
-    pub fn full_zip_layout(
+    fn full_zip_layout(
         bits_rep: u8,
         bits_def: u8,
+        details: full_zip_layout::Details,
         value_encoding: ArrayEncoding,
         def_meaning: &[DefinitionInterpretation],
+        num_items: u32,
+        num_visible_items: u32,
     ) -> PageLayout {
         PageLayout {
             layout: Some(Layout::FullZipLayout(pb::FullZipLayout {
                 bits_rep: bits_rep as u32,
                 bits_def: bits_def as u32,
+                details: Some(details),
                 value_compression: Some(value_encoding),
+                num_items,
+                num_visible_items,
                 layers: def_meaning
                     .iter()
                     .map(|&def| Self::def_inter_to_repdef_layer(def))
@@ -328,6 +321,46 @@ impl ProtobufUtils {
         }
     }
 
+    pub fn fixed_full_zip_layout(
+        bits_rep: u8,
+        bits_def: u8,
+        bits_per_value: u32,
+        value_encoding: ArrayEncoding,
+        def_meaning: &[DefinitionInterpretation],
+        num_items: u32,
+        num_visible_items: u32,
+    ) -> PageLayout {
+        Self::full_zip_layout(
+            bits_rep,
+            bits_def,
+            full_zip_layout::Details::BitsPerValue(bits_per_value),
+            value_encoding,
+            def_meaning,
+            num_items,
+            num_visible_items,
+        )
+    }
+
+    pub fn variable_full_zip_layout(
+        bits_rep: u8,
+        bits_def: u8,
+        bits_per_offset: u32,
+        value_encoding: ArrayEncoding,
+        def_meaning: &[DefinitionInterpretation],
+        num_items: u32,
+        num_visible_items: u32,
+    ) -> PageLayout {
+        Self::full_zip_layout(
+            bits_rep,
+            bits_def,
+            full_zip_layout::Details::BitsPerOffset(bits_per_offset),
+            value_encoding,
+            def_meaning,
+            num_items,
+            num_visible_items,
+        )
+    }
+
     pub fn all_null_layout(def_meaning: &[DefinitionInterpretation]) -> PageLayout {
         PageLayout {
             layout: Some(Layout::AllNullLayout(AllNullLayout {

rust/lance-encoding/src/format.rs

@@ -1783,6 +1783,7 @@ pub enum ControlWordIterator<'a> {
 }
 
 /// Describes the properties of a control word
+#[derive(Debug)]
 pub struct ControlWordDesc {
     pub is_new_row: bool,
     pub is_visible: bool,