16277982668

Committed 14 Jul 2025 09:12PM UTC coverage: 81.564% (+0.4%) from 81.147%

Build # 16277982668

Build Type

Pull #3852

github

Committed by

web-flow

Commit Message

Merge e78f6e62e into b0be264bf

Pull Request Pull Request #3852: feat: call optimize in compressor

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

381 existing lines in 36 files now uncovered.

46289 of 56752 relevant lines covered (81.56%)

157514.17 hits per line

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

74.29

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

    // 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::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::ArrayRef;
13	use vortex_array::stats::Precision;
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`].
71	pub fn mask_future_ready(mask: Mask) -> MaskFuture {	×
72	async move { Ok::<_, Arc<VortexError>>(mask) }	×
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,369✔
90	Ok(mask)	2,369✔
91	}	4,738✔
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 {
108	async fn invoke(&self, mask: Mask) -> VortexResult<Mask> {	×
109	Ok(mask)	×
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
124	// TODO(ngates): we may want a hash map of some sort here?
125	cache: Vec<OnceCell<LayoutReaderRef>>,
126	}
127
128	impl LazyReaderChildren {
129	pub fn new(children: Arc<dyn LayoutChildren>, segment_source: Arc<dyn SegmentSource>) -> Self {	1,539✔
130	let nchildren = children.nchildren();	1,539✔
131	let cache = (0..nchildren).map(\|_\| OnceCell::new()).collect();	7,423✔
132	Self {	1,539✔
133	children,	1,539✔
134	segment_source,	1,539✔
135	cache,	1,539✔
136	}	1,539✔
137	}	1,539✔
138
139	pub fn get(	31,047✔
140	&self,	31,047✔
141	idx: usize,	31,047✔
142	dtype: &DType,	31,047✔
143	name: &Arc<str>,	31,047✔
144	) -> VortexResult<&LayoutReaderRef> {	31,047✔
145	if idx >= self.cache.len() {	31,047✔
UNCOV 146	vortex_bail!("Child index out of bounds: {} of {}", idx, self.cache.len());	×
147	}	31,047✔
148		31,047✔
149	self.cache[idx].get_or_try_init(\|\| {	31,047✔
150	let child = self.children.child(idx, dtype)?;	4,987✔
151	child.new_reader(name.clone(), self.segment_source.clone())	4,987✔
152	})	31,047✔
153	}	31,047✔
154	}

vortex-data / vortex / 16277982668

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