16448798227

Committed 22 Jul 2025 03:30PM UTC coverage: 81.013% (-0.1%) from 81.109%

Build # 16448798227

Build Type

Pull #3876

github

Committed by

web-flow

Commit Message

Merge 5ae90ffd4 into db33b9fe9

Pull Request Pull Request #3876: feat[layout]: replace register_splits with a layout splits stream

Run Details

460 of 571 new or added lines in 17 files covered. (80.56%)

29 existing lines in 4 files now uncovered.

42262 of 52167 relevant lines covered (81.01%)

169362.77 hits per line

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

73.53

/vortex-layout/src/reader.rs

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

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::ArrayRef;
use vortex_array::stats::Precision;
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::masks::BoxMaskIterator;
use crate::row_selection::RowSelectionRef;
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.
    /// FIXME(ngates): remove this.
    fn row_count(&self) -> Precision<u64>;

    /// Given a [`SlicedSelection`] which can answer range included queryies, returns an iterator of
    /// [`Mask`]s from the layout reader that cover the full range of rows.
    /// These masks are likely to be partitioned in a way that is reasonable efficient for
    /// partitioning evaluation of the [`LayoutReader`] - but there's no guarantee.
    fn row_masks(&self, selection: &RowSelectionRef, field_mask: &[FieldMask]) -> BoxMaskIterator;

    /// 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>,

    // 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>) -> Self {
        let nchildren = children.nchildren();
        let cache = (0..nchildren).map(|_| OnceCell::new()).collect();
        Self {
            children,
            segment_source,
            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())
        })
    }
}

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

vortex-data / vortex / 16448798227

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