Opensource internals

The Arrow Row Format: How DataFusion Packs a Multi-Column Sort Key Into One Comparable Byte String

11 min read Jay Hubbert

DataFusion converts columnar sort keys into Arrow's order-preserving row format so a multi-column comparison becomes one memcmp

TL;DR: Arrow’s RowConverter turns N Arrow columns into one flat byte buffer per row, using an order-preserving encoding: signed ints get their sign bit flipped, floats become their IEEE-754 totalOrder integer image, strings are written in self-delimiting blocks, and each field starts with a null sentinel. Because the encoding is order-preserving, memcmp on two concatenated rows is the lexicographic comparison of the original columns. DataFusion uses this in the sort merge path, in range partitioning, in the TopK heap, and in the generic GROUP BY fallback — everywhere a multi-column key has to be compared against itself millions of times.

The problem: comparing a tuple of columns, over and over

Sorting is the canonical case. ORDER BY country, city, population produces a composite key. The naive implementation keeps the three columns as three Arrow arrays and compares them pairwise:

text 
compare(a, b):
    if a.country != b.country: return a.country < b.country
    if a.city    != b.city:    return a.city    < b.city
    return a.population < b.population

That is fine for one comparison. A sort does O(N log N) of them, and a merge of K sorted runs does O(N log K) more. With 100 million rows and a 3-column key, you are looking at billions of tuple comparisons, each of which has to branch on “which column are we in”, re-dispatch on the runtime type of that column, and handle the possibility that the column is null.

The classic fix in the database world is the “tuple” or “serialized key” representation: pack the whole composite key into one opaque byte string so that comparing keys is a single memcmp. The trouble is doing it without losing information. If you just concatenate the raw bytes, ("a", "bc") and ("ab", "c") become the same bytes. The encoding has to be injective and order-preserving at the same time.

What Arrow’s row format actually looks like

Arrow’s arrow-row crate (vendored here as arrow-row-60.0.0) defines exactly such an encoding. The format is documented at length in arrow-row-60.0.0/src/lib.rs:199-350, and the interesting parts are:

Signed integers get their sign bit flipped and are written big-endian, so the byte sequence is monotonically increasing in the numeric value:

text 
// arrow-row-60.0.0/src/lib.rs:225-241
/// Signed integers have their most significant sign bit flipped, and are then encoded in the
/// same manner as an unsigned integer.
///
/// ```text
///        ┌──┬──┬──┬──┐       ┌──┬──┬──┬──┐       ┌──┬──┬──┬──┬──┐
///     5  │05│00│00│00│       │05│00│00│80│       │01│80│00│00│05│
///    -5  │FB│FF│FF│FF│       │FB│FF│FF│7F│       │01│7F│FF│FF│FB│

Nulls are encoded as a leading sentinel byte, 0x00 for nulls-first and 0xFF for nulls-last, with a valid value starting with 0x01:

text 
// arrow-row-60.0.0/src/fixed.rs:224-241
    let null_sentinel = null_sentinel(opts);
    for (value_idx, is_valid) in nulls.iter().enumerate() {
        ...
        if is_valid {
            to_write[0] = 1;
            ...
        } else {
            data[*offset] = null_sentinel;
        }

Variable-length strings use a COBS-inspired block scheme so the decoder never has to scan for a terminator:

text 
// arrow-row-60.0.0/src/variable.rs:82-102
/// - single `0_u8` if null
/// - single `1_u8` if empty array
/// - `2_u8` if not empty, followed by one or more blocks
/// where a block is encoded as
/// - [`BLOCK_SIZE`] bytes of string data, padded with 0s
/// - `0xFF_u8` if this is not the last block for this string
/// - otherwise the length of the block as a `u8`

The first four blocks are only 8 bytes wide, to avoid space amplification for short strings (variable.rs:30-38).

Descending order is obtained by negating the encoded bytes of the non-null payload — not by reversing the comparator:

text 
// arrow-row-60.0.0/src/lib.rs:561-563
/// ## Reverse Column Ordering
///
/// The order of a given column can be reversed by negating the encoded bytes of non-null values

This is the crucial design decision. Because the same byte-comparison code handles ASC and DESC, the merge loop needs no special cases, and the sort direction becomes a property of the encoding, not of the comparison.

Floats are converted to their IEEE-754 totalOrder integer image, so -0.0 sorts before +0.0 and NaNs have a defined position. This is not the same as SQL’s = semantics — DataFusion has to normalize -0.0 to +0.0 explicitly wherever it uses the row format for equality, which it does in the GROUP BY fallback:

rust 
// datafusion/physical-plan/src/aggregates/group_values/row.rs:124-130
    fn intern(&mut self, cols: &[ArrayRef], groups: &mut Vec<usize>) -> Result<()> {
        // Normalize -0.0 → +0.0 so RowConverter (IEEE 754 totalOrder) and
        // primitive hashing both group ±0 together. No-op for non-float
        // columns.
        let normalized_cols: Vec<ArrayRef> =
            cols.iter().map(normalize_float_zero).collect();

The payoff: one comparison, no dispatch

The Rows type is a flat Vec<u8> plus a Vec<usize> of row offsets:

rust 
// arrow-row-60.0.0/src/lib.rs:1331-1341
pub struct Rows {
    /// Underlying row bytes
    buffer: Vec<u8>,
    /// Row `i` has data `&buffer[offsets[i]..offsets[i+1]]`
    offsets: Vec<usize>,
    /// The config for these rows
    config: RowConfig,
}

and comparing two rows is literally slice::cmp:

rust 
// arrow-row-60.0.0/src/lib.rs:1593-1599
impl Ord for Row<'_> {
    #[inline]
    fn cmp(&self, other: &Self) -> Ordering {
        self.data.cmp(other.data)
    }
}

memcmp on two byte strings. No type dispatch, no null branch, no “which column are we in” branch. That is the entire trick.

Where DataFusion uses it

1. The sort merge path

RowCursorStream is the multi-column path of the sort-preserving merge. It builds one RowConverter for the whole ordering and appends each batch’s key columns into a reusable Rows buffer:

rust 
// datafusion/physical-plan/src/sorts/stream.rs:191-201
        let sort_fields = expressions
            .iter()
            .map(|expr| {
                let data_type = expr.expr.data_type(schema)?;
                Ok(SortField::new_with_options(data_type, expr.options))
            })
            .collect::<Result<Vec<_>>>()?;

        let streams: Vec<_> = streams.into_iter().map(|s| s.fuse()).collect();
        let stream_count = streams.len();
        let converter = RowConverter::new(sort_fields)?;
rust 
// datafusion/physical-plan/src/sorts/stream.rs:216-240
    fn convert_batch(
        &mut self,
        batch: &RecordBatch,
        stream_idx: usize,
    ) -> Result<RowValues> {
        let cols = evaluate_expressions_to_arrays(&self.column_expressions, batch)?;

        // At this point, ownership should of this Rows should be unique
        let mut rows = self.rows.take_next(stream_idx, &self.converter)?;

        rows.clear();

        self.converter.append(&mut rows, &cols)?;
        self.reservation.try_resize(self.converter.size())?;
        ...

Note that the sort itself does not use the row format — sort_batch uses Arrow’s lexsort_to_indices (datafusion/physical-plan/src/sorts/sort.rs:1097-1116). The row format is for the merge, where the alternative is comparing two streams of differently-shaped columns against each other, repeatedly.

The RowValues cursor goes one step further and caches the raw pointer and length of the current row, so the hot comparison is a pointer dereference plus memcmp rather than two offset lookups:

rust 
// datafusion/physical-plan/src/sorts/cursor.rs:293-317
    fn compare(l: &Self, l_idx: usize, r: &Self, r_idx: usize) -> Ordering {
        ...
        let _ = (l_idx, r_idx);
        l.current_slice().cmp(r.current_slice())
    }

Interestingly, the comment above it says the compiler’s own inlining choice wins here:

rust 
// datafusion/physical-plan/src/sorts/cursor.rs:279-281
    // No inline hint on purpose: for the heavyweight `Rows` byte comparison the
    // compiler's own choice wins — both `#[inline]` and `#[inline(never)]`
    // measurably regress the multi-column merge path.

2. Range partitioning

The range-partition router needs to answer “which of the N split points does this row fall after?” for an arbitrary composite key. The trick is to encode the split points into the row format once, at construction time, and then binary-search the encoded rows:

rust 
// datafusion/physical-plan/src/repartition/range.rs:543-555
        let split_point_rows = if split_points.is_empty() {
            None
        } else {
            let split_point_arrays = (0..num_cols)
                .map(|col_idx| {
                    let col_scalars =
                        split_points.iter().map(|sp| sp.values()[col_idx].clone());
                    ScalarValue::iter_to_array(col_scalars)
                })
                .collect::<Result<Vec<_>>>()?;

            Some(row_converter.convert_columns(&split_point_arrays)?)
        };
rust 
// datafusion/physical-plan/src/repartition/range.rs:583-596
#[inline]
fn partition_point_rows(sp_rows: &Rows, row: &Row<'_>) -> usize {
    let mut left = 0;
    let mut right = sp_rows.num_rows();
    while left < right {
        let mid = left + (right - left) / 2;
        if sp_rows.row(mid) <= *row {
            left = mid + 1;
        } else {
            right = mid;
        }
    }
    left
}

A composite-key range lookup is now O(log N) memcmps. This is a genuine reuse: the sort path and the shuffle path share one encoding, so a query that both sorts and range-shuffles does not pay two different serialization costs.

3. The TopK heap

TopK keeps a bounded heap of the best k rows. The heap’s comparison key is the row-format bytes, so the heap is a BinaryHeap<Row> and the eviction test is a byte compare:

rust 
// datafusion/physical-plan/src/topk/mod.rs:515-535
    fn find_new_topk_items(
        &mut self,
        items: impl Iterator<Item = usize>,
        batch_entry: &mut RecordBatchEntry,
    ) -> usize {
        let mut replacements = 0;
        let rows = &mut self.scratch_rows;
        for (index, row) in items.zip(rows.iter()) {
            match self.heap.max() {
                // heap has k items, and the new row is greater (by byte order,
                // after row conversion) than the top K ==> it is not a new topk
                Some(max_row) if row.as_ref() >= max_row.row() => {}
                // don't yet have k items or new item is lower than the currently k low values
                None | Some(_) => {
                    self.heap.add(batch_entry, row, index);
                    replacements += 1;
                }
            }
        }
        replacements
    }

TopK also keeps a second converter for the common sort prefix between the input ordering and the TopK ordering, so it can early-exit the whole operator when the input has moved past the heap boundary (datafusion/physical-plan/src/topk/mod.rs:393-398 and :708-784). That early exit is only expressible because “past the boundary” is a byte comparison on a prefix row.

4. The generic GROUP BY fallback

GroupValuesRows is the fallback used for nested or unsupported group key types. It interns keys by hashing the columnar input (so the hash is computed on the cheap typed representation) and then uses the row format only for the equality check:

rust 
// datafusion/physical-plan/src/aggregates/group_values/row.rs:152-163
        for (row, &target_hash) in batch_hashes.iter().enumerate() {
            let entry = self.map.find_mut(target_hash, |(exist_hash, group_idx)| {
                // Somewhat surprisingly, this closure can be called even if the
                // hash doesn't match, so check the hash first with an integer
                // comparison first avoid the more expensive comparison with
                // group value. https://github.com/apache/datafusion/pull/11718
                target_hash == *exist_hash
                    // verify that the group that we are inserting with hash is
                    // actually the same key value as the group in
                    // existing_idx  (aka group_values @ row)
                    && group_rows.row(row) == group_values.row(*group_idx)
            });

So the hash is over typed columns, and only the tie-break on a hash collision is a memcmp of two row buffers. That is the right split: hashing is O(key width) per row either way, but the equality check only runs on collisions, which are rare.

Why not just use a struct of arrays everywhere?

Because the row format has real costs.

Space. Every row pays a per-field sentinel byte, plus big-endian padding for integers, plus the block padding for strings. A 3-column key of (Utf8, Int32, Utf8) costs roughly 1 + len(country) + 1 + 4 + 1 + len(city) bytes plus block padding, versus the “natural” 4 + 4 + 8 for a tight columnar layout. For a narrow key it is a wash; for a wide string key it can be 2x.

Encode cost. You must serialize every row before you can compare any of them. If the query only needs the first column for comparison (a common case: ORDER BY a with b as a tiebreaker that rarely matters), the row format does strictly more work than a single-column compare. This is why DataFusion keeps a specialized single-column path:

rust 
// datafusion/physical-plan/src/sorts/streaming_merge.rs:279-292
        // Special case single column comparisons with optimized cursor implementations
        if expressions.len() == 1 {
            let sort = expressions[0].clone();
            let data_type = sort.expr.data_type(schema.as_ref())?;
            downcast_primitive! {
                data_type => (primitive_merge_helper, ...),
                DataType::Utf8 => merge_helper!(StringArray, ...),
                ...
            }
        }

Lossy for some types. The row format hydrates dictionaries and run-end-encoded arrays to their plain value type during encoding (arrow-row-60.0.0/src/lib.rs:305-312). DataFusion has to re-encode the group keys on the way out to restore the declared schema type — see encode_array_if_necessary in datafusion/physical-plan/src/aggregates/group_values/row.rs:329-464, which walks nested list/map/struct types and re-applies dictionary encoding to each child. That is a nontrivial amount of code that exists purely to undo a lossy step in the encoding.

Type-stability. The format is explicitly documented as unstable:

rust 
// arrow-row-60.0.0/src/lib.rs:187
/// *Note: The encoding of the row format may change from release to release.*

Rows produced by one RowConverter can only be compared with rows produced by the same converter (arrow-row-60.0.0/src/lib.rs:1041-1045 asserts this with Arc::ptr_eq). DataFusion therefore has to plumb a single converter through the whole operator lifetime and never let two of them meet.

Failure modes to watch for

  • Comparing rows from different converters. The Arc::ptr_eq assertion in RowConverter::append catches the common case, but a Row that outlives its converter’s fields is a use-after-free hazard in principle. DataFusion sidesteps this by keeping the converter alive in the same struct as the rows.
  • Float totalOrder vs SQL equality. The row format says -0.0 and +0.0 are different; SQL says they are equal. Any use of the row format for equality must normalize first, exactly as GroupValuesRows::intern does.
  • Null ordering drift. SortField::new_with_options carries nulls_first, and the encoding flips the null sentinel accordingly. Getting the direction wrong silently reorders nulls. DataFusion’s build_sort_fields threads the PhysicalSortExpr’s options straight through, which is the right thing to do but leaves no room for a mistake.
  • Memory accounting. Rows holds a Vec<u8> that grows to the high water mark of the batch. DataFusion’s ReusableRows in datafusion/physical-plan/src/sorts/stream.rs:97-162 keeps one buffer per partition alive and re-reserves it against the memory pool, precisely because the retained capacity is what the pool is actually charged for.

The takeaway

The row format is a serialization format with a comparison contract. By making the encoding order-preserving and self-deliminating at the same time, Arrow gets a key type whose comparison is memcmp — no type dispatch, no null branches, no per-column loop. DataFusion leans on this everywhere a composite key is compared repeatedly: merges, range routing, bounded heaps, and the generic grouping fallback. The price is encode cost, space, and a documented-unstable wire format — which is why the single column fast path still exists, and why the dictionary flattening in the encoder has to be undone by hand on the way out.

DatafusionArrowPerformanceInternals