feat(encoding): add generic block compression

docs/src/format/file/encoding.md

@@ -584,7 +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)              | ❓                       | ❓                         |
+| 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)                   |
@@ -594,6 +597,64 @@ on a per-value basis. We use ☑️ to mark a technique that is applied on a per
 In the following sections we will describe each technique in a bit more detail and explain how it is utilized
 in various contexts.
 
+### Generic Block Sequences
+
+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.
+
+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 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 %%%
+```
+
+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
+  indices payload
+  dictionary-items payload
+```
+
+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
 
 Flat compression is the uncompressed representation of fixed-width data. There is a single buffer of data
@@ -611,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.

protos/encodings_v2_1.proto

@@ -406,8 +406,9 @@ message Flat {
 // This is a transparent encoding by definition.
 //
 // The input is a variable-width data block.
-// The output is a single fixed-width data block (the offsets) and
-// a single buffer (the values)
+// The output is an optional encoded offset payload and one values buffer.
+// Legacy mini-block Variable encoding instead stores Flat offsets and values
+// interleaved in one buffer.
 message Variable {
   // Describes how the offsets data block is compressed
   CompressiveEncoding offsets = 1;
@@ -426,6 +427,39 @@ message Constant {
   optional bytes value = 1;
 }
 
+// A metadata-only arithmetic sequence of unsigned fixed-width values.
+//
+// The value at index i is `start + step * i`. The container supplies the
+// number of values. Writers use this encoding only when step is positive and
+// the final value fits in the declared bit width.
+//
+// The input is a u32 or u64 fixed-width data block.
+// There is no output buffer.
+message Range {
+  // The width of each decompressed value. Must be 32 or 64.
+  uint64 uncompressed_bits_per_value = 1;
+  // The first value in the sequence.
+  uint64 start = 2;
+  // The positive difference between adjacent values.
+  uint64 step = 3;
+}
+
+// A non-decreasing unsigned fixed-width sequence represented by its first
+// value and the differences between adjacent values.
+//
+// The child contains one fewer value than the input. The container supplies
+// the input cardinality. Delta has a payload exactly when its child has one.
+//
+// The input is a u32 or u64 fixed-width data block.
+message Delta {
+  // The width of each decompressed value. Must be 32 or 64.
+  uint64 uncompressed_bits_per_value = 1;
+  // The first uncompressed value.
+  uint64 base = 2;
+  // Compression applied to the adjacent differences.
+  CompressiveEncoding deltas = 3;
+}
+
 // A compression scheme in which a single fixed-width block is "packed" into
 // a smaller fixed-width block values where each value has fewer bits.
 //
@@ -631,5 +665,7 @@ message CompressiveEncoding {
         FixedSizeList fixed_size_list = 11;
         PackedStruct packed_struct = 12;
         VariablePackedStruct variable_packed_struct = 13;
+        Range range = 14;
+        Delta delta = 15;
     }
 }

rust/lance-encoding/benches/decoder.rs

@@ -2,14 +2,15 @@
 // SPDX-FileCopyrightText: Copyright The Lance Authors
 use std::{collections::HashMap, sync::Arc};
 
-use arrow_array::{RecordBatch, UInt32Array};
+use arrow_array::{RecordBatch, StringArray, UInt32Array};
 use arrow_schema::{DataType, Field, Schema, TimeUnit};
 use arrow_select::take::take;
 use criterion::{Criterion, criterion_group, criterion_main};
 use futures::StreamExt;
 use lance_core::cache::LanceCache;
 use lance_datagen::ArrayGeneratorExt;
 use lance_encoding::{
+    BufferScheduler, EncodingsIo,
     decoder::{
         DecodeBatchScheduler, DecoderConfig, DecoderPlugins, FilterExpression, create_decode_stream,
     },
@@ -563,20 +564,155 @@ fn bench_decode_compressed_parallel(c: &mut Criterion) {
     }
 }
 
+async fn decode_take(
+    encoded: &lance_encoding::encoder::EncodedBatch,
+    indices: &[u64],
+    cache: Arc<LanceCache>,
+) -> RecordBatch {
+    let io_scheduler = Arc::new(BufferScheduler::new(encoded.data.clone())) as Arc<dyn EncodingsIo>;
+    let filter = FilterExpression::no_filter();
+    let mut decode_scheduler = DecodeBatchScheduler::try_new(
+        encoded.schema.as_ref(),
+        &encoded.top_level_columns,
+        &encoded.page_table,
+        &vec![],
+        encoded.num_rows,
+        Arc::<DecoderPlugins>::default(),
+        io_scheduler.clone(),
+        cache,
+        &filter,
+        &DecoderConfig::default(),
+    )
+    .await
+    .unwrap();
+    let (tx, rx) = unbounded_channel();
+    decode_scheduler.schedule_take(indices, &filter, tx, io_scheduler);
+    let mut stream = create_decode_stream(
+        &encoded.schema,
+        indices.len() as u64,
+        indices.len() as u32,
+        true,
+        false,
+        true,
+        rx,
+        None,
+    )
+    .unwrap();
+    stream.next().await.unwrap().task.await.unwrap()
+}
+
+fn bench_variable_offsets_decode(c: &mut Criterion) {
+    const NUM_ROWS: usize = 262_144;
+    const NUM_TAKES: usize = 512;
+
+    let metadata = HashMap::from([
+        (
+            "lance-encoding:structural-encoding".to_string(),
+            "miniblock".to_string(),
+        ),
+        (
+            "lance-encoding:dict-divisor".to_string(),
+            "100000".to_string(),
+        ),
+        ("lance-encoding:compression".to_string(), "none".to_string()),
+    ]);
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("value", DataType::Utf8, false).with_metadata(metadata),
+    ]));
+    let lance_schema = Arc::new(lance_core::datatypes::Schema::try_from(schema.as_ref()).unwrap());
+    let corpora = [
+        (
+            "range",
+            Arc::new(StringArray::from_iter_values(
+                (0..NUM_ROWS).map(|index| format!("row_{index:012}")),
+            )) as Arc<dyn arrow_array::Array>,
+        ),
+        (
+            "delta",
+            Arc::new(StringArray::from_iter_values(
+                (0..NUM_ROWS).map(|index| "x".repeat(4 + index % 64)),
+            )) as Arc<dyn arrow_array::Array>,
+        ),
+    ];
+    let mut take_indices = (0..NUM_TAKES)
+        .map(|index| (index as u64).wrapping_mul(104_729).wrapping_add(8_191) % NUM_ROWS as u64)
+        .collect::<Vec<_>>();
+    take_indices.sort_unstable();
+    take_indices.dedup();
+
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let mut group = c.benchmark_group("variable_offsets_decode");
+
+    for (workload, array) in corpora {
+        let batch = RecordBatch::try_new(schema.clone(), vec![array]).unwrap();
+        for version in [LanceFileVersion::V2_2, LanceFileVersion::V2_3] {
+            let strategy = default_encoding_strategy(version);
+            let options = EncodingOptions {
+                version,
+                ..Default::default()
+            };
+            let encoded = rt
+                .block_on(encode_batch(
+                    &batch,
+                    lance_schema.clone(),
+                    strategy.as_ref(),
+                    &options,
+                ))
+                .unwrap();
+            let encoded_bytes = encoded.data.len();
+            let benchmark_suffix = format!("{workload}/{version}/{encoded_bytes}B");
+
+            group.throughput(criterion::Throughput::Elements(NUM_ROWS as u64));
+            group.bench_function(format!("scan/{benchmark_suffix}"), |bencher| {
+                bencher.iter(|| {
+                    let decoded = rt
+                        .block_on(lance_encoding::decoder::decode_batch(
+                            &encoded,
+                            &FilterExpression::no_filter(),
+                            Arc::<DecoderPlugins>::default(),
+                            false,
+                            version,
+                            Some(Arc::new(LanceCache::no_cache())),
+                        ))
+                        .unwrap();
+                    assert_eq!(decoded.num_rows(), NUM_ROWS);
+                })
+            });
+
+            for cache_mode in ["cold", "warm"] {
+                let cache = if cache_mode == "cold" {
+                    Arc::new(LanceCache::no_cache())
+                } else {
+                    Arc::new(LanceCache::with_capacity(64 * 1024 * 1024))
+                };
+                group.throughput(criterion::Throughput::Elements(take_indices.len() as u64));
+                group.bench_function(format!("take/{benchmark_suffix}/{cache_mode}"), |bencher| {
+                    bencher.iter(|| {
+                        let decoded =
+                            rt.block_on(decode_take(&encoded, &take_indices, cache.clone()));
+                        assert_eq!(decoded.num_rows(), take_indices.len());
+                    })
+                });
+            }
+        }
+    }
+}
+
 #[cfg(target_os = "linux")]
 criterion_group!(
     name=benches;
     config = Criterion::default().significance_level(0.1).sample_size(10)
         .with_profiler(lance_testing::pprof::PProfProfiler::new(100, lance_testing::pprof::Output::Flamegraph(None)));
     targets = bench_decode, bench_decode_fsl, bench_decode_str_with_dict_encoding, bench_decode_packed_struct,
                 bench_decode_str_with_fixed_size_binary_encoding, bench_decode_compressed,
-                bench_decode_compressed_parallel);
+                bench_decode_compressed_parallel, bench_variable_offsets_decode);
 
 // Non-linux version does not support pprof.
 #[cfg(not(target_os = "linux"))]
 criterion_group!(
     name=benches;
     config = Criterion::default().significance_level(0.1).sample_size(10);
     targets = bench_decode, bench_decode_fsl, bench_decode_str_with_dict_encoding, bench_decode_packed_struct,
-                bench_decode_compressed, bench_decode_compressed_parallel);
+                bench_decode_compressed, bench_decode_compressed_parallel,
+                bench_variable_offsets_decode);
 criterion_main!(benches);