Skip to content

Commit a0f96a5

Browse files
committed
list: top-level streaming decomposition behind VORTEX_EXPERIMENTAL_LIST_LAYOUT
Rework ListLayoutStrategy into a streaming transpose driven by the TableStrategy dispatcher (global u64 offsets, memory-bounded), add bounded element reads and is_null pruning routing in ListReader, and gate the whole thing off by default behind the VORTEX_EXPERIMENTAL_LIST_LAYOUT env var / with_list_layout(). Signed-off-by: Matt Katz <mhkatz97@gmail.com>
1 parent 7f69190 commit a0f96a5

6 files changed

Lines changed: 630 additions & 118 deletions

File tree

vortex-file/src/strategy.rs

Lines changed: 26 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -50,10 +50,10 @@ 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;
5453
use vortex_layout::layouts::repartition::RepartitionStrategy;
5554
use vortex_layout::layouts::repartition::RepartitionWriterOptions;
5655
use vortex_layout::layouts::table::TableStrategy;
56+
use vortex_layout::layouts::table::use_experimental_list_layout;
5757
use vortex_layout::layouts::zoned::writer::ZonedLayoutOptions;
5858
use vortex_layout::layouts::zoned::writer::ZonedStrategy;
5959
#[cfg(feature = "unstable_encodings")]
@@ -157,6 +157,10 @@ pub struct WriteStrategyBuilder {
157157
field_writers: HashMap<FieldPath, Arc<dyn LayoutStrategy>>,
158158
allow_encodings: Option<HashSet<ArrayId>>,
159159
flat_strategy: Option<Arc<dyn LayoutStrategy>>,
160+
/// Force list-column decomposition on, overriding the
161+
/// [`use_experimental_list_layout`](vortex_layout::layouts::table::use_experimental_list_layout)
162+
/// env-var default of off.
163+
list_layout: bool,
160164
}
161165

162166
impl Default for WriteStrategyBuilder {
@@ -169,6 +173,7 @@ impl Default for WriteStrategyBuilder {
169173
field_writers: HashMap::new(),
170174
allow_encodings: Some(ALLOWED_ENCODINGS.clone()),
171175
flat_strategy: None,
176+
list_layout: false,
172177
}
173178
}
174179
}
@@ -183,6 +188,14 @@ impl WriteStrategyBuilder {
183188
self
184189
}
185190

191+
/// Enable list-column decomposition, overriding the env-var default of off. When enabled, list
192+
/// columns are written as a `ListLayout` (independently compressed/chunked
193+
/// elements/offsets/validity) rather than as flat arrays.
194+
pub fn with_list_layout(mut self) -> Self {
195+
self.list_layout = true;
196+
self
197+
}
198+
186199
/// Override the write layout for a specific field somewhere in the nested schema tree.
187200
///
188201
/// The field path is matched after the root struct is split into columns. This is useful when a
@@ -243,20 +256,10 @@ impl WriteStrategyBuilder {
243256
Arc::new(FlatLayoutStrategy::default())
244257
};
245258

246-
// 7. for each chunk create a layout. List-typed chunks route through
247-
// `ListLayoutStrategy` (separately-addressable elements/offsets/validity sub-layouts;
248-
// non-list chunks fall through its built-in fallback to `flat`).
249-
let leaf: Arc<dyn LayoutStrategy> = Arc::new(
250-
// Thread the configured `flat` (which carries `allow_encodings` / any custom flat
251-
// override) through every child.
252-
ListLayoutStrategy::default()
253-
.with_elements(Arc::clone(&flat))
254-
.with_offsets(Arc::clone(&flat))
255-
.with_validity(Arc::clone(&flat))
256-
.with_fallback(Arc::clone(&flat)),
257-
);
258-
259-
let chunked = ChunkedLayoutStrategy::new(leaf);
259+
// 7. for each chunk create a flat layout. List columns are decomposed above this point by
260+
// the `TableStrategy` dispatcher (into independently-compressed elements/offsets/validity
261+
// sub-columns), so the per-chunk leaf only ever sees flat, non-list chunks.
262+
let chunked = ChunkedLayoutStrategy::new(Arc::clone(&flat));
260263
// 6. buffer chunks so they end up with closer segment ids physically
261264
let buffered = BufferedStrategy::new(chunked, 2 * ONE_MEG); // 2MB
262265

@@ -334,8 +337,14 @@ impl WriteStrategyBuilder {
334337
let validity_strategy = CollectStrategy::new(compress_then_flat);
335338

336339
// Take any field overrides from the builder and apply them to the final strategy.
337-
let table_strategy = TableStrategy::new(Arc::new(validity_strategy), Arc::new(repartition))
338-
.with_field_writers(self.field_writers);
340+
let mut table_strategy =
341+
TableStrategy::new(Arc::new(validity_strategy), Arc::new(repartition))
342+
.with_field_writers(self.field_writers);
343+
// List decomposition is experimental: enabled by an explicit builder opt-in or the
344+
// `VORTEX_EXPERIMENTAL_LIST_LAYOUT` env var; off otherwise.
345+
if self.list_layout || use_experimental_list_layout() {
346+
table_strategy = table_strategy.with_list_layout();
347+
}
339348

340349
Arc::new(table_strategy)
341350
}

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;
@@ -1668,6 +1669,75 @@ async fn test_writer_with_complex_types() -> VortexResult<()> {
16681669
Ok(())
16691670
}
16701671

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

0 commit comments

Comments
 (0)