Pluggable Compressor (#7018)

## Summary: Extensible and Pluggable Compressor

Tracking Issue: #7216

You can see a lot of the details in the tracking issue.

This is a major step in supporting extension types as a first-class
feature in Vortex. The entire compressor has been rewritten, see the
tracking issue for full design details and motivation.

The new `vortex-compressor` crate extracts the encoding-agnostic
compression framework from `vortex-btrblocks`, inverting the dependency
graph so that encoding crates can implement a single `Scheme` trait and
register themselves with the compressor. Additionally,
`vortex-btrblocks` remains the "batteries-included" default compressor,
and depends on `vortex-compressor`.

---

The compression benchmark comment is
[here](#7018 (comment)).

For reviewers: I would just look at the whole `vortex-compressor` and
`vortex-btrblocks` crates instead of the git diff since basically
everything has changed.

## Changes

- [x] Extract `vortex-compressor` crate with unified `Scheme` trait
- [x] Migrate `vortex-btrblocks` to depend on `vortex-compressor`
- [ ] Verify what APIs we want to preserve from `vortex-btrblocks`
(re-exports)
- [ ] Figure out why the `RunEndScheme` exclusion in `rle.rs` is broken
and re-enable it

## Testing

Existing tests pass, so that's a good sign. I added a few new tests that
check the newer parts of the compressor as well.

---------

Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: connortsui20

Cargo.lock

@@ -12,6 +12,7 @@ members = [
     "vortex-proto",
     "vortex-array",
     "vortex-tensor",
+    "vortex-compressor",
     "vortex-btrblocks",
     "vortex-layout",
     "vortex-scan",
@@ -264,6 +265,7 @@ vortex-array = { version = "0.1.0", path = "./vortex-array", default-features =
 vortex-btrblocks = { version = "0.1.0", path = "./vortex-btrblocks", default-features = false }
 vortex-buffer = { version = "0.1.0", path = "./vortex-buffer", default-features = false }
 vortex-bytebool = { version = "0.1.0", path = "./encodings/bytebool", default-features = false }
+vortex-compressor = { version = "0.1.0", path = "./vortex-compressor", default-features = false }
 vortex-datafusion = { version = "0.1.0", path = "./vortex-datafusion", default-features = false }
 vortex-datetime-parts = { version = "0.1.0", path = "./encodings/datetime-parts", default-features = false }
 vortex-decimal-byte-parts = { version = "0.1.0", path = "encodings/decimal-byte-parts", default-features = false }

Cargo.toml

@@ -61,9 +61,10 @@ use vortex_array::search_sorted::SearchSorted;
 use vortex_array::search_sorted::SearchSortedSide;
 use vortex_btrblocks::BtrBlocksCompressor;
 use vortex_btrblocks::BtrBlocksCompressorBuilder;
-use vortex_btrblocks::FloatCode;
-use vortex_btrblocks::IntCode;
-use vortex_btrblocks::StringCode;
+use vortex_btrblocks::SchemeExt;
+use vortex_btrblocks::schemes::float;
+use vortex_btrblocks::schemes::integer;
+use vortex_btrblocks::schemes::string;
 use vortex_error::VortexExpect;
 use vortex_error::vortex_panic;
 use vortex_mask::Mask;
@@ -546,9 +547,11 @@ pub fn compress_array(array: &ArrayRef, strategy: CompressorStrategy) -> ArrayRe
             .compress(array)
             .vortex_expect("BtrBlocksCompressor compress should succeed in fuzz test"),
         CompressorStrategy::Compact => BtrBlocksCompressorBuilder::default()
-            .include_string([StringCode::Zstd])
-            .include_int([IntCode::Pco])
-            .include_float([FloatCode::Pco])
+            .include([
+                string::ZstdScheme.id(),
+                integer::PcoScheme.id(),
+                float::PcoScheme.id(),
+            ])
             .build()
             .compress(array)
             .vortex_expect("Compact compress should succeed in fuzz test"),

fuzz/src/array/mod.rs

@@ -24250,6 +24250,14 @@ pub fn vortex_array::ExecutionCtx::new(session: vortex_session::VortexSession) -
 
 pub fn vortex_array::ExecutionCtx::session(&self) -> &vortex_session::VortexSession
 
+impl core::clone::Clone for vortex_array::ExecutionCtx
+
+pub fn vortex_array::ExecutionCtx::clone(&self) -> vortex_array::ExecutionCtx
+
+impl core::fmt::Debug for vortex_array::ExecutionCtx
+
+pub fn vortex_array::ExecutionCtx::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+
 impl core::fmt::Display for vortex_array::ExecutionCtx
 
 pub fn vortex_array::ExecutionCtx::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result

vortex-array/public-api.lock

@@ -190,6 +190,7 @@ impl dyn DynArray + '_ {
 ///
 /// Accumulates a trace of execution steps. Individual steps are logged at TRACE level for
 /// real-time following, and the full trace is dumped at DEBUG level when the context is dropped.
+#[derive(Debug, Clone)]
 pub struct ExecutionCtx {
     id: usize,
     session: VortexSession,

vortex-array/src/executor.rs

@@ -14,7 +14,6 @@ rust-version = { workspace = true }
 version = { workspace = true }
 
 [dependencies]
-enum-iterator = { workspace = true }
 getrandom_v03 = { workspace = true }
 itertools = { workspace = true }
 num-traits = { workspace = true }
@@ -25,6 +24,7 @@ tracing = { workspace = true }
 vortex-alp = { workspace = true }
 vortex-array = { workspace = true }
 vortex-buffer = { workspace = true }
+vortex-compressor = { workspace = true }
 vortex-datetime-parts = { workspace = true }
 vortex-decimal-byte-parts = { workspace = true }
 vortex-error = { workspace = true }

vortex-btrblocks/Cargo.toml

@@ -9,7 +9,6 @@ use vortex_array::arrays::BoolArray;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::builders::dict::dict_encode;
 use vortex_array::validity::Validity;
-use vortex_btrblocks::CompressorStats;
 use vortex_btrblocks::IntegerStats;
 use vortex_btrblocks::integer_dictionary_encode;
 use vortex_buffer::BufferMut;

vortex-btrblocks/benches/dict_encode.rs

@@ -10,7 +10,6 @@ mod benchmarks {
     use divan::Bencher;
     use vortex_array::arrays::PrimitiveArray;
     use vortex_array::validity::Validity;
-    use vortex_btrblocks::CompressorStats;
     use vortex_btrblocks::GenerateStatsOptions;
     use vortex_btrblocks::IntegerStats;
     use vortex_buffer::Buffer;