feat(encoding): add bounded block dictionary

docs/src/format/file/encoding.md

@@ -587,6 +587,7 @@ on a per-value basis. We use ☑️ to mark a technique that is applied on a per
 | Constant        | ✅ (2.1)              | ❓                       | ❓                         |
 | Range           | ✅ (2.3)              | ❌                       | ❓                         |
 | Delta           | ✅ (2.3)              | ❌                       | ❓                         |
+| Dictionary      | ✅ (2.3)              | ❌                       | ❓                         |
 | Bitpacking      | ✅ (2.1)              | ❓                       | ✅ (2.1)                   |
 | Fsst            | ❓                    | ✅ (2.1)                 | ✅ (2.1)                   |
 | Rle             | ✅ (2.2)              | ❌                       | ✅ (2.1)                   |
@@ -616,7 +617,19 @@ with checked prefix sums.
 %%% proto.message.Delta %%%
 ```
 
-Lance 2.0 through 2.2 writers do not emit either encoding.
+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:
+
+```text
+Dictionary payload:
+  u64 indices_payload_bytes
+  u64 items_payload_bytes
+  indices 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.
 
 ### Flat
 

rust/lance-encoding/src/compression.rs

@@ -1209,7 +1209,7 @@ impl DecompressionStrategy for DefaultDecompressionStrategy {
             Compression::OutOfLineBitpacking(_) => Err(Error::not_supported_source(
                 "this runtime was not built with bitpacking support".into(),
             )),
-            Compression::Range(_) | Compression::Delta(_) => {
+            Compression::Dictionary(_) | Compression::Range(_) | Compression::Delta(_) => {
                 let value_type = block::infer_block_value_type(description)?;
                 block::create_block_decompressor(description, value_type)
                     .map(|(decompressor, _)| decompressor)

rust/lance-encoding/src/compression/block.rs

@@ -8,6 +8,7 @@
 
 use lance_core::{Error, Result};
 
+pub(crate) const MAX_DICTIONARY_ITEMS: usize = 4096;
 #[cfg(feature = "bitpacking")]
 pub(crate) const BITPACK_CHUNK_VALUES: u64 = 1024;
 

rust/lance-encoding/src/compression/block/factory.rs

@@ -14,6 +14,7 @@ use crate::{
         block::{CompressionConfig, CompressionScheme},
         constant::ConstantBlockDecompressor,
         delta::DeltaDecompressor,
+        dictionary::DictionaryBlockDecompressor,
         general::GenericGeneralBlockDecompressor,
         range::RangeDecompressor,
         rle::{BlockRleDecompressor, BlockRunCount, MetadataRunLengths},
@@ -436,6 +437,47 @@ fn create_inner(
                 values_have_payload || run_lengths_have_payload,
             ))
         }
+        Compression::Dictionary(dictionary) => {
+            if position != Position::Root {
+                return Err(Error::invalid_input(
+                    "Dictionary is not supported as a block codec child",
+                ));
+            }
+            if dictionary.num_dictionary_items == 0
+                || dictionary.num_dictionary_items as usize > MAX_DICTIONARY_ITEMS
+            {
+                return Err(Error::invalid_input(format!(
+                    "Dictionary item count {} is outside 1..={MAX_DICTIONARY_ITEMS}",
+                    dictionary.num_dictionary_items
+                )));
+            }
+            let indices_encoding = dictionary.indices.as_deref().ok_or_else(|| {
+                Error::invalid_input("Dictionary is missing its indices encoding")
+            })?;
+            let items_encoding = dictionary
+                .items
+                .as_deref()
+                .ok_or_else(|| Error::invalid_input("Dictionary is missing its items encoding"))?;
+            let (indices, indices_have_payload) = create_inner(
+                indices_encoding,
+                BlockValueType::UInt32,
+                Position::Child,
+                false,
+            )?;
+            let (items, items_have_payload) =
+                create_inner(items_encoding, expected_type, Position::Child, false)?;
+            Ok((
+                Box::new(DictionaryBlockDecompressor::new(
+                    expected_type,
+                    dictionary.num_dictionary_items,
+                    indices,
+                    items,
+                    indices_have_payload,
+                    items_have_payload,
+                )),
+                indices_have_payload || items_have_payload,
+            ))
+        }
         other => Err(Error::invalid_input(format!(
             "Unsupported block sequence encoding: {}",
             compression_name(other)
@@ -559,6 +601,13 @@ fn infer_inner(encoding: &CompressiveEncoding, position: Position) -> Result<Blo
                 .ok_or_else(|| Error::invalid_input("RLE is missing its values encoding"))?,
             Position::Child,
         ),
+        Compression::Dictionary(dictionary) if position == Position::Root => infer_inner(
+            dictionary
+                .items
+                .as_deref()
+                .ok_or_else(|| Error::invalid_input("Dictionary is missing its items encoding"))?,
+            Position::Child,
+        ),
         other => Err(Error::invalid_input(format!(
             "Cannot infer bounded block value type from {} at {position:?}",
             compression_name(other)

rust/lance-encoding/src/compression/block/tests.rs

@@ -3,13 +3,15 @@
 
 use super::*;
 
+use std::sync::Arc;
+
 use crate::{
     buffer::LanceBuffer,
     compression::BlockCompressor,
     data::{BlockInfo, DataBlock, FixedWidthDataBlock},
     encodings::physical::{
-        constant::ConstantBlockCompressor, range::RangeEncoder, rle::BlockRleCompressor,
-        value::FixedWidthBlockCompressor,
+        constant::ConstantBlockCompressor, dictionary::DictionaryBlockCompressor,
+        range::RangeEncoder, rle::BlockRleCompressor, value::FixedWidthBlockCompressor,
     },
     format::{ProtobufUtils21, pb21::CompressiveEncoding},
 };
@@ -145,6 +147,30 @@ fn rle_compressor_owns_and_reuses_children() {
     round_trip_u64(&values, compressor, encoding);
 }
 
+#[test]
+fn dictionary_materializes_once_when_built() {
+    use crate::encodings::physical::dictionary::BLOCK_MATERIALIZATION_COUNT;
+
+    let values = (0..4096_u64)
+        .map(|index| [u64::MAX - 9, 17, 1_u64 << 50, 991][index as usize % 4])
+        .collect::<Vec<_>>();
+    let dictionary_items = Arc::from([17_u64, 991, 1_u64 << 50, u64::MAX - 9].as_slice());
+    let compressor = Box::new(DictionaryBlockCompressor::new(
+        BlockValueType::UInt64,
+        dictionary_items,
+        Box::new(FixedWidthBlockCompressor::new(BlockValueType::UInt32)),
+        Box::new(FixedWidthBlockCompressor::new(BlockValueType::UInt64)),
+    ));
+    let encoding = ProtobufUtils21::dictionary(
+        ProtobufUtils21::flat(32, None),
+        ProtobufUtils21::flat(64, None),
+        4,
+    );
+    BLOCK_MATERIALIZATION_COUNT.with(|count| count.set(0));
+    round_trip_u64(&values, compressor, encoding);
+    BLOCK_MATERIALIZATION_COUNT.with(|count| assert_eq!(count.get(), 1));
+}
+
 #[cfg(feature = "bitpacking")]
 #[test]
 fn out_of_line_bitpacking_round_trip() {
@@ -206,6 +232,18 @@ fn factory_rejects_unbounded_or_mistyped_trees() {
             .to_string()
             .contains("expected 64")
     );
+
+    let oversized = ProtobufUtils21::dictionary(
+        ProtobufUtils21::flat(32, None),
+        ProtobufUtils21::flat(64, None),
+        (MAX_DICTIONARY_ITEMS + 1) as u32,
+    );
+    assert!(
+        create_block_decompressor(&oversized, BlockValueType::UInt64)
+            .unwrap_err()
+            .to_string()
+            .contains("outside")
+    );
 }
 
 #[test]
@@ -328,6 +366,60 @@ fn rle_framing_is_fallible() {
     assert!(error.to_string().contains("Metadata-only RLE run-length"));
 }
 
+#[test]
+fn metadata_dictionary_checks_index_bounds() {
+    let encoding = ProtobufUtils21::dictionary(
+        ProtobufUtils21::range(32, 0, 1),
+        ProtobufUtils21::range(64, 10, 1),
+        2,
+    );
+    let (decoder, has_payload) =
+        create_block_decompressor(&encoding, BlockValueType::UInt64).unwrap();
+    assert!(!has_payload);
+    let error = decoder.decompress(None, 3).unwrap_err();
+    assert!(error.to_string().contains("out of bounds"));
+}
+
+#[test]
+fn dictionary_framing_rejects_inconsistent_lengths() {
+    let encoding = ProtobufUtils21::dictionary(
+        ProtobufUtils21::flat(32, None),
+        ProtobufUtils21::flat(64, None),
+        2,
+    );
+    let (decoder, has_payload) =
+        create_block_decompressor(&encoding, BlockValueType::UInt64).unwrap();
+    assert!(has_payload);
+    let mut payload = 4_u64.to_le_bytes().to_vec();
+    payload.extend_from_slice(&16_u64.to_le_bytes());
+    payload.extend_from_slice(&[0; 8]);
+    assert!(
+        decoder
+            .decompress(Some(LanceBuffer::from(payload)), 1)
+            .is_err()
+    );
+}
+
+#[test]
+fn dictionary_framing_rejects_payload_for_metadata_child() {
+    let encoding = ProtobufUtils21::dictionary(
+        ProtobufUtils21::flat(32, None),
+        ProtobufUtils21::range(64, 10, 1),
+        2,
+    );
+    let (decoder, has_payload) =
+        create_block_decompressor(&encoding, BlockValueType::UInt64).unwrap();
+    assert!(has_payload);
+    let mut payload = 4_u64.to_le_bytes().to_vec();
+    payload.extend_from_slice(&1_u64.to_le_bytes());
+    payload.extend_from_slice(&0_u32.to_le_bytes());
+    payload.push(0);
+    let error = decoder
+        .decompress(Some(LanceBuffer::from(payload)), 1)
+        .unwrap_err();
+    assert!(error.to_string().contains("Metadata-only Dictionary items"));
+}
+
 #[test]
 fn decoder_allocation_overflow_is_fallible() {
     let (decoder, has_payload) = create_block_decompressor(

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

@@ -10,6 +10,7 @@ pub mod block;
 pub mod byte_stream_split;
 pub mod constant;
 pub mod delta;
+pub mod dictionary;
 pub mod fsst;
 pub mod general;
 pub mod packed;