Pipeline execution (#5251)

Signed-off-by: Nicholas Gates <[email protected]>

Author: gatesn

encodings/fastlanes/src/bitpacking/vtable/mod.rs

@@ -10,7 +10,6 @@ mod array;
 mod canonical;
 mod encode;
 mod operations;
-mod operator;
 mod serde;
 mod validity;
 mod visitor;
@@ -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")

encodings/fastlanes/src/for/mod.rs

@@ -18,7 +18,6 @@ use vortex_scalar::Scalar;
 mod compress;
 mod compute;
 mod ops;
-mod pipeline;
 mod serde;
 
 vtable!(FoR);
@@ -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")

encodings/fastlanes/src/for/pipeline.rs

@@ -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,
 };
@@ -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};
@@ -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()

encodings/fsst/src/array.rs

@@ -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")

encodings/fsst/src/lib.rs

@@ -15,7 +15,6 @@ mod array;
 mod canonical;
 mod compress;
 mod compute;
-mod operator;
 mod ops;
 mod serde;
 #[cfg(test)]

encodings/fsst/src/operator.rs

@@ -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,
@@ -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> {
@@ -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.
@@ -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> {

vortex-array/src/array/mod.rs

@@ -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`.
@@ -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.
@@ -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)?;
 

vortex-array/src/array/operator.rs

@@ -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;