Skip to content

wip: chunked iter exe #7263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
joseph-isaacs wants to merge 3 commits into
base: develop
Choose a base branch
from
Draft

wip: chunked iter exe #7263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
25a2f01
wip: chunked iter exe
joseph-isaacs Apr 2, 2026
c4dd28a
wip: chunked iter exe
joseph-isaacs Apr 2, 2026
d5c96d5
wip
joseph-isaacs Apr 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 12 additions & 12 deletions vortex-array/public-api.lock
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3182,7 +3182,7 @@ pub fn vortex_array::arrays::patched::Patched::child_name(array: vortex_array::A

pub fn vortex_array::arrays::patched::Patched::deserialize(&self, dtype: &vortex_array::dtype::DType, len: usize, metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::patched::Patched::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -3922,7 +3922,7 @@ pub fn vortex_array::arrays::slice::Slice::child_name(array: vortex_array::Array

pub fn vortex_array::arrays::slice::Slice::deserialize(&self, _dtype: &vortex_array::dtype::DType, _len: usize, _metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], _children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::slice::Slice::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -6022,7 +6022,7 @@ pub fn vortex_array::arrays::patched::Patched::child_name(array: vortex_array::A

pub fn vortex_array::arrays::patched::Patched::deserialize(&self, dtype: &vortex_array::dtype::DType, len: usize, metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::patched::Patched::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -6320,7 +6320,7 @@ pub fn vortex_array::arrays::slice::Slice::child_name(array: vortex_array::Array

pub fn vortex_array::arrays::slice::Slice::deserialize(&self, _dtype: &vortex_array::dtype::DType, _len: usize, _metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], _children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::slice::Slice::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -19676,7 +19676,7 @@ pub fn vortex_array::arrays::patched::Patched::child_name(array: vortex_array::A

pub fn vortex_array::arrays::patched::Patched::deserialize(&self, dtype: &vortex_array::dtype::DType, len: usize, metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::patched::Patched::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -19760,7 +19760,7 @@ pub fn vortex_array::arrays::slice::Slice::child_name(array: vortex_array::Array

pub fn vortex_array::arrays::slice::Slice::deserialize(&self, _dtype: &vortex_array::dtype::DType, _len: usize, _metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], _children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::slice::Slice::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -20692,7 +20692,7 @@ pub fn vortex_array::arrays::patched::Patched::child_name(array: vortex_array::A

pub fn vortex_array::arrays::patched::Patched::deserialize(&self, dtype: &vortex_array::dtype::DType, len: usize, metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::patched::Patched::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -20776,7 +20776,7 @@ pub fn vortex_array::arrays::slice::Slice::child_name(array: vortex_array::Array

pub fn vortex_array::arrays::slice::Slice::deserialize(&self, _dtype: &vortex_array::dtype::DType, _len: usize, _metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], _children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::slice::Slice::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -23398,7 +23398,7 @@ pub fn vortex_array::arrays::patched::Patched::child_name(array: vortex_array::A

pub fn vortex_array::arrays::patched::Patched::deserialize(&self, dtype: &vortex_array::dtype::DType, len: usize, metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::patched::Patched::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -23482,7 +23482,7 @@ pub fn vortex_array::arrays::slice::Slice::child_name(array: vortex_array::Array

pub fn vortex_array::arrays::slice::Slice::deserialize(&self, _dtype: &vortex_array::dtype::DType, _len: usize, _metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], _children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::slice::Slice::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -24662,7 +24662,7 @@ pub fn vortex_array::arrays::patched::Patched::child_name(array: vortex_array::A

pub fn vortex_array::arrays::patched::Patched::deserialize(&self, dtype: &vortex_array::dtype::DType, len: usize, metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::patched::Patched::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::patched::Patched::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down Expand Up @@ -24746,7 +24746,7 @@ pub fn vortex_array::arrays::slice::Slice::child_name(array: vortex_array::Array

pub fn vortex_array::arrays::slice::Slice::deserialize(&self, _dtype: &vortex_array::dtype::DType, _len: usize, _metadata: &[u8], _buffers: &[vortex_array::buffer::BufferHandle], _children: &dyn vortex_array::serde::ArrayChildren, _session: &vortex_session::VortexSession) -> vortex_error::VortexResult<vortex_array::ArrayParts<Self>>

pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>
pub fn vortex_array::arrays::slice::Slice::execute(array: vortex_array::Array<Self>, _ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<vortex_array::ExecutionResult>

pub fn vortex_array::arrays::slice::Slice::execute_parent(array: vortex_array::ArrayView<'_, Self>, parent: &vortex_array::ArrayRef, child_idx: usize, ctx: &mut vortex_array::ExecutionCtx) -> vortex_error::VortexResult<core::option::Option<vortex_array::ArrayRef>>

Expand Down
26 changes: 15 additions & 11 deletions vortex-array/src/arrays/chunked/vtable/canonical.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,10 @@ use crate::IntoArray;
use crate::array::ArrayView;
use crate::arrays::Chunked;
use crate::arrays::ChunkedArray;
use crate::arrays::ListView;
use crate::arrays::ListViewArray;
use crate::arrays::PrimitiveArray;
use crate::arrays::Struct;
use crate::arrays::StructArray;
use crate::arrays::chunked::ChunkedArrayExt;
use crate::arrays::listview::ListViewArrayExt;
Expand Down Expand Up @@ -45,7 +47,6 @@ pub(super) fn _canonicalize(
&owned_chunks,
Validity::copy_from_array(array.array())?,
struct_dtype,
ctx,
)?;
Canonical::Struct(struct_array)
}
Expand All @@ -66,24 +67,22 @@ pub(super) fn _canonicalize(
/// Packs many [`StructArray`]s to instead be a single [`StructArray`], where the [`DynArray`] for each
/// field is a [`ChunkedArray`].
///
/// The caller guarantees there are at least 2 chunks.
/// The caller guarantees there are at least 2 chunks, and that all chunks are already
/// canonicalized to [`StructArray`] by iterative execution.
fn pack_struct_chunks(
chunks: &[ArrayRef],
validity: Validity,
struct_dtype: &StructFields,
ctx: &mut ExecutionCtx,
) -> VortexResult<StructArray> {
let len = chunks.iter().map(|chunk| chunk.len()).sum();
let mut field_arrays = Vec::new();

let executed_chunks: Vec<StructArray> = chunks
.iter()
.map(|c| c.clone().execute::<StructArray>(ctx))
.collect::<VortexResult<_>>()?;

for (field_idx, field_dtype) in struct_dtype.fields().enumerate() {
let mut field_chunks = Vec::with_capacity(chunks.len());
for struct_array in &executed_chunks {
for chunk in chunks {
let struct_array = chunk
.as_opt::<Struct>()
.vortex_expect("struct chunk pre-canonicalized by iterative execution");
let field = struct_array.unmasked_field(field_idx).clone();
field_chunks.push(field);
}
Expand All @@ -103,7 +102,8 @@ fn pack_struct_chunks(
///
/// We use the existing arrays (chunks) to form a chunked array of `elements` (the child array).
///
/// The caller guarantees there are at least 2 chunks.
/// The caller guarantees there are at least 2 chunks, and that all chunks are already
/// canonicalized to [`ListViewArray`] by iterative execution.
fn swizzle_list_chunks(
chunks: &[ArrayRef],
validity: Validity,
Expand Down Expand Up @@ -135,7 +135,11 @@ fn swizzle_list_chunks(
let mut sizes = BufferMut::<u64>::with_capacity(len);

for chunk in chunks {
let chunk_array = chunk.clone().execute::<ListViewArray>(ctx)?;
let chunk_array = chunk
.clone()
.try_downcast::<ListView>()
.ok()
.vortex_expect("list chunk pre-canonicalized by iterative execution");
// By rebuilding as zero-copy to `List` and trimming all elements (to prevent gaps), we make
// the final output `ListView` also zero-copyable to `List`.
let chunk_array = chunk_array.rebuild(ListViewRebuildMode::MakeExact)?;
Expand Down
14 changes: 14 additions & 0 deletions vortex-array/src/arrays/chunked/vtable/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ use vortex_error::vortex_err;
use vortex_error::vortex_panic;
use vortex_session::VortexSession;

use crate::AnyCanonical;
use crate::ArrayEq;
use crate::ArrayHash;
use crate::ArrayRef;
Expand Down Expand Up @@ -237,6 +238,19 @@ impl VTable for Chunked {
}

fn execute(array: Array<Self>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
// Iteratively request execution of each chunk until it reaches canonical form.
// This gives the scheduler visibility into child execution and enables
// cross-step optimizations between chunk decoding steps.
for i in 0..array.nchunks() {
if !array.chunk(i).is::<AnyCanonical>() {
return Ok(ExecutionResult::execute_slot::<AnyCanonical>(
array,
i + CHUNKS_OFFSET,
));
}
}

// All chunks are now canonical — combine them.
Ok(ExecutionResult::done(_canonicalize(array.as_view(), ctx)?))
}

Expand Down
12 changes: 9 additions & 3 deletions vortex-array/src/arrays/filter/vtable.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ use vortex_error::vortex_panic;
use vortex_mask::Mask;
use vortex_session::VortexSession;

use crate::AnyCanonical;
use crate::ArrayEq;
use crate::ArrayHash;
use crate::ArrayRef;
Expand All @@ -34,6 +35,7 @@ use crate::buffer::BufferHandle;
use crate::dtype::DType;
use crate::executor::ExecutionCtx;
use crate::executor::ExecutionResult;
use crate::require_child;
use crate::scalar::Scalar;
use crate::serde::ArrayChildren;
use crate::validity::Validity;
Expand Down Expand Up @@ -142,14 +144,18 @@ impl VTable for Filter {
if let Some(canonical) = execute_filter_fast_paths(array.as_view(), ctx)? {
return Ok(ExecutionResult::done(canonical));
}

let array = require_child!(array, array.child(), CHILD_SLOT => AnyCanonical);

let Mask::Values(mask_values) = &array.mask else {
unreachable!("`execute_filter_fast_paths` handles AllTrue and AllFalse")
};

// We rely on the optimization pass that runs prior to this execution for filter pushdown,
// so now we can just execute the filter without worrying.
// Child is pre-canonicalized — apply the filter directly.
debug_assert!(array.child().is_canonical());
let child = array.child().to_canonical()?;
Ok(ExecutionResult::done(
execute_filter(array.child().clone().execute(ctx)?, mask_values).into_array(),
execute_filter(child, mask_values).into_array(),
))
}

Expand Down
6 changes: 5 additions & 1 deletion vortex-array/src/arrays/list/vtable/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,9 +23,11 @@ use crate::array::Array;
use crate::array::ArrayId;
use crate::array::ArrayView;
use crate::array::VTable;
use crate::arrays::Primitive;
use crate::arrays::list::ListArrayExt;
use crate::arrays::list::ListData;
use crate::arrays::list::array::NUM_SLOTS;
use crate::arrays::list::array::OFFSETS_SLOT;
use crate::arrays::list::array::SLOT_NAMES;
use crate::arrays::list::compute::PARENT_KERNELS;
use crate::arrays::list::compute::rules::PARENT_RULES;
Expand All @@ -34,6 +36,7 @@ use crate::buffer::BufferHandle;
use crate::dtype::DType;
use crate::dtype::Nullability;
use crate::dtype::PType;
use crate::require_child;
use crate::serde::ArrayChildren;
use crate::validity::Validity;
use crate::vtable;
Expand Down Expand Up @@ -114,7 +117,7 @@ impl VTable for List {
let elements = slots[crate::arrays::list::array::ELEMENTS_SLOT]
.as_ref()
.vortex_expect("ListArray elements slot");
let offsets = slots[crate::arrays::list::array::OFFSETS_SLOT]
let offsets = slots[OFFSETS_SLOT]
.as_ref()
.vortex_expect("ListArray offsets slot");
vortex_ensure!(
Expand Down Expand Up @@ -180,6 +183,7 @@ impl VTable for List {
}

fn execute(array: Array<Self>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
let array = require_child!(array, array.offsets(), OFFSETS_SLOT => Primitive);
Ok(ExecutionResult::done(
list_view_from_list(array, ctx)?.into_array(),
))
Expand Down
Loading
Loading