16437857475

Committed 22 Jul 2025 07:36AM UTC coverage: 81.275% (-0.3%) from 81.54%

Build # 16437857475

Build Type

Pull #3966

github

Committed by

web-flow

Commit Message

Merge a6483689c into e93723551

Pull Request Pull Request #3966: [wip] DuckDB Dynamic Expressions

Run Details

76 of 341 new or added lines in 7 files covered. (22.29%)

145 existing lines in 5 files now uncovered.

42138 of 51846 relevant lines covered (81.28%)

170221.77 hits per line

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

6.4

/vortex-expr/src/exprs/dynamic.rs

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

use std::fmt::{Debug, Display};
use std::hash::{Hash, Hasher};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use arc_swap::ArcSwapOption;
use parking_lot::Mutex;
use vortex_array::arrays::ConstantArray;
use vortex_array::compute::{Operator, compare};
use vortex_array::{Array, ArrayRef, DeserializeMetadata, IntoArray, ProstMetadata};
use vortex_dtype::DType;
use vortex_error::{VortexExpect, VortexResult, vortex_bail};
use vortex_proto::expr as pb;
use vortex_scalar::{Scalar, ScalarValue};

use crate::traversal::{Node, NodeVisitor, TraversalOrder};
use crate::{
    AnalysisExpr, ExprEncodingRef, ExprId, ExprRef, IntoExpr, Scope, StatsCatalog, VTable, vtable,
};

vtable!(DynamicComparison);

/// A dynamic comparison expression can be used to capture a comparison to a value that can change
/// during the execution of a query, such as when a compute engine pushes down an ORDER BY + LIMIT
/// operation and is able to progressively tighten the bounds of the filter.
#[allow(clippy::derived_hash_with_manual_eq)]
#[derive(Clone, Debug, Hash)]
pub struct DynamicComparisonExpr {
    lhs: ExprRef,
    operator: Operator,
    rhs: Arc<Rhs>,
    // Default value for the dynamic comparison.
    default: bool,
}

impl PartialEq for DynamicComparisonExpr {
    fn eq(&self, other: &Self) -> bool {
        self.lhs.eq(&other.lhs)
            && self.operator == other.operator
            && self.rhs == other.rhs
            && self.default == other.default
    }
}
impl Eq for DynamicComparisonExpr {}

/// Hash and PartialEq are implemented based on the ptr of the value function, such that the
/// internal value doesn't impact the hash of an expression tree.
struct Rhs {
    // The right-hand side value is a function that returns an `Option<ScalarValue>`.
    value: Arc<dyn Fn() -> Option<ScalarValue> + Send + Sync>,
    // The data type of the right-hand side value.
    dtype: DType,
    // Tracks how many times the value has changed. Note that this may over-estimate the changes
    // in order to avoid lock contention.
    version: AtomicU64,
    // The previous value of the right-hand side, used to detect changes.
    previous_value: ArcSwapOption<Scalar>,
}

impl Hash for Rhs {
    fn hash<H: Hasher>(&self, state: &mut H) {
        Arc::as_ptr(&self.value).hash(state);
    }
}

impl PartialEq for Rhs {
    fn eq(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.value, &other.value)
    }
}
impl Eq for Rhs {}

impl Debug for Rhs {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Rhs")
            .field("value", &"<dyn Fn() -> Option<ScalarValue> + Send + Sync>")
            .field("dtype", &self.dtype)
            .finish()
    }
}

pub struct DynamicComparisonExprEncoding;

impl VTable for DynamicComparisonVTable {
    type Expr = DynamicComparisonExpr;
    type Encoding = DynamicComparisonExprEncoding;
    type Metadata = ProstMetadata<pb::LiteralOpts>;

    fn id(_encoding: &Self::Encoding) -> ExprId {
        ExprId::new_ref("dynamic")
    }

    fn encoding(_expr: &Self::Expr) -> ExprEncodingRef {
        ExprEncodingRef::new_ref(DynamicComparisonExprEncoding.as_ref())
    }

    fn metadata(_expr: &Self::Expr) -> Option<Self::Metadata> {
        None
    }

    fn children(expr: &Self::Expr) -> Vec<&ExprRef> {
        vec![&expr.lhs]
    }

    fn with_children(expr: &Self::Expr, children: Vec<ExprRef>) -> VortexResult<Self::Expr> {
        Ok(DynamicComparisonExpr {
            lhs: children[0].clone(),
            operator: expr.operator,
            rhs: expr.rhs.clone(),
            default: expr.default,
        })
    }

    fn build(
        _encoding: &Self::Encoding,
        _metadata: &<Self::Metadata as DeserializeMetadata>::Output,
        _children: Vec<ExprRef>,
    ) -> VortexResult<Self::Expr> {
        vortex_bail!("DynamicComparison expression does not support building from metadata");
    }

    fn evaluate(expr: &Self::Expr, scope: &Scope) -> VortexResult<ArrayRef> {
        if let Some(value) = expr.scalar() {
            let lhs = expr.lhs.evaluate(scope)?;
            let rhs = ConstantArray::new(value, scope.len());
            return compare(lhs.as_ref(), rhs.as_ref(), expr.operator);
        }

        // Otherwise, we return the default value.
        let lhs = expr.return_dtype(scope.dtype())?;
        Ok(ConstantArray::new(
            Scalar::new(
                DType::Bool(lhs.nullability() | expr.rhs.dtype.nullability()),
                expr.default.into(),
            ),
            scope.len(),
        )
        .into_array())
    }

    fn return_dtype(expr: &Self::Expr, scope: &DType) -> VortexResult<DType> {
        let lhs = expr.lhs.return_dtype(scope)?;
        if !expr.rhs.dtype.eq_ignore_nullability(&lhs) {
            vortex_bail!(
                "Incompatible dtypes for dynamic comparison: expected {} (ignore nullability) but got {}",
                &expr.rhs.dtype,
                lhs
            );
        }
        Ok(DType::Bool(
            lhs.nullability() | expr.rhs.dtype.nullability(),
        ))
    }
}

impl DynamicComparisonExpr {
    pub fn new(
        rhs: ExprRef,
        operator: Operator,
        rhs_value: impl Fn() -> Option<ScalarValue> + Send + Sync + 'static,
        rhs_dtype: DType,
        default: bool,
    ) -> Self {
        DynamicComparisonExpr {
            lhs: rhs,
            operator,
            rhs: Arc::new(Rhs {
                value: Arc::new(rhs_value),
                dtype: rhs_dtype,
                version: Default::default(),
                previous_value: Default::default(),
            }),
            default,
        }
    }

    pub fn scalar(&self) -> Option<Scalar> {
        if let Some(next) = (self.rhs.value)().map(|v| Scalar::new(self.rhs.dtype.clone(), v)) {
            if self
                .rhs
                .previous_value
                .load()
                .as_deref()
                .is_none_or(|prev| prev != &next)
            {
                self.rhs.version.fetch_add(1, Ordering::Relaxed);
                self.rhs.previous_value.store(Some(Arc::new(next.clone())));
            }
            Some(next)
        } else {
            None
        }
    }
}

impl Display for DynamicComparisonExpr {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "dynamic({}) {} {}",
            &self.lhs, self.operator, &self.rhs.dtype,
        )
    }
}

impl AnalysisExpr for DynamicComparisonExpr {
    fn stat_falsification(&self, catalog: &mut dyn StatsCatalog) -> Option<ExprRef> {
        match self.operator {
            Operator::Gt => Some(
                DynamicComparisonExpr {
                    lhs: self.lhs.max(catalog)?,
                    operator: Operator::Lte,
                    rhs: self.rhs.clone(),
                    default: !self.default,
                }
                .into_expr(),
            ),
            Operator::Gte => Some(
                DynamicComparisonExpr {
                    lhs: self.lhs.max(catalog)?,
                    operator: Operator::Lt,
                    rhs: self.rhs.clone(),
                    default: !self.default,
                }
                .into_expr(),
            ),
            Operator::Lt => Some(
                DynamicComparisonExpr {
                    lhs: self.lhs.min(catalog)?,
                    operator: Operator::Gte,
                    rhs: self.rhs.clone(),
                    default: !self.default,
                }
                .into_expr(),
            ),
            Operator::Lte => Some(
                DynamicComparisonExpr {
                    lhs: self.lhs.min(catalog)?,
                    operator: Operator::Gt,
                    rhs: self.rhs.clone(),
                    default: !self.default,
                }
                .into_expr(),
            ),
            _ => None,
        }
    }
}

/// A utility for checking whether any dynamic expressions have been updated.
pub struct DynamicExprUpdates {
    exprs: Box<[DynamicComparisonExpr]>,
    prev_versions: Mutex<Vec<u64>>,
}

impl DynamicExprUpdates {
    pub fn new(expr: &ExprRef) -> Option<Self> {
        #[derive(Default)]
        struct Visitor(Vec<DynamicComparisonExpr>);

        impl NodeVisitor<'_> for Visitor {
            type NodeTy = ExprRef;

            fn visit_down(&mut self, node: &'_ Self::NodeTy) -> VortexResult<TraversalOrder> {
                if let Some(dynamic) = node.as_opt::<DynamicComparisonVTable>() {
                    self.0.push(dynamic.clone());
                }
                Ok(TraversalOrder::Continue)
            }
        }

        let mut visitor = Visitor::default();
        expr.accept(&mut visitor).vortex_expect("Infallible");

        if visitor.0.is_empty() {
            return None;
        }

        let exprs = visitor.0.into_boxed_slice();
        let prev_versions = exprs
            .iter()
            .map(|expr| expr.rhs.version.load(Ordering::Relaxed))
            .collect();

        Some(Self {
            exprs,
            prev_versions: Mutex::new(prev_versions),
        })
    }

    pub fn is_updated(&self) -> bool {
        let mut prev_versions = self.prev_versions.lock();
        let mut updated = false;
        for (i, expr) in self.exprs.iter().enumerate() {
            let current_version = expr.rhs.version.load(Ordering::Relaxed);
            if current_version > prev_versions[i] {
                prev_versions[i] = current_version;
                // At least one expression has been updated.
                // We don't bail out early in order to avoid false positives for future calls
                // to `is_updated`.
                updated = true;
            }
        }
        updated
    }
}

1	// SPDX-License-Identifier: Apache-2.0
2	// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4	use std::fmt::{Debug, Display};
5	use std::hash::{Hash, Hasher};
6	use std::sync::Arc;
7	use std::sync::atomic::{AtomicU64, Ordering};
8
9	use arc_swap::ArcSwapOption;
10	use parking_lot::Mutex;
11	use vortex_array::arrays::ConstantArray;
12	use vortex_array::compute::{Operator, compare};
13	use vortex_array::{Array, ArrayRef, DeserializeMetadata, IntoArray, ProstMetadata};
14	use vortex_dtype::DType;
15	use vortex_error::{VortexExpect, VortexResult, vortex_bail};
16	use vortex_proto::expr as pb;
17	use vortex_scalar::{Scalar, ScalarValue};
18
19	use crate::traversal::{Node, NodeVisitor, TraversalOrder};
20	use crate::{
21	AnalysisExpr, ExprEncodingRef, ExprId, ExprRef, IntoExpr, Scope, StatsCatalog, VTable, vtable,
22	};
23
24	vtable!(DynamicComparison);
25
26	/// A dynamic comparison expression can be used to capture a comparison to a value that can change
27	/// during the execution of a query, such as when a compute engine pushes down an ORDER BY + LIMIT
28	/// operation and is able to progressively tighten the bounds of the filter.
29	#[allow(clippy::derived_hash_with_manual_eq)]
30	#[derive(Clone, Debug, Hash)]
31	pub struct DynamicComparisonExpr {
32	lhs: ExprRef,
33	operator: Operator,
34	rhs: Arc<Rhs>,
35	// Default value for the dynamic comparison.
36	default: bool,
37	}
38
39	impl PartialEq for DynamicComparisonExpr {
NEW 40	fn eq(&self, other: &Self) -> bool {	×
NEW 41	self.lhs.eq(&other.lhs)	×
NEW 42	&& self.operator == other.operator	×
NEW 43	&& self.rhs == other.rhs	×
NEW 44	&& self.default == other.default	×
NEW 45	}	×
46	}
47	impl Eq for DynamicComparisonExpr {}
48
49	/// Hash and PartialEq are implemented based on the ptr of the value function, such that the
50	/// internal value doesn't impact the hash of an expression tree.
51	struct Rhs {
52	// The right-hand side value is a function that returns an `Option<ScalarValue>`.
53	value: Arc<dyn Fn() -> Option<ScalarValue> + Send + Sync>,
54	// The data type of the right-hand side value.
55	dtype: DType,
56	// Tracks how many times the value has changed. Note that this may over-estimate the changes
57	// in order to avoid lock contention.
58	version: AtomicU64,
59	// The previous value of the right-hand side, used to detect changes.
60	previous_value: ArcSwapOption<Scalar>,
61	}
62
63	impl Hash for Rhs {
NEW 64	fn hash<H: Hasher>(&self, state: &mut H) {	×
NEW 65	Arc::as_ptr(&self.value).hash(state);	×
NEW 66	}	×
67	}
68
69	impl PartialEq for Rhs {
NEW 70	fn eq(&self, other: &Self) -> bool {	×
NEW 71	Arc::ptr_eq(&self.value, &other.value)	×
NEW 72	}	×
73	}
74	impl Eq for Rhs {}
75
76	impl Debug for Rhs {
NEW 77	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {	×
NEW 78	f.debug_struct("Rhs")	×
NEW 79	.field("value", &"<dyn Fn() -> Option<ScalarValue> + Send + Sync>")	×
NEW 80	.field("dtype", &self.dtype)	×
NEW 81	.finish()	×
NEW 82	}	×
83	}
84
85	pub struct DynamicComparisonExprEncoding;
86
87	impl VTable for DynamicComparisonVTable {
88	type Expr = DynamicComparisonExpr;
89	type Encoding = DynamicComparisonExprEncoding;
90	type Metadata = ProstMetadata<pb::LiteralOpts>;
91
NEW 92	fn id(_encoding: &Self::Encoding) -> ExprId {	×
NEW 93	ExprId::new_ref("dynamic")	×
NEW 94	}	×
95
NEW 96	fn encoding(_expr: &Self::Expr) -> ExprEncodingRef {	×
NEW 97	ExprEncodingRef::new_ref(DynamicComparisonExprEncoding.as_ref())	×
NEW 98	}	×
99
NEW 100	fn metadata(_expr: &Self::Expr) -> Option<Self::Metadata> {	×
NEW 101	None	×
NEW 102	}	×
103
NEW 104	fn children(expr: &Self::Expr) -> Vec<&ExprRef> {	×
NEW 105	vec![&expr.lhs]	×
NEW 106	}	×
107
NEW 108	fn with_children(expr: &Self::Expr, children: Vec<ExprRef>) -> VortexResult<Self::Expr> {	×
NEW 109	Ok(DynamicComparisonExpr {	×
NEW 110	lhs: children[0].clone(),	×
NEW 111	operator: expr.operator,	×
NEW 112	rhs: expr.rhs.clone(),	×
NEW 113	default: expr.default,	×
NEW 114	})	×
NEW 115	}	×
116
NEW 117	fn build(	×
NEW 118	_encoding: &Self::Encoding,	×
NEW 119	_metadata: &<Self::Metadata as DeserializeMetadata>::Output,	×
NEW 120	_children: Vec<ExprRef>,	×
NEW 121	) -> VortexResult<Self::Expr> {	×
NEW 122	vortex_bail!("DynamicComparison expression does not support building from metadata");	×
NEW 123	}	×
124
NEW 125	fn evaluate(expr: &Self::Expr, scope: &Scope) -> VortexResult<ArrayRef> {	×
NEW 126	if let Some(value) = expr.scalar() {	×
NEW 127	let lhs = expr.lhs.evaluate(scope)?;	×
NEW 128	let rhs = ConstantArray::new(value, scope.len());	×
NEW 129	return compare(lhs.as_ref(), rhs.as_ref(), expr.operator);	×
NEW 130	}	×
131
132	// Otherwise, we return the default value.
NEW 133	let lhs = expr.return_dtype(scope.dtype())?;	×
NEW 134	Ok(ConstantArray::new(	×
NEW 135	Scalar::new(	×
NEW 136	DType::Bool(lhs.nullability() \| expr.rhs.dtype.nullability()),	×
NEW 137	expr.default.into(),	×
NEW 138	),	×
NEW 139	scope.len(),	×
NEW 140	)	×
NEW 141	.into_array())	×
NEW 142	}	×
143
NEW 144	fn return_dtype(expr: &Self::Expr, scope: &DType) -> VortexResult<DType> {	×
NEW 145	let lhs = expr.lhs.return_dtype(scope)?;	×
NEW 146	if !expr.rhs.dtype.eq_ignore_nullability(&lhs) {	×
NEW 147	vortex_bail!(	×
NEW 148	"Incompatible dtypes for dynamic comparison: expected {} (ignore nullability) but got {}",	×
NEW 149	&expr.rhs.dtype,	×
150	lhs
151	);
NEW 152	}	×
NEW 153	Ok(DType::Bool(	×
NEW 154	lhs.nullability() \| expr.rhs.dtype.nullability(),	×
NEW 155	))	×
NEW 156	}	×
157	}
158
159	impl DynamicComparisonExpr {
NEW 160	pub fn new(	×
NEW 161	rhs: ExprRef,	×
NEW 162	operator: Operator,	×
NEW 163	rhs_value: impl Fn() -> Option<ScalarValue> + Send + Sync + 'static,	×
NEW 164	rhs_dtype: DType,	×
NEW 165	default: bool,	×
NEW 166	) -> Self {	×
NEW 167	DynamicComparisonExpr {	×
NEW 168	lhs: rhs,	×
NEW 169	operator,	×
NEW 170	rhs: Arc::new(Rhs {	×
NEW 171	value: Arc::new(rhs_value),	×
NEW 172	dtype: rhs_dtype,	×
NEW 173	version: Default::default(),	×
NEW 174	previous_value: Default::default(),	×
NEW 175	}),	×
NEW 176	default,	×
NEW 177	}	×
NEW 178	}	×
179
NEW 180	pub fn scalar(&self) -> Option<Scalar> {	×
NEW 181	if let Some(next) = (self.rhs.value)().map(\|v\| Scalar::new(self.rhs.dtype.clone(), v)) {	×
NEW 182	if self	×
NEW 183	.rhs	×
NEW 184	.previous_value	×
NEW 185	.load()	×
NEW 186	.as_deref()	×
NEW 187	.is_none_or(\|prev\| prev != &next)	×
NEW 188	{	×
NEW 189	self.rhs.version.fetch_add(1, Ordering::Relaxed);	×
NEW 190	self.rhs.previous_value.store(Some(Arc::new(next.clone())));	×
NEW 191	}	×
NEW 192	Some(next)	×
193	} else {
NEW 194	None	×
195	}
NEW 196	}	×
197	}
198
199	impl Display for DynamicComparisonExpr {
NEW 200	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {	×
NEW 201	write!(	×
NEW 202	f,	×
NEW 203	"dynamic({}) {} {}",	×
NEW 204	&self.lhs, self.operator, &self.rhs.dtype,	×
205	)
NEW 206	}	×
207	}
208
209	impl AnalysisExpr for DynamicComparisonExpr {
NEW 210	fn stat_falsification(&self, catalog: &mut dyn StatsCatalog) -> Option<ExprRef> {	×
NEW 211	match self.operator {	×
212	Operator::Gt => Some(
213	DynamicComparisonExpr {
NEW 214	lhs: self.lhs.max(catalog)?,	×
NEW 215	operator: Operator::Lte,	×
NEW 216	rhs: self.rhs.clone(),	×
NEW 217	default: !self.default,	×
218	}
NEW 219	.into_expr(),	×
220	),
221	Operator::Gte => Some(
222	DynamicComparisonExpr {
NEW 223	lhs: self.lhs.max(catalog)?,	×
NEW 224	operator: Operator::Lt,	×
NEW 225	rhs: self.rhs.clone(),	×
NEW 226	default: !self.default,	×
227	}
NEW 228	.into_expr(),	×
229	),
230	Operator::Lt => Some(
231	DynamicComparisonExpr {
NEW 232	lhs: self.lhs.min(catalog)?,	×
NEW 233	operator: Operator::Gte,	×
NEW 234	rhs: self.rhs.clone(),	×
NEW 235	default: !self.default,	×
236	}
NEW 237	.into_expr(),	×
238	),
239	Operator::Lte => Some(
240	DynamicComparisonExpr {
NEW 241	lhs: self.lhs.min(catalog)?,	×
NEW 242	operator: Operator::Gt,	×
NEW 243	rhs: self.rhs.clone(),	×
NEW 244	default: !self.default,	×
245	}
NEW 246	.into_expr(),	×
247	),
NEW 248	_ => None,	×
249	}
NEW 250	}	×
251	}
252
253	/// A utility for checking whether any dynamic expressions have been updated.
254	pub struct DynamicExprUpdates {
255	exprs: Box<[DynamicComparisonExpr]>,
256	prev_versions: Mutex<Vec<u64>>,
257	}
258
259	impl DynamicExprUpdates {
260	pub fn new(expr: &ExprRef) -> Option<Self> {	2,264✔
261	#[derive(Default)]
262	struct Visitor(Vec<DynamicComparisonExpr>);
263
264	impl NodeVisitor<'_> for Visitor {
265	type NodeTy = ExprRef;
266
267	fn visit_down(&mut self, node: &'_ Self::NodeTy) -> VortexResult<TraversalOrder> {	7,056✔
268	if let Some(dynamic) = node.as_opt::<DynamicComparisonVTable>() {	7,056✔
NEW 269	self.0.push(dynamic.clone());	×
270	}	7,056✔
271	Ok(TraversalOrder::Continue)	7,056✔
272	}	7,056✔
273	}
274
275	let mut visitor = Visitor::default();	2,264✔
276	expr.accept(&mut visitor).vortex_expect("Infallible");	2,264✔
277
278	if visitor.0.is_empty() {	2,264✔
279	return None;	2,264✔
NEW 280	}	×
281
NEW 282	let exprs = visitor.0.into_boxed_slice();	×
NEW 283	let prev_versions = exprs	×
NEW 284	.iter()	×
NEW 285	.map(\|expr\| expr.rhs.version.load(Ordering::Relaxed))	×
NEW 286	.collect();	×
287
NEW 288	Some(Self {	×
NEW 289	exprs,	×
NEW 290	prev_versions: Mutex::new(prev_versions),	×
NEW 291	})	×
292	}	2,264✔
293
NEW 294	pub fn is_updated(&self) -> bool {	×
NEW 295	let mut prev_versions = self.prev_versions.lock();	×
NEW 296	let mut updated = false;	×
NEW 297	for (i, expr) in self.exprs.iter().enumerate() {	×
NEW 298	let current_version = expr.rhs.version.load(Ordering::Relaxed);	×
NEW 299	if current_version > prev_versions[i] {	×
NEW 300	prev_versions[i] = current_version;	×
NEW 301	// At least one expression has been updated.	×
NEW 302	// We don't bail out early in order to avoid false positives for future calls	×
NEW 303	// to `is_updated`.	×
NEW 304	updated = true;	×
NEW 305	}	×
306	}
NEW 307	updated	×
NEW 308	}	×
309	}

vortex-data / vortex / 16437857475

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