Vortex as a storage layer

Vortex is a toolkit for compressed columnar data, with a file format built on top. Its arrays describe physical representations; its layouts organize data for reading from storage. This example illustrates Vortex’s extensibility as a storage layer, not just as a traditional file format.

Illustrative Vortex-style Rust, not a drop-in implementation for a specific release. Helper imports, typed accessors and some registration details are omitted.

STORE AN INVERTED INDEX

Imagine a column with millions of repeated strings:

["London", "Paris", "London", "Tokyo", "London", ...]

Logically, this is an ordinary Utf8[]. Suppose our workload is dominated by city == "London", city IN ("London", "Paris"), and city != "Tokyo". A dictionary encoding avoids storing the strings repeatedly, but queries still conceptually inspect one code per row.

What if the physical representation were an inverted index instead?

InvertedIndex<Utf8>
├── dictionary
│   └── ["London", "Paris", "Tokyo"]
└── postings
    ├── "London" → {0, 2, 4, 7, 9, ...}
    ├── "Paris"  → {1, 8, ...}
    └── "Tokyo"  → {3, 5, 6, ...}

The logical type is still Utf8[]. This is why the separation of logical and physical data types is important.

Define the physical representation

InvertedIndex is a physical encoding of non-null strings. Metadata describes the index; the dictionary, offsets and row IDs are child arrays that can each use their own encoding.

use std::collections::HashMap;
use roaring::RoaringBitmap;
use vortex::{
    Array, ArrayId, ArrayParts, ArrayRef, ArrayView, ExecutionCtx,
    ExecutionResult, IntoArray, VTable, BoolArray, PrimitiveArray,
    VarBinViewArray, DType, Nullability, CompareKernel, CompareOperator,
    ArraySessionExt, VortexResult, vortex_bail, VortexSession, CachedId,
};

#[derive(Clone, Debug)]
pub struct InvertedIndex;

pub type InvertedIndexArray = Array<InvertedIndex>;

// Metadata needed to reconstruct the physical representation.
//
// The actual values are children, not metadata.
//
// That matters because children are themselves arbitrary Vortex arrays.
#[derive(Clone, Debug)]
pub struct InvertedIndexData {
    /// Number of unique values.
    cardinality: usize,
}

// Physical children:
//
//   child 0: dictionary
//
//       VarBinView["London", "Paris", "Tokyo"]
//
//   child 1: posting offsets
//
//       [0, 3, 5, 8]
//
//       London -> postings[0..3]
//       Paris  -> postings[3..5]
//       Tokyo  -> postings[5..8]
//
//   child 2: posting row IDs
//
//       [0, 2, 4, 1, 7, 3, 5, 6]
//
// But here's the important Vortex bit:
//
// These children do NOT have to be canonical arrays.
//
// They could physically become:
//
//   dictionary
//       -> FSST
//
//   offsets
//       -> Delta
//       -> BitPacked
//
//   row_ids
//       -> Delta
//       -> BitPacked
//
// So our custom physical representation is itself composed from
// independently extensible physical representations.

impl InvertedIndex {
    pub fn encode(
        input: ArrayRef,
        ctx: &mut ExecutionCtx,
    ) -> VortexResult<InvertedIndexArray> {
        // This sketch is deliberately non-null. A nullable encoding
        // also needs validity and SQL three-valued comparison semantics.
        ensure_non_null_utf8(&input)?;
        ensure!(input.len() <= u32::MAX as usize);
        let strings =
            input.clone().execute_as::<VarBinViewArray>(ctx)?;

        let mut dictionary = Vec::<String>::new();

        let mut dictionary_ids =
            HashMap::<String, usize>::new();

        let mut postings =
            Vec::<RoaringBitmap>::new();

        // ------------------------------------------------------------
        // Build the inverted index.
        // ------------------------------------------------------------

        for row in 0..strings.len() {
            let value = strings.value_as_str(row);

            let id = match dictionary_ids.get(value) {
                Some(&id) => id,

                None => {
                    let id = dictionary.len();

                    dictionary.push(value.to_owned());
                    dictionary_ids.insert(value.to_owned(), id);
                    postings.push(RoaringBitmap::new());

                    id
                }
            };

            postings[id].insert(row as u32);
        }

        // ------------------------------------------------------------
        // Turn the logical structure into Vortex children.
        // ------------------------------------------------------------

        let dictionary =
            VarBinViewArray::from_iter_str(
                dictionary.iter().map(String::as_str)
            )
            .into_array();

        // Flatten posting lists into:
        //
        //   offsets = [0, 3, 5, 8]
        //   rows    = [0, 2, 4, 1, 7, 3, 5, 6]
        //
        // Real implementation could alternatively make each posting
        // list its own child, or use a custom bitmap child encoding.

        let mut offsets = Vec::<u32>::with_capacity(
            postings.len() + 1
        );

        let mut rows = Vec::<u32>::new();

        offsets.push(0);

        for bitmap in &postings {
            rows.extend(bitmap.iter());

            offsets.push(rows.len() as u32);
        }

        let offsets = offsets.into_array();
        let rows = rows.into_array();

        // ------------------------------------------------------------
        // Now let the normal Vortex compressor optimize our children.
        // ------------------------------------------------------------
        //
        // Maybe:
        //
        //   dictionary -> FSST
        //
        // Maybe:
        //
        //   offsets -> Delta -> BitPacked
        //
        // Maybe:
        //
        //   rows -> Delta -> BitPacked
        //
        // Or maybe three completely new encodings invented next year.
        //
        // InvertedIndex doesn't care. Concatenated lists are not
        // globally sorted: any delta encoding must restart per list,
        // or the compressor must select a different valid encoding.

        let dictionary =
            compress(dictionary, ctx)?;

        let offsets =
            compress(offsets, ctx)?;

        let rows =
            compress(rows, ctx)?;

        // ------------------------------------------------------------
        // CRITICAL:
        //
        // dtype is copied from the original string array.
        //
        // Physical representation changed radically.
        // Logical semantics did not.
        // ------------------------------------------------------------

        Array::try_from_parts(
            ArrayParts::new(
                InvertedIndex,
                input.dtype().clone(),       // still Utf8
                input.len(),
                InvertedIndexData {
                    cardinality: postings.len(),
                },
            )
            .with_slots([
                Some(dictionary),
                Some(offsets),
                Some(rows),
            ]),
        )
    }
}

Serialize and reconstruct

The encoding serializes its own metadata; its children serialize themselves. Canonical execution reconstructs the strings, giving ordinary operations a correctness fallback before any specialized kernels exist.

impl VTable for InvertedIndex {
    type TypedArrayData = InvertedIndexData;

    type OperationsVTable = Self;
    type ValidityVTable = Self;

    fn id(&self) -> ArrayId {
        static ID: CachedId =
            CachedId::new("example.inverted_index");

        *ID
    }

    fn serialize(
        array: ArrayView<'_, Self>,
        _session: &VortexSession,
    ) -> VortexResult<Option<Vec<u8>>> {
        let cardinality =
            array.data().cardinality as u64;

        Ok(Some(
            cardinality
                .to_le_bytes()
                .to_vec()
        ))
    }

    fn deserialize(
        &self,
        dtype: &DType,
        len: usize,
        metadata: &[u8],
        _buffers: &[BufferHandle],
        children: &dyn ArrayChildren,
        _session: &VortexSession,
    ) -> VortexResult<ArrayParts<Self>> {
        let cardinality =
            u64::from_le_bytes(
                metadata.try_into()?
            ) as usize;

        // We know only the LOGICAL requirements of the children here.
        //
        // Their physical encoding is determined recursively by Vortex.

        let dictionary = children.get(
            0,
            &DType::Utf8(Nullability::NonNullable),
            cardinality,
        )?;

        let offsets = children.get(
            1,
            &DType::from(u32::default()),
            cardinality + 1,
        )?;

        let rows = children.get(
            2,
            &DType::from(u32::default()),
            len,
        )?;

        // Validate offsets: zero start, monotonic, final offset == len.
        // Dictionary values are unique; posting lists are sorted and
        // partition all rows exactly once. Reject malformed input.
        validate_inverted_index(dtype, len, &dictionary, &offsets, &rows)?;

        Ok(
            ArrayParts::new(
                InvertedIndex,
                dtype.clone(),
                len,
                InvertedIndexData { cardinality },
            )
            .with_slots([
                Some(dictionary),
                Some(offsets),
                Some(rows),
            ]),
        )
    }

    fn execute(
        array: Array<Self>,
        ctx: &mut ExecutionCtx,
    ) -> VortexResult<ExecutionResult> {
        let dictionary =
            array.dictionary()
                .clone()
                .execute_as::<VarBinViewArray>(ctx)?;

        let offsets =
            array.offsets()
                .clone()
                .execute_as::<PrimitiveArray>(ctx)?;

        let rows =
            array.rows()
                .clone()
                .execute_as::<PrimitiveArray>(ctx)?;

        // Allocate the canonical output only on this fallback path.
        let mut result =
            vec![String::new(); array.len()];

        for value_id in 0..array.data().cardinality {
            let start =
                offsets.value::<u32>(value_id) as usize;

            let end =
                offsets.value::<u32>(value_id + 1) as usize;

            let value =
                dictionary.value_as_str(value_id);

            for i in start..end {
                let row =
                    rows.value::<u32>(i) as usize;

                result[row] = value.to_owned();
            }
        }

        let canonical =
            VarBinViewArray::from_iter_str(
                result.iter().map(String::as_str)
            )
            .into_array();

        Ok(ExecutionResult::done(canonical))
    }
}

Specialize equality

For city = 'London', find the string once in the dictionary and read its posting list. The kernel handles equality and inequality with a constant string; other comparisons can fall back to canonical execution.

impl CompareKernel for InvertedIndex {
    fn compare(
        lhs: ArrayView<'_, Self>,
        rhs: &ArrayRef,
        operator: CompareOperator,
        ctx: &mut ExecutionCtx,
    ) -> VortexResult<Option<ArrayRef>> {
        match operator {
            CompareOperator::Eq |
            CompareOperator::NotEq => {}

            _ => {
                // We haven't optimized <, >, etc.
                //
                // Vortex can fall back instead.
                return Ok(None);
            }
        }

        // This optimized kernel only handles a scalar RHS:
        //
        //     city = "London"

        let Some(needle) =
            rhs.as_constant_string()
        else {
            return Ok(None);
        };

        // ------------------------------------------------------------
        // Search ONLY the dictionary.
        //
        // cardinality might be 200 even if there are 100M rows.
        // ------------------------------------------------------------

        let dictionary =
            lhs.dictionary()
                .clone()
                .execute_as::<VarBinViewArray>(ctx)?;

        let value_id =
            dictionary
                .iter_str()
                .position(|value| value == needle);

        let Some(value_id) = value_id else {
            return Ok(Some(
                match operator {
                    CompareOperator::Eq =>
                        BoolArray::all_false(lhs.len()),

                    CompareOperator::NotEq =>
                        BoolArray::all_true(lhs.len()),

                    _ => unreachable!(),
                }
                .into_array()
            ));
        };

        // ------------------------------------------------------------
        // Locate this value's posting list.
        // ------------------------------------------------------------

        let offsets =
            lhs.offsets()
                .clone()
                .execute_as::<PrimitiveArray>(ctx)?;

        let start =
            offsets.value::<u32>(value_id) as usize;

        let end =
            offsets.value::<u32>(value_id + 1) as usize;

        // Important:
        //
        // We STILL do not decode the entire `rows` child.
        //
        // We slice only the posting list we need.
        //
        // And because rows is itself a Vortex array, slicing can also
        // remain encoding-aware.

        let posting_rows =
            lhs.rows()
                .slice(start..end)?;

        let posting_rows =
            posting_rows
                .execute_as::<PrimitiveArray>(ctx)?;

        // Construct:
        //
        //   false false true false true ...
        //
        // from the posting positions.

        let mut matches =
            MutableBitmap::zeros(lhs.len());

        for row in posting_rows.iter::<u32>() {
            matches.set(row as usize, true);
        }

        if operator == CompareOperator::NotEq {
            matches.invert();
        }

        Ok(Some(
            BoolArray::from_bitmap(matches.freeze())
                .into_array()
        ))
    }
}

Union posting lists

city IN ('London', 'Paris', 'Berlin') becomes a union of posting sets. Posting work scales with matches; a sparse Boolean result can also avoid allocating a dense mask for every row.

fn isin(
    array: &InvertedIndexArray,
    needles: &[&str],
    ctx: &mut ExecutionCtx,
) -> VortexResult<ArrayRef> {
    let dictionary =
        array.dictionary()
            .clone()
            .execute_as::<VarBinViewArray>(ctx)?;

    let offsets =
        array.offsets()
            .clone()
            .execute_as::<PrimitiveArray>(ctx)?;

    // ------------------------------------------------------------
    // Build dictionary lookup ONCE.
    //
    // A production encoding could even persist a hash/FST child
    // instead of scanning the dictionary here.
    // ------------------------------------------------------------

    let mut lookup =
        HashMap::<&str, usize>::new();

    for (id, value) in dictionary.iter_str().enumerate() {
        lookup.insert(value, id);
    }

    let mut matches =
        RoaringBitmap::new();

    for needle in needles {
        let Some(&value_id) =
            lookup.get(needle)
        else {
            continue;
        };

        let start =
            offsets.value::<u32>(value_id) as usize;

        let end =
            offsets.value::<u32>(value_id + 1) as usize;

        let posting =
            array.rows()
                .slice(start..end)?
                .execute_as::<PrimitiveArray>(ctx)?;

        for row in posting.iter::<u32>() {
            matches.insert(row);
        }
    }

    // Convert to whatever boolean representation is optimal.
    //
    // Note that even THIS result doesn't fundamentally have to be a
    // canonical bitmap.
    //
    // Vortex could represent:
    //
    //   Bool[]
    //
    // physically as:
    //
    //   SparseTruePositions
    //
    // if only 0.01% of rows match.

    Ok(
        BooleanFromPositions::new(
            array.len(),
            matches,
        )
        .into_array()
    )
}

Filter with row IDs

A scan can take matching row IDs directly from the index and use them to select rows from other columns. The indexed string column never needs to become canonical for that lookup.

fn rows_equal_to(
    array: &InvertedIndexArray,
    needle: &str,
    ctx: &mut ExecutionCtx,
) -> VortexResult<ArrayRef> {
    let dictionary =
        array.dictionary()
            .clone()
            .execute_as::<VarBinViewArray>(ctx)?;

    let Some(value_id) =
        dictionary
            .iter_str()
            .position(|value| value == needle)
    else {
        return Ok(
            PrimitiveArray::empty_u32().into_array()
        );
    };

    let offsets =
        array.offsets()
            .clone()
            .execute_as::<PrimitiveArray>(ctx)?;

    let start =
        offsets.value::<u32>(value_id) as usize;

    let end =
        offsets.value::<u32>(value_id + 1) as usize;

    // Keep the selection encoded until its consumer needs row IDs.
    Ok(array.rows().slice(start..end)?)
}

Register and use it

Register the encoding and its optional fast paths, then use the same logical column through either posting-based operations or canonical execution.

fn initialize(session: &VortexSession) {
    session.arrays().register(InvertedIndex);
    // Illustrative kernel registration; exact APIs depend on the version.
    session.kernels().register(InvertedIndex);
}

fn main() -> VortexResult<()> {
    let session = VortexSession::default();
    initialize(&session);
    let mut ctx = session.create_execution_ctx();
    let cities = VarBinViewArray::from_iter_str([
        "London", "Paris", "London", "Tokyo", "London",
    ]).into_array();

    let index = InvertedIndex::encode(cities.clone(), &mut ctx)?;
    assert_eq!(index.dtype(), cities.dtype());

    let london = rows_equal_to(&index, "London", &mut ctx)?
        .execute_as::<PrimitiveArray>(&mut ctx)?;
    assert_eq!(london.as_slice::<u32>(), &[0, 2, 4]);

    let _mask = isin(&index, &["London", "Paris"], &mut ctx)?;
    let canonical = index.into_array()
        .execute_as::<VarBinViewArray>(&mut ctx)?;
    assert_eq!(
        canonical.iter_str().collect::<Vec<_>>(),
        vec!["London", "Paris", "London", "Tokyo", "London"],
    );
    Ok(())
}