feat(encoding): adopt generic offsets in Lance 2.3

docs/src/format/file/encoding.md

@@ -584,10 +584,10 @@ on a per-value basis. We use ☑️ to mark a technique that is applied on a per
 | --------------- | --------------------- | ------------------------ | -------------------------- |
 | Flat            | ✅ (2.1)              | ✅ (2.1)                 | ✅ (2.1)                   |
 | Variable        | ✅ (2.1)              | ✅ (2.1)                 | ✅ (2.1)                   |
-| Constant        | ✅ (2.1)              | ❓                       | ❓                         |
-| Range           | ✅ (2.3)              | ❌                       | ❓                         |
-| Delta           | ✅ (2.3)              | ❌                       | ❓                         |
-| Dictionary      | ✅ (2.3)              | ❌                       | ❓                         |
+| Constant        | ✅ (2.1)              | ❓                       | ✅ (2.3, variable offsets) |
+| Range           | ✅ (2.3)              | ❌                       | ✅ (2.3, variable offsets) |
+| Delta           | ✅ (2.3)              | ❌                       | ✅ (2.3, variable offsets) |
+| Dictionary      | ✅ (2.3)              | ❌                       | ✅ (2.3, variable offsets) |
 | Bitpacking      | ✅ (2.1)              | ❓                       | ✅ (2.1)                   |
 | Fsst            | ❓                    | ✅ (2.1)                 | ✅ (2.1)                   |
 | Rle             | ✅ (2.2)              | ❌                       | ✅ (2.1)                   |
@@ -599,37 +599,61 @@ in various contexts.
 
 ### Generic Block Sequences
 
-Starting in Lance 2.3, block compression can describe unsigned `u32` and `u64` sequences with a bounded,
-zero-or-one-payload codec tree. The containing layout supplies the value type and cardinality.
+Starting in Lance 2.3, block compression can encode unsigned `u32` and `u64` sequences with a shared descriptor
+contract. Direct codecs such as Flat, bitpacking, RLE, and Dictionary also support non-monotonic values; Range and
+Delta require non-decreasing input. The containing layout supplies the value type and cardinality. A descriptor
+constructs a concrete decoder tree whose nodes own their child codecs and framing validation.
 
-`Range` is metadata-only. It stores the unsigned width, first value, and positive step. The value at index `i`
-is `start + step * i`; readers reject cardinalities below two, overflow, and widths other than 32 or 64 bits.
+Writers select a concrete compressor from a bounded set: `Constant`, `Range`, `Flat`, bitpacking, RLE,
+`Dictionary`, `Delta`, and general compression. Candidate costs include the protobuf descriptor, codec framing,
+buffer entries, and alignment. Payload estimates are exact except for `General`, which extrapolates from a bounded
+sample; only the selected compressor is invoked to materialize payloads. Writers use the following canonical
+metadata-only codecs:
+
+- An empty sequence uses `Constant` with no scalar.
+- A non-empty constant sequence uses `Constant` with one little-endian scalar.
+- An arithmetic progression with a positive step uses `Range`.
+
+The first block-dictionary writer only emits Dictionary for `u64` sequences.
+The first block-sequence grammar permits `General` only as the outer root around a `Flat` child; readers reject
+inner, sibling, repeated, or non-`Flat` `General` transforms.
+
+`Range` stores the unsigned width, first value, and positive step. The value at index `i` is
+`start + step * i`; readers reject overflow and widths other than 32 or 64 bits.
 
 ```protobuf
 %%% proto.message.Range %%%
 ```
 
-`Delta` stores the first value inline. Its child describes the `n - 1` non-negative adjacent differences and
-determines whether the Delta tree has a payload. Zero differences are valid. Readers reconstruct the sequence
-with checked prefix sums.
+`Delta` stores the first value inline. Its child represents the `n - 1` non-negative adjacent differences. Delta
+has a payload exactly when its child has one. Zero differences are valid. Readers reconstruct the sequence with
+checked prefix sums.
 
 ```protobuf
 %%% proto.message.Delta %%%
 ```
 
-Block Dictionary stores `u32` indices and typed dictionary items as child codec trees. When at least one child
-has a payload, both children are combined into one outer payload:
+RLE and block Dictionary expose no outer payload when both children are metadata-only. Otherwise, they
+combine their children into one framed outer payload:
 
 ```text
+RLE payload:
+  u64 values_payload_bytes
+  values payload
+  run-lengths payload
+
 Dictionary payload:
   u64 indices_payload_bytes
   u64 items_payload_bytes
   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.
+The framed length for a metadata-only child must be zero. Readers validate child cardinalities, run-length sums,
+dictionary index bounds, and all frame boundaries.
+
+Lance 2.0 through 2.2 writers do not emit `Range`, `Delta`, or block Dictionary. Their block selector order
+and payload shapes remain unchanged.
 
 ### Flat
 
@@ -648,6 +672,25 @@ When applied in a mini-block context each block may have a different number of v
 until we find the point that would exceed 4,096 bytes and then use the most recent power of 2 number of values that
 we have passed.
 
+Lance 2.0 through 2.2 store each mini-block as one legacy buffer containing chunk-local Flat offsets followed by
+the value bytes. Starting in Lance 2.3, the writer also evaluates a generic-offset container. Generic offsets are
+zero-based and independently encoded in each chunk with one page-wide concrete offset codec:
+
+```text
+Legacy chunk:
+  [adjusted Flat offsets][value bytes]
+
+Generic chunk:
+  [optional offset payload][value bytes]
+```
+
+The generic form has one buffer size per chunk for metadata-only offset codecs and two for payload-bearing offset
+codecs. It is selected only when its complete serialized size is strictly smaller than the legacy form. A Flat
+offset descriptor always denotes the legacy form, which keeps the two wire shapes unambiguous. Both forms keep
+offsets and values in the same mini-block chunk, so a random read still fetches only the selected chunk. Fields that
+explicitly request an outer General compressor retain the legacy form because that transform can change the
+complete-container winner after offset selection.
+
 ### Constant
 
 Constant compression is currently only utilized in a few specialized scenarios such as all-null arrays.

rust/lance-encoding/src/compression.rs

@@ -684,9 +684,21 @@ impl DefaultCompressionStrategy {
         let data_size = data.expect_single_stat::<UInt64Type>(Stat::DataSize);
         let max_len = data.expect_single_stat::<UInt64Type>(Stat::MaxLength);
 
-        // Explicitly disable all compression.
+        let build_binary = || {
+            let generic_offsets_own_final_payload =
+                compression.is_none() || compression == Some("none");
+            if self.version.resolve() >= LanceFileVersion::V2_3 && generic_offsets_own_final_payload
+            {
+                BinaryMiniBlockEncoder::with_generic_offsets(params.minichunk_size, params.clone())
+            } else {
+                BinaryMiniBlockEncoder::new(params.minichunk_size)
+            }
+        };
+
+        // "none" disables general compression but still permits structural
+        // offset codecs in 2.3.
         if compression == Some("none") {
-            return Ok(Box::new(BinaryMiniBlockEncoder::new(params.minichunk_size)));
+            return Ok(Box::new(build_binary()));
         }
 
         let use_fsst = compression == Some("fsst")
@@ -699,7 +711,7 @@ impl DefaultCompressionStrategy {
         let mut base_encoder: Box<dyn MiniBlockCompressor> = if use_fsst {
             Box::new(FsstMiniBlockEncoder::new(params.minichunk_size))
         } else {
-            Box::new(BinaryMiniBlockEncoder::new(params.minichunk_size))
+            Box::new(build_binary())
         };
 
         // Wrap with general compression when configured (except FSST / none).
@@ -1887,6 +1899,109 @@ mod tests {
         check_uncompressed_encoding(&encoding, true);
     }
 
+    #[test]
+    fn test_variable_offset_codec_is_version_gated() {
+        let num_values = 2_048_u64;
+        let offsets = (0..=num_values)
+            .map(|index| (index * 3) as i32)
+            .collect::<Vec<_>>();
+        let mut variable = VariableWidthBlock {
+            data: LanceBuffer::from(vec![7_u8; num_values as usize * 3]),
+            offsets: LanceBuffer::reinterpret_vec(offsets),
+            bits_per_offset: 32,
+            num_values,
+            block_info: BlockInfo::default(),
+        };
+        variable.compute_stat();
+        let data = DataBlock::VariableWidth(variable);
+        let field = create_test_field("bytes", DataType::Binary);
+
+        for version in [
+            LanceFileVersion::V2_0,
+            LanceFileVersion::V2_1,
+            LanceFileVersion::V2_2,
+        ] {
+            let strategy = DefaultCompressionStrategy::new().with_version(version);
+            let compressor = strategy.create_miniblock_compressor(&field, &data).unwrap();
+            let (compressed, encoding) = compressor
+                .compress(data.clone(), miniblock_context())
+                .unwrap();
+            let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else {
+                panic!("expected Variable encoding for {version}");
+            };
+            assert!(matches!(
+                variable
+                    .offsets
+                    .as_deref()
+                    .and_then(|offsets| offsets.compression.as_ref()),
+                Some(Compression::Flat(_))
+            ));
+            assert_eq!(compressed.data.len(), 1);
+            assert!(
+                compressed
+                    .chunks
+                    .iter()
+                    .all(|chunk| chunk.buffer_sizes.len() == 1)
+            );
+        }
+
+        let strategy = DefaultCompressionStrategy::new().with_version(LanceFileVersion::V2_3);
+        let compressor = strategy.create_miniblock_compressor(&field, &data).unwrap();
+        let (compressed, encoding) = compressor.compress(data, miniblock_context()).unwrap();
+        let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else {
+            panic!("expected Variable encoding for 2.3");
+        };
+        assert!(matches!(
+            variable
+                .offsets
+                .as_deref()
+                .and_then(|offsets| offsets.compression.as_ref()),
+            Some(Compression::Range(_))
+        ));
+        assert_eq!(compressed.data.len(), 1);
+    }
+
+    #[test]
+    #[cfg(any(feature = "lz4", feature = "zstd"))]
+    fn test_v2_3_explicit_general_compression_keeps_legacy_offsets() {
+        let num_values = 2_048_u64;
+        let offsets = (0..=num_values)
+            .map(|index| (index * 3) as i32)
+            .collect::<Vec<_>>();
+        let mut variable = VariableWidthBlock {
+            data: LanceBuffer::from(vec![7_u8; num_values as usize * 3]),
+            offsets: LanceBuffer::reinterpret_vec(offsets),
+            bits_per_offset: 32,
+            num_values,
+            block_info: BlockInfo::default(),
+        };
+        variable.compute_stat();
+        let data = DataBlock::VariableWidth(variable);
+        let mut field = create_test_field("bytes", DataType::Binary);
+        field.metadata.insert(
+            COMPRESSION_META_KEY.to_string(),
+            if cfg!(feature = "lz4") { "lz4" } else { "zstd" }.to_string(),
+        );
+
+        let strategy = DefaultCompressionStrategy::new().with_version(LanceFileVersion::V2_3);
+        let compressor = strategy.create_miniblock_compressor(&field, &data).unwrap();
+        let (_, encoding) = compressor.compress(data, miniblock_context()).unwrap();
+        let value_encoding = match encoding.compression.as_ref().unwrap() {
+            Compression::General(general) => general.values.as_deref().unwrap(),
+            _ => &encoding,
+        };
+        let Some(Compression::Variable(variable)) = value_encoding.compression.as_ref() else {
+            panic!("expected Variable encoding");
+        };
+        assert!(matches!(
+            variable
+                .offsets
+                .as_deref()
+                .and_then(|offsets| offsets.compression.as_ref()),
+            Some(Compression::Flat(_))
+        ));
+    }
+
     #[test]
     fn test_field_metadata_none_compression() {
         // Prepare field with metadata for none compression