16204612549

Committed 10 Jul 2025 07:50PM UTC coverage: 81.152% (+2.9%) from 78.263%

Build # 16204612549

Build Type

Pull #3825

github

Committed by

web-flow

Commit Message

Merge d0d2717da into be9c2fd3e

Pull Request Pull Request #3825: feat: Add optimize ArrayOp with VBView implementation

Run Details

178 of 211 new or added lines in 4 files covered. (84.36%)

330 existing lines in 34 files now uncovered.

45433 of 55985 relevant lines covered (81.15%)

145951.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.5

/vortex-layout/src/reader.rs

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright the Vortex contributors

use std::collections::BTreeSet;
use std::ops::Range;
use std::sync::Arc;

use async_trait::async_trait;
use futures::FutureExt;
use futures::future::{BoxFuture, Shared};
use once_cell::sync::OnceCell;
use vortex_array::stats::Precision;
use vortex_array::{ArrayContext, ArrayRef};
use vortex_dtype::{DType, FieldMask};
use vortex_error::{SharedVortexResult, VortexError, VortexResult, vortex_bail};
use vortex_expr::ExprRef;
use vortex_mask::Mask;

use crate::children::LayoutChildren;
use crate::segments::SegmentSource;

pub type LayoutReaderRef = Arc<dyn LayoutReader>;

/// A [`LayoutReader`] is used to read a [`crate::Layout`] in a way that can cache state across multiple
/// evaluation operations.
pub trait LayoutReader: 'static + Send + Sync {
    /// Returns the name of the layout reader for debugging.
    fn name(&self) -> &Arc<str>;

    /// Returns the un-projected dtype of the layout reader.
    fn dtype(&self) -> &DType;

    /// Returns the number of rows in the layout reader.
    /// An inexact count may be larger or smaller than the actual row count.
    fn row_count(&self) -> Precision<u64>;

    /// Register the splits of this layout reader.
    // TODO(ngates): this is a temporary API until we make layout readers stream based.
    fn register_splits(
        &self,
        field_mask: &[FieldMask],
        row_offset: u64,
        splits: &mut BTreeSet<u64>,
    ) -> VortexResult<()>;

    /// Performs an approximate evaluation of the expression against the layout reader.
    fn pruning_evaluation(
        &self,
        row_range: &Range<u64>,
        expr: &ExprRef,
    ) -> VortexResult<Box<dyn PruningEvaluation>>;

    /// Performs an exact evaluation of the expression against the layout reader.
    fn filter_evaluation(
        &self,
        row_range: &Range<u64>,
        expr: &ExprRef,
    ) -> VortexResult<Box<dyn MaskEvaluation>>;

    /// Evaluates the expression against the layout.
    fn projection_evaluation(
        &self,
        row_range: &Range<u64>,
        expr: &ExprRef,
    ) -> VortexResult<Box<dyn ArrayEvaluation>>;
}

pub type MaskFuture = Shared<BoxFuture<'static, SharedVortexResult<Mask>>>;

/// Create a resolved [`MaskFuture`] from a [`Mask`].
pub fn mask_future_ready(mask: Mask) -> MaskFuture {
    async move { Ok::<_, Arc<VortexError>>(mask) }
        .boxed()
        .shared()
}

/// Returns a mask where all false values are proven to be false in the given expression.
///
/// The returned mask **does not** need to have been intersected with the input mask.
#[async_trait]
pub trait PruningEvaluation: 'static + Send + Sync {
    async fn invoke(&self, mask: Mask) -> VortexResult<Mask>;
}

pub struct NoOpPruningEvaluation;

#[async_trait]
impl PruningEvaluation for NoOpPruningEvaluation {
    async fn invoke(&self, mask: Mask) -> VortexResult<Mask> {
        Ok(mask)
    }
}

/// Refines the given mask, returning a mask equal in length to the input mask.
///
/// ## Post-conditions
///
/// The returned mask **MUST** have been intersected with the input mask.
#[async_trait]
pub trait MaskEvaluation: 'static + Send + Sync {
    async fn invoke(&self, mask: Mask) -> VortexResult<Mask>;
}

pub struct NoOpMaskEvaluation;

#[async_trait]
impl MaskEvaluation for NoOpMaskEvaluation {
    async fn invoke(&self, mask: Mask) -> VortexResult<Mask> {
        Ok(mask)
    }
}

/// Evaluates an expression against an array, returning an array equal in length to the true count
/// of the input mask.
#[async_trait]
pub trait ArrayEvaluation: 'static + Send + Sync {
    async fn invoke(&self, mask: Mask) -> VortexResult<ArrayRef>;
}

pub struct LazyReaderChildren {
    children: Arc<dyn LayoutChildren>,
    segment_source: Arc<dyn SegmentSource>,
    ctx: ArrayContext,

    // TODO(ngates): we may want a hash map of some sort here?
    cache: Vec<OnceCell<LayoutReaderRef>>,
}

impl LazyReaderChildren {
    pub fn new(
        children: Arc<dyn LayoutChildren>,
        segment_source: Arc<dyn SegmentSource>,
        ctx: ArrayContext,
    ) -> Self {
        let nchildren = children.nchildren();
        let cache = (0..nchildren).map(|_| OnceCell::new()).collect();
        Self {
            children,
            segment_source,
            ctx,
            cache,
        }
    }

    pub fn get(
        &self,
        idx: usize,
        dtype: &DType,
        name: &Arc<str>,
    ) -> VortexResult<&LayoutReaderRef> {
        if idx >= self.cache.len() {
            vortex_bail!("Child index out of bounds: {} of {}", idx, self.cache.len());
        }

        self.cache[idx].get_or_try_init(|| {
            let child = self.children.child(idx, dtype)?;
            child.new_reader(name.clone(), self.segment_source.clone(), self.ctx.clone())
        })
    }
}

1	// SPDX-License-Identifier: Apache-2.0
2	// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4	use std::collections::BTreeSet;
5	use std::ops::Range;
6	use std::sync::Arc;
7
8	use async_trait::async_trait;
9	use futures::FutureExt;
10	use futures::future::{BoxFuture, Shared};
11	use once_cell::sync::OnceCell;
12	use vortex_array::stats::Precision;
13	use vortex_array::{ArrayContext, ArrayRef};
14	use vortex_dtype::{DType, FieldMask};
15	use vortex_error::{SharedVortexResult, VortexError, VortexResult, vortex_bail};
16	use vortex_expr::ExprRef;
17	use vortex_mask::Mask;
18
19	use crate::children::LayoutChildren;
20	use crate::segments::SegmentSource;
21
22	pub type LayoutReaderRef = Arc<dyn LayoutReader>;
23
24	/// A [`LayoutReader`] is used to read a [`crate::Layout`] in a way that can cache state across multiple
25	/// evaluation operations.
26	pub trait LayoutReader: 'static + Send + Sync {
27	/// Returns the name of the layout reader for debugging.
28	fn name(&self) -> &Arc<str>;
29
30	/// Returns the un-projected dtype of the layout reader.
31	fn dtype(&self) -> &DType;
32
33	/// Returns the number of rows in the layout reader.
34	/// An inexact count may be larger or smaller than the actual row count.
35	fn row_count(&self) -> Precision<u64>;
36
37	/// Register the splits of this layout reader.
38	// TODO(ngates): this is a temporary API until we make layout readers stream based.
39	fn register_splits(
40	&self,
41	field_mask: &[FieldMask],
42	row_offset: u64,
43	splits: &mut BTreeSet<u64>,
44	) -> VortexResult<()>;
45
46	/// Performs an approximate evaluation of the expression against the layout reader.
47	fn pruning_evaluation(
48	&self,
49	row_range: &Range<u64>,
50	expr: &ExprRef,
51	) -> VortexResult<Box<dyn PruningEvaluation>>;
52
53	/// Performs an exact evaluation of the expression against the layout reader.
54	fn filter_evaluation(
55	&self,
56	row_range: &Range<u64>,
57	expr: &ExprRef,
58	) -> VortexResult<Box<dyn MaskEvaluation>>;
59
60	/// Evaluates the expression against the layout.
61	fn projection_evaluation(
62	&self,
63	row_range: &Range<u64>,
64	expr: &ExprRef,
65	) -> VortexResult<Box<dyn ArrayEvaluation>>;
66	}
67
68	pub type MaskFuture = Shared<BoxFuture<'static, SharedVortexResult<Mask>>>;
69
70	/// Create a resolved [`MaskFuture`] from a [`Mask`].
UNCOV 71	pub fn mask_future_ready(mask: Mask) -> MaskFuture {	×
UNCOV 72	async move { Ok::<_, Arc<VortexError>>(mask) }	×
UNCOV 73	.boxed()	×
74	.shared()	×
75	}	×
76
77	/// Returns a mask where all false values are proven to be false in the given expression.
78	///
79	/// The returned mask does not need to have been intersected with the input mask.
80	#[async_trait]
81	pub trait PruningEvaluation: 'static + Send + Sync {
82	async fn invoke(&self, mask: Mask) -> VortexResult<Mask>;
83	}
84
85	pub struct NoOpPruningEvaluation;
86
87	#[async_trait]
88	impl PruningEvaluation for NoOpPruningEvaluation {
89	async fn invoke(&self, mask: Mask) -> VortexResult<Mask> {	2,379✔
90	Ok(mask)	2,379✔
91	}	4,758✔
92	}
93
94	/// Refines the given mask, returning a mask equal in length to the input mask.
95	///
96	/// ## Post-conditions
97	///
98	/// The returned mask MUST have been intersected with the input mask.
99	#[async_trait]
100	pub trait MaskEvaluation: 'static + Send + Sync {
101	async fn invoke(&self, mask: Mask) -> VortexResult<Mask>;
102	}
103
104	pub struct NoOpMaskEvaluation;
105
106	#[async_trait]
107	impl MaskEvaluation for NoOpMaskEvaluation {
UNCOV 108	async fn invoke(&self, mask: Mask) -> VortexResult<Mask> {	×
UNCOV 109	Ok(mask)	×
UNCOV 110	}	×
111	}
112
113	/// Evaluates an expression against an array, returning an array equal in length to the true count
114	/// of the input mask.
115	#[async_trait]
116	pub trait ArrayEvaluation: 'static + Send + Sync {
117	async fn invoke(&self, mask: Mask) -> VortexResult<ArrayRef>;
118	}
119
120	pub struct LazyReaderChildren {
121	children: Arc<dyn LayoutChildren>,
122	segment_source: Arc<dyn SegmentSource>,
123	ctx: ArrayContext,
124
125	// TODO(ngates): we may want a hash map of some sort here?
126	cache: Vec<OnceCell<LayoutReaderRef>>,
127	}
128
129	impl LazyReaderChildren {
130	pub fn new(	1,541✔
131	children: Arc<dyn LayoutChildren>,	1,541✔
132	segment_source: Arc<dyn SegmentSource>,	1,541✔
133	ctx: ArrayContext,	1,541✔
134	) -> Self {	1,541✔
135	let nchildren = children.nchildren();	1,541✔
136	let cache = (0..nchildren).map(\|_\| OnceCell::new()).collect();	7,375✔
137	Self {	1,541✔
138	children,	1,541✔
139	segment_source,	1,541✔
140	ctx,	1,541✔
141	cache,	1,541✔
142	}	1,541✔
143	}	1,541✔
144
145	pub fn get(	31,131✔
146	&self,	31,131✔
147	idx: usize,	31,131✔
148	dtype: &DType,	31,131✔
149	name: &Arc<str>,	31,131✔
150	) -> VortexResult<&LayoutReaderRef> {	31,131✔
151	if idx >= self.cache.len() {	31,131✔
UNCOV 152	vortex_bail!("Child index out of bounds: {} of {}", idx, self.cache.len());	×
153	}	31,131✔
154		31,131✔
155	self.cache[idx].get_or_try_init(\|\| {	31,131✔
156	let child = self.children.child(idx, dtype)?;	4,967✔
157	child.new_reader(name.clone(), self.segment_source.clone(), self.ctx.clone())	4,967✔
158	})	31,131✔
159	}	31,131✔
160	}

vortex-data / vortex / 16204612549

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous