Skip to content

Pipeline execution #5251

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

Merged
gatesn merged 10 commits into from
Nov 10, 2025
Merged

Pipeline execution #5251

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
be9f62e
pipelined execution
gatesn Nov 7, 2025
c0399c0
pipelined execution
gatesn Nov 7, 2025
70d73cc
pipelined execution
gatesn Nov 7, 2025
332e3fa
pipelined execution
gatesn Nov 7, 2025
d53c68b
pipelined execution
gatesn Nov 7, 2025
f6815c2
pipelined execution
gatesn Nov 8, 2025
cef3f51
pipelined execution
gatesn Nov 8, 2025
fe16812
pipelined execution
gatesn Nov 9, 2025
54ae1e4
pipelined execution
gatesn Nov 10, 2025
7a65acb
pipelined execution
gatesn Nov 10, 2025
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
3 changes: 1 addition & 2 deletions encodings/fastlanes/src/bitpacking/vtable/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@ mod array;
mod canonical;
mod encode;
mod operations;
mod operator;
mod serde;
mod validity;
mod visitor;
Expand All @@ -29,7 +28,7 @@ impl VTable for BitPackedVTable {
type ComputeVTable = NotSupported;
type EncodeVTable = Self;
type SerdeVTable = Self;
type OperatorVTable = Self;
type OperatorVTable = NotSupported;

fn id(_encoding: &Self::Encoding) -> EncodingId {
EncodingId::new_ref("fastlanes.bitpacked")
Expand Down
3 changes: 1 addition & 2 deletions encodings/fastlanes/src/for/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ use vortex_scalar::Scalar;
mod compress;
mod compute;
mod ops;
mod pipeline;
mod serde;

vtable!(FoR);
Expand All @@ -35,7 +34,7 @@ impl VTable for FoRVTable {
type ComputeVTable = NotSupported;
type EncodeVTable = Self;
type SerdeVTable = Self;
type OperatorVTable = Self;
type OperatorVTable = NotSupported;

fn id(_encoding: &Self::Encoding) -> EncodingId {
EncodingId::new_ref("fastlanes.for")
Expand Down
8 changes: 4 additions & 4 deletions encodings/fastlanes/src/for/pipeline.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@ use std::marker::PhantomData;
use std::sync::Arc;

use num_traits::WrappingAdd;
use vortex_array::Array;
use vortex_array::operator::{
LengthBounds, Operator, OperatorEq, OperatorHash, OperatorId, OperatorRef,
};
Expand All @@ -17,8 +16,9 @@ use vortex_array::pipeline::{
BindContext, Element, Kernel, KernelContext, PipelinedOperator, RowSelection, VectorId,
};
use vortex_array::vtable::OperatorVTable;
use vortex_dtype::{DType, NativePType, PType, match_each_integer_ptype};
use vortex_error::{VortexExpect, VortexResult, vortex_bail};
use vortex_array::Array;
use vortex_dtype::{match_each_integer_ptype, DType, NativePType, PType};
use vortex_error::{vortex_bail, VortexExpect, VortexResult};
use vortex_scalar::Scalar;

use crate::{FoRArray, FoRVTable};
Expand Down Expand Up @@ -150,7 +150,7 @@ impl PipelinedOperator for FoROperator {
match_each_integer_ptype!(ptype, |T| {
match_each_integer_ptype!(self.encoded_ptype, |E| {
Ok(Box::new(FoRKernel::<T, E> {
child: ctx.children()[0],
child: ctx.pipelined_input()[0],
reference: self
.reference
.as_primitive()
Expand Down
2 changes: 1 addition & 1 deletion encodings/fsst/src/array.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ impl VTable for FSSTVTable {
type ComputeVTable = NotSupported;
type EncodeVTable = Self;
type SerdeVTable = Self;
type OperatorVTable = Self;
type OperatorVTable = NotSupported;

fn id(_encoding: &Self::Encoding) -> EncodingId {
EncodingId::new_ref("vortex.fsst")
Expand Down
1 change: 0 additions & 1 deletion encodings/fsst/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@ mod array;
mod canonical;
mod compress;
mod compute;
mod operator;
mod ops;
mod serde;
#[cfg(test)]
Expand Down
194 changes: 0 additions & 194 deletions encodings/fsst/src/operator.rs
View file
Open in desktop

This file was deleted.

18 changes: 2 additions & 16 deletions vortex-array/src/array/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,11 @@ use crate::arrays::{
};
use crate::builders::ArrayBuilder;
use crate::compute::{ComputeFn, Cost, InvocationArgs, IsConstantOpts, Output, is_constant_opts};
use crate::operator::OperatorRef;
use crate::serde::ArrayChildren;
use crate::stats::{Precision, Stat, StatsProviderExt, StatsSetRef};
use crate::vtable::{
ArrayVTable, CanonicalVTable, ComputeVTable, OperationsVTable, OperatorVTable, SerdeVTable,
VTable, ValidityVTable, VisitorVTable,
ArrayVTable, CanonicalVTable, ComputeVTable, OperationsVTable, SerdeVTable, VTable,
ValidityVTable, VisitorVTable,
};
use crate::{
ArrayEq, ArrayHash, Canonical, DynArrayEq, DynArrayHash, EncodingId, EncodingRef,
Expand Down Expand Up @@ -168,11 +167,6 @@ pub trait Array:
/// call.
fn invoke(&self, compute_fn: &ComputeFn, args: &InvocationArgs)
-> VortexResult<Option<Output>>;

/// Convert the array to an operator if supported by the encoding.
///
/// Returns `None` if the encoding does not support operator operations.
fn to_operator(&self) -> VortexResult<Option<OperatorRef>>;
}

impl Array for Arc<dyn Array> {
Expand Down Expand Up @@ -275,10 +269,6 @@ impl Array for Arc<dyn Array> {
) -> VortexResult<Option<Output>> {
self.as_ref().invoke(compute_fn, args)
}

fn to_operator(&self) -> VortexResult<Option<OperatorRef>> {
self.as_ref().to_operator()
}
}

/// A reference counted pointer to a dynamic [`Array`] trait object.
Expand Down Expand Up @@ -649,10 +639,6 @@ impl<V: VTable> Array for ArrayAdapter<V> {
) -> VortexResult<Option<Output>> {
<V::ComputeVTable as ComputeVTable<V>>::invoke(&self.0, compute_fn, args)
}

fn to_operator(&self) -> VortexResult<Option<OperatorRef>> {
<V::OperatorVTable as OperatorVTable<V>>::to_operator(&self.0)
}
}

impl<V: VTable> ArrayHash for ArrayAdapter<V> {
Expand Down
13 changes: 11 additions & 2 deletions vortex-array/src/array/operator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,8 @@ use vortex_mask::Mask;
use vortex_vector::{Vector, VectorOps, vector_matches_dtype};

use crate::execution::{BatchKernelRef, BindCtx, DummyExecutionCtx, ExecutionCtx};
use crate::vtable::{OperatorVTable, VTable};
use crate::pipeline::source_driver::PipelineSourceDriver;
use crate::vtable::{OperatorVTable, PipelineNode, VTable};
use crate::{Array, ArrayAdapter, ArrayRef};

/// Array functions as provided by the `OperatorVTable`.
Expand All @@ -21,7 +22,7 @@ pub trait ArrayOperator: 'static + Send + Sync {
/// # Panics
///
/// If the mask length does not match the array length.
/// If the array's implementation returns an invalid vector (wrong length, wrong type, etc).
/// If the array's implementation returns an invalid vector (wrong length, wrong type, etc.).
fn execute_batch(&self, selection: &Mask, ctx: &mut dyn ExecutionCtx) -> VortexResult<Vector>;

/// Optimize the array by running the optimization rules.
Expand Down Expand Up @@ -62,6 +63,14 @@ impl ArrayOperator for Arc<dyn Array> {

impl<V: VTable> ArrayOperator for ArrayAdapter<V> {
fn execute_batch(&self, selection: &Mask, ctx: &mut dyn ExecutionCtx) -> VortexResult<Vector> {
// Check if the array is a pipeline node
if let Some(pipeline_node) =
<V::OperatorVTable as OperatorVTable<V>>::pipeline_node(&self.0)
&& let PipelineNode::Source(source) = pipeline_node
{
return PipelineSourceDriver::new(source).execute(selection);
}

let vector =
<V::OperatorVTable as OperatorVTable<V>>::execute_batch(&self.0, selection, ctx)?;

Expand Down
4 changes: 2 additions & 2 deletions vortex-array/src/arrays/varbin/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,15 @@ mod array;
pub use array::VarBinArray;

mod compute;
pub(crate) use compute::varbin_compute_min_max; // For use in `varbinview`.
pub(crate) use compute::varbin_compute_min_max;
// For use in `varbinview`.

mod vtable;
pub use vtable::{VarBinEncoding, VarBinVTable};

pub mod builder;

mod accessor;
mod operator;

use vortex_buffer::ByteBuffer;
use vortex_dtype::DType;
Expand Down
Loading
Loading