Skip to content

Commit b51140a

Browse files
mhk197Matthew Katz
andauthored
ListLayout baseline (#8706)
# List Layout Decomposes a `list<T>` column into three independently-written sub-columns — `elements`, `offsets`, and (when nullable) `validity` — each backed by its own layout, under a single `vortex.list` node. ``` vortex.list (ListLayout) ├── elements : <T's layout> element space — one entry per list *element* ├── offsets : primitive u64 row space + 1 — n+1 entries for n lists, monotonic └── validity : bool row space — present iff the list is nullable ``` There are three pieces: the **dispatcher** enables using `ListLayout` for arrays with `DType::List`, the **writer** shreds a list stream into the three child streams, and the **reader** reassembles only the children an expression actually needs. Note that this a prototype and is not a performance improvement over flattening lists... yet. As such, this PR does not yet enable writing `ListLayout` for benchmarks. --- ## 1. `Recursive Writer Dispatch` `TableStrategy` inspects the dtype of the stream it's handed and routes: | dtype | written by | | -------- | ------------------------------------------------------ | | struct | `StructStrategy` | | list | `ListLayoutStrategy` (when list decomposition is on) | | anything | the configured leaf strategy | To enable recursive handling of nested lists (elements are lists) or structs it hands **a descended copy of itself** as the child strategy. So a `list<struct<{ a, list<i32> }>>` recurses with no manual wiring — each level dispatches on its own dtype. List decomposition is gated (`use_experimental_list_layout()` / `VORTEX_EXPERIMENTAL_LIST_LAYOUT`, or an explicit builder call). When off, lists fall through to the leaf strategy unchanged. --- ## 2. `ListLayoutStrategy` A *structural* writer: it does not compress or inspect element dtypes; it shreds a list stream into its three sub-streams and hands each to its own downstream strategy, producing one `ListLayout`. To do so it must: 1. **Canonicalize** each incoming chunk to `List` parts. A gapped/reordered `ListView` is rebuilt into a gapless `List` first (`list_from_list_view`). 2. **Rebase offsets to global `u64`.** Each chunk's local offsets are shifted by `element_base` (the running count of elements emitted so far) and widened to `u64`, and the duplicated boundary offset is dropped on every chunk after the first — so the concatenation of all chunks is one monotonic `[0 … total_elements]` array of length `row_count + 1` that indexes into the single concatenated `elements` child. 3. **Fan out** `elements`, `offsets`, and `validity` onto three bounded (capacity-1) channels; each child is written concurrently by its own strategy via `spawn_nested`. The producer future is joined with the child writers (`try_join`) so a producer error surfaces instead of being hidden as an early channel close. Non-list input is forwarded to a `fallback` strategy unchanged. Because each child is written through an *independent* strategy, the children's physical shape is decoupled: routing `elements` through the recursive dispatcher (→ repartition → chunked) makes it a `ChunkedLayout` chunked in element space, while `offsets`/`validity` can be written with a different (e.g. row-space) strategy. --- ## 3. `ListReader` To read as little as possible, `projection_evaluation` first **classifies the expression** (`get_necessary_list_children`) into the minimal set of children it needs, then routes to a matching read path: | class (`ListChildrenNeeded`) | example expr | reads | path | | ---------------------------- | ------------------------------- | --------------------- | ------------------------- | | `Validity` | `is_null(x)` / `is_not_null(x)` | validity only | `project_validity` | | `OffsetsAndValidity` | `list_length(x)` | offsets + validity | `project_offsets_validity`| | `All` | `x`, or anything over elements | elements + the rest | `project_all` | `project_all` materializes the list, and itself picks between two sub-paths: - **Whole-chunk / concurrent** (`project_all_concurrent`) — used for a full-column range, or when `elements` is a single flat segment (nothing to skip). Fires the `elements`, `offsets`, and `validity` reads concurrently, assembles the list, slices to the requested range, and filters. No offsets→elements ordering dependency. - **Bounded** (`project_all_elements_bounded`) — used for a strict sub-range over **chunked** elements. Reads `offsets[range]`, decodes the first/last offset to bound the elements read to `[first … last)`, fetches only the element chunks that range overlaps, and rebases the offsets to index into the sliced buffer. Costs one offsets→elements round-trip but avoids reading the whole elements column for a partial scan split. ## TODOs: * Add list-related stats for better pruning. * Pruning for element zones. * Correctly handle validity. Right now, validity is written as a flat layout. We also probably do not need a reader. * Better handling of arbitrarily nested lists. * Perf improvement. --------- Signed-off-by: "Matthew Katz" <katz@spiraldb.com> Signed-off-by: "Matt Katz" <mhkatz97@gmail.com> Signed-off-by: Matt Katz <mhkatz97@gmail.com> Co-authored-by: Matthew Katz <katz@spiraldb.com>
1 parent 407720c commit b51140a

10 files changed

Lines changed: 2552 additions & 8 deletions

File tree

vortex-file/src/strategy.rs

Lines changed: 50 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -50,9 +50,11 @@ use vortex_layout::layouts::compressed::CompressingStrategy;
5050
use vortex_layout::layouts::compressed::CompressorPlugin;
5151
use vortex_layout::layouts::dict::writer::DictStrategy;
5252
use vortex_layout::layouts::flat::writer::FlatLayoutStrategy;
53+
use vortex_layout::layouts::list::writer::ListLayoutStrategy;
5354
use vortex_layout::layouts::repartition::RepartitionStrategy;
5455
use vortex_layout::layouts::repartition::RepartitionWriterOptions;
5556
use vortex_layout::layouts::table::TableStrategy;
57+
use vortex_layout::layouts::table::use_experimental_list_layout;
5658
use vortex_layout::layouts::zoned::writer::ZonedLayoutOptions;
5759
use vortex_layout::layouts::zoned::writer::ZonedStrategy;
5860
#[cfg(feature = "unstable_encodings")]
@@ -157,6 +159,10 @@ pub struct WriteStrategyBuilder {
157159
allow_encodings: Option<HashSet<ArrayId>>,
158160
flat_strategy: Option<Arc<dyn LayoutStrategy>>,
159161
probe_compressor: Option<Arc<dyn CompressorPlugin>>,
162+
/// Whether to write list fields using [`ListLayoutStrategy`].
163+
///
164+
/// [`ListLayoutStrategy`]: vortex_layout::layouts::list::writer::ListLayoutStrategy
165+
use_list_layout: bool,
160166
}
161167

162168
impl Default for WriteStrategyBuilder {
@@ -170,6 +176,7 @@ impl Default for WriteStrategyBuilder {
170176
allow_encodings: Some(ALLOWED_ENCODINGS.clone()),
171177
flat_strategy: None,
172178
probe_compressor: None,
179+
use_list_layout: use_experimental_list_layout(),
173180
}
174181
}
175182
}
@@ -184,6 +191,17 @@ impl WriteStrategyBuilder {
184191
self
185192
}
186193

194+
/// Enable writing list fields with [`ListLayoutStrategy`].
195+
///
196+
/// **Note**: this is an unstable and experimental layout that is expected to change.
197+
/// Using it may lead to unreadable files in the future.
198+
///
199+
/// [`ListLayoutStrategy`]: vortex_layout::layouts::list::writer::ListLayoutStrategy
200+
pub fn with_list_layout(mut self) -> Self {
201+
self.use_list_layout = true;
202+
self
203+
}
204+
187205
/// Override the write layout for a specific field somewhere in the nested schema tree.
188206
///
189207
/// The field path is matched after the root struct is split into columns. This is useful when a
@@ -308,12 +326,14 @@ impl WriteStrategyBuilder {
308326
probe_compressor,
309327
);
310328

329+
let row_block_size = NonZeroUsize::new(self.row_block_size).vortex_expect("must be non 0");
330+
311331
// 2. calculate stats for each row group
312332
let stats = ZonedStrategy::new(
313333
dict,
314334
compress_then_flat.clone(),
315335
ZonedLayoutOptions {
316-
block_size: NonZeroUsize::new(self.row_block_size).vortex_expect("must be non 0"),
336+
block_size: row_block_size,
317337
..Default::default()
318338
},
319339
);
@@ -332,11 +352,37 @@ impl WriteStrategyBuilder {
332352
);
333353

334354
// 0. start with splitting columns
335-
let validity_strategy = CollectStrategy::new(compress_then_flat);
355+
let validity_strategy = CollectStrategy::new(compress_then_flat.clone());
336356

337357
// Take any field overrides from the builder and apply them to the final strategy.
338-
let table_strategy = TableStrategy::new(Arc::new(validity_strategy), Arc::new(repartition))
339-
.with_field_writers(self.field_writers);
358+
let mut table_strategy =
359+
TableStrategy::new(Arc::new(validity_strategy), Arc::new(repartition))
360+
.with_field_writers(self.field_writers);
361+
362+
if self.use_list_layout {
363+
// We need a closure here to enable recursive application of list layout.
364+
table_strategy = table_strategy.with_list_layout_factory(
365+
move |list_layout: ListLayoutStrategy| -> Arc<dyn LayoutStrategy> {
366+
let zoned = ZonedStrategy::new(
367+
list_layout,
368+
compress_then_flat.clone(),
369+
ZonedLayoutOptions {
370+
block_size: row_block_size,
371+
..Default::default()
372+
},
373+
);
374+
Arc::new(RepartitionStrategy::new(
375+
zoned,
376+
RepartitionWriterOptions {
377+
block_size_minimum: 0,
378+
block_len_multiple: row_block_size.get(),
379+
block_size_target: None,
380+
canonicalize: false,
381+
},
382+
))
383+
},
384+
);
385+
}
340386

341387
Arc::new(table_strategy)
342388
}

vortex-file/src/tests.rs

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ use vortex_array::ArrayRef;
1515
use vortex_array::IntoArray;
1616
use vortex_array::VortexSessionExecute;
1717
use vortex_array::array_session;
18+
use vortex_array::arrays::BoolArray;
1819
use vortex_array::arrays::ChunkedArray;
1920
use vortex_array::arrays::ConstantArray;
2021
use vortex_array::arrays::DecimalArray;
@@ -1671,6 +1672,75 @@ async fn test_writer_with_complex_types() -> VortexResult<()> {
16711672
Ok(())
16721673
}
16731674

1675+
/// Write `array` with list decomposition forced on (through the full compress/zone pipeline) and
1676+
/// read the whole thing back.
1677+
async fn write_read_roundtrip(array: ArrayRef) -> VortexResult<ArrayRef> {
1678+
let strategy = crate::strategy::WriteStrategyBuilder::default()
1679+
.with_list_layout()
1680+
.build();
1681+
let mut buf = ByteBufferMut::empty();
1682+
SESSION
1683+
.write_options()
1684+
.with_strategy(strategy)
1685+
.write(&mut buf, array.to_array_stream())
1686+
.await?;
1687+
SESSION
1688+
.open_options()
1689+
.open_buffer(buf)?
1690+
.scan()?
1691+
.into_array_stream()?
1692+
.read_all()
1693+
.await
1694+
}
1695+
1696+
/// A `list<list<i32>>` column round-trips through the `TableStrategy` dispatcher, exercising list
1697+
/// decomposition recursing into itself (the outer list's `elements` are themselves lists).
1698+
#[tokio::test]
1699+
#[cfg_attr(miri, ignore)]
1700+
async fn nested_list_of_list_roundtrip() -> VortexResult<()> {
1701+
let inner = ListArray::try_new(
1702+
buffer![1i32, 2, 3, 4, 5, 6].into_array(),
1703+
buffer![0u32, 2, 5, 5, 6].into_array(),
1704+
Validity::NonNullable,
1705+
)?
1706+
.into_array();
1707+
let outer = ListArray::try_new(
1708+
inner,
1709+
buffer![0u32, 2, 4].into_array(),
1710+
Validity::NonNullable,
1711+
)?
1712+
.into_array();
1713+
let st = StructArray::from_fields(&[("nested", outer)])?.into_array();
1714+
1715+
let result = write_read_roundtrip(st.clone()).await?;
1716+
assert_arrays_eq!(result, st, &mut SESSION.create_execution_ctx());
1717+
Ok(())
1718+
}
1719+
1720+
/// A `struct<{ items: list<struct<{a,b}>>? }>` column round-trips, exercising list decomposition
1721+
/// recursing into struct decomposition (list `elements` are structs) plus a nullable list validity
1722+
/// child.
1723+
#[tokio::test]
1724+
#[cfg_attr(miri, ignore)]
1725+
async fn nested_struct_list_struct_roundtrip() -> VortexResult<()> {
1726+
let inner_struct = StructArray::from_fields(&[
1727+
("a", buffer![1i32, 2, 3, 4, 5].into_array()),
1728+
("b", buffer![10i32, 20, 30, 40, 50].into_array()),
1729+
])?
1730+
.into_array();
1731+
let items = ListArray::try_new(
1732+
inner_struct,
1733+
buffer![0u32, 2, 5, 5].into_array(),
1734+
Validity::Array(BoolArray::from_iter([true, false, true]).into_array()),
1735+
)?
1736+
.into_array();
1737+
let st = StructArray::from_fields(&[("items", items)])?.into_array();
1738+
1739+
let result = write_read_roundtrip(st.clone()).await?;
1740+
assert_arrays_eq!(result, st, &mut SESSION.create_execution_ctx());
1741+
Ok(())
1742+
}
1743+
16741744
#[tokio::test]
16751745
async fn test_writer_with_statistics() -> VortexResult<()> {
16761746
let array = StructArray::from_fields(&[("numbers", buffer![1u32, 2, 3, 4, 5].into_array())])?

vortex-layout/src/layouts/chunked/mod.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
// SPDX-License-Identifier: Apache-2.0
22
// SPDX-FileCopyrightText: Copyright the Vortex contributors
33

4-
mod reader;
4+
pub(crate) mod reader;
55
pub mod writer;
66

77
use std::sync::Arc;
Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
// SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
use vortex_array::expr::Expression;
5+
use vortex_array::expr::is_root;
6+
use vortex_array::expr::not;
7+
use vortex_array::expr::root;
8+
use vortex_array::scalar_fn::fns::is_not_null::IsNotNull;
9+
use vortex_array::scalar_fn::fns::is_null::IsNull;
10+
use vortex_array::scalar_fn::fns::list_length::ListLength;
11+
use vortex_error::VortexResult;
12+
13+
/// The minimal set of list children an expression needs for evaluation.
14+
///
15+
/// For example:
16+
/// - `is_null(root())` only needs the validity child.
17+
/// - `list_length(root())` only needs the offsets and validity children.
18+
/// - `root()` needs elements, offsets, and validity children.
19+
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
20+
pub(super) enum ListChildrenNeeded {
21+
/// Only the validity child is needed (`is_null` / `is_not_null`).
22+
Validity,
23+
/// Only the offsets and validity children are needed (`list_length`).
24+
OffsetsAndValidity,
25+
/// All children are needed.
26+
All,
27+
}
28+
29+
/// The minimal set of list children needed to evaluate `expr`, where `root()` is a field with list dtype.
30+
pub(super) fn get_necessary_list_children(expr: &Expression) -> ListChildrenNeeded {
31+
if is_null_root(expr) {
32+
return ListChildrenNeeded::Validity;
33+
}
34+
35+
if is_list_length_root(expr) {
36+
return ListChildrenNeeded::OffsetsAndValidity;
37+
}
38+
39+
if is_root(expr) {
40+
return ListChildrenNeeded::All;
41+
}
42+
43+
// Otherwise the requirement is the max over the operands. Childless expressions that never
44+
// touch the list, such as literals, fall back to the cheapest usable child.
45+
expr.children()
46+
.iter()
47+
.map(get_necessary_list_children)
48+
.max()
49+
.unwrap_or(ListChildrenNeeded::Validity)
50+
}
51+
52+
fn is_null_root(expr: &Expression) -> bool {
53+
(expr.is::<IsNull>() || expr.is::<IsNotNull>())
54+
&& expr.children().len() == 1
55+
&& is_root(expr.child(0))
56+
}
57+
58+
fn is_list_length_root(expr: &Expression) -> bool {
59+
expr.is::<ListLength>() && expr.children().len() == 1 && is_root(expr.child(0))
60+
}
61+
62+
/// Rewrite a validity-class expression so it can be evaluated against the list's validity bool
63+
/// array (`true` == valid row): `is_not_null(root())` becomes `root()` and `is_null(root())`
64+
/// becomes `not(root())`. All other nodes are rebuilt with rewritten children.
65+
pub(super) fn rewrite_validity_expr(expr: &Expression) -> VortexResult<Expression> {
66+
if expr.is::<IsNotNull>() && expr.children().len() == 1 && is_root(expr.child(0)) {
67+
return Ok(root());
68+
}
69+
if expr.is::<IsNull>() && expr.children().len() == 1 && is_root(expr.child(0)) {
70+
return Ok(not(root()));
71+
}
72+
let children = expr
73+
.children()
74+
.iter()
75+
.map(rewrite_validity_expr)
76+
.collect::<VortexResult<Vec<_>>>()?;
77+
expr.clone().with_children(children)
78+
}
79+
80+
/// Rewrite an offsets-class expression so it can be evaluated against an array of list lengths.
81+
/// `list_length(root())` becomes `root()`. Other references to `root()` are left intact: for
82+
/// offsets-class expressions they can only be validity checks, and the lengths array carries the
83+
/// same validity as the original list.
84+
pub(super) fn rewrite_offsets_expr(expr: &Expression) -> VortexResult<Expression> {
85+
if is_list_length_root(expr) {
86+
return Ok(root());
87+
}
88+
89+
let children = expr
90+
.children()
91+
.iter()
92+
.map(rewrite_offsets_expr)
93+
.collect::<VortexResult<Vec<_>>>()?;
94+
expr.clone().with_children(children)
95+
}
96+
97+
#[cfg(test)]
98+
mod tests {
99+
use rstest::rstest;
100+
use vortex_array::dtype::DType;
101+
use vortex_array::dtype::Nullability;
102+
use vortex_array::dtype::PType;
103+
use vortex_array::expr::cast;
104+
use vortex_array::expr::eq;
105+
use vortex_array::expr::gt;
106+
use vortex_array::expr::is_not_null;
107+
use vortex_array::expr::is_null;
108+
use vortex_array::expr::list_length;
109+
use vortex_array::expr::lit;
110+
use vortex_array::expr::not;
111+
use vortex_array::expr::root;
112+
113+
use super::*;
114+
115+
/// `get_necessary_list_children` keys off the deepest list child an expression touches; `All`
116+
/// is the always-correct default for anything not specifically recognized.
117+
#[rstest]
118+
// `is_null` / `is_not_null` of the list itself need only validity.
119+
#[case::is_null(is_null(root()), ListChildrenNeeded::Validity)]
120+
#[case::is_not_null(is_not_null(root()), ListChildrenNeeded::Validity)]
121+
// Compound over validity-only operands stays validity.
122+
#[case::not_is_null(not(is_null(root())), ListChildrenNeeded::Validity)]
123+
// A list-independent (constant) expression falls to the cheapest usable child.
124+
#[case::constant(lit(5), ListChildrenNeeded::Validity)]
125+
// `list_length(root())` needs offsets and validity, but not elements.
126+
#[case::list_length(list_length(root()), ListChildrenNeeded::OffsetsAndValidity)]
127+
// Compound over offsets-only operands stays offsets.
128+
#[case::list_length_filter(
129+
gt(list_length(root()), lit(1u64)),
130+
ListChildrenNeeded::OffsetsAndValidity
131+
)]
132+
#[case::cast_list_length(
133+
cast(
134+
list_length(root()),
135+
DType::Primitive(PType::I64, Nullability::Nullable),
136+
),
137+
ListChildrenNeeded::OffsetsAndValidity
138+
)]
139+
// A bare list reference needs the elements.
140+
#[case::bare_root(root(), ListChildrenNeeded::All)]
141+
// Any other fn over the list needs the elements.
142+
#[case::not_root(not(root()), ListChildrenNeeded::All)]
143+
// `is_null` only short-circuits to validity when its argument is the list itself.
144+
#[case::is_null_of_derived(is_null(not(root())), ListChildrenNeeded::All)]
145+
// Max over operands: validity + elements => elements.
146+
#[case::validity_and_elements(eq(is_null(root()), root()), ListChildrenNeeded::All)]
147+
fn classify_expr_class(#[case] expr: Expression, #[case] expected: ListChildrenNeeded) {
148+
assert_eq!(get_necessary_list_children(&expr), expected);
149+
}
150+
}

0 commit comments

Comments
 (0)