16446615099

Committed 22 Jul 2025 02:03PM UTC coverage: 81.189% (-0.3%) from 81.5%

Build # 16446615099

Build Type

push

github

Committed by

web-flow

Commit Message

DuckDB Dynamic Expressions (#3966)

Adds support for dynamic expressions to DuckDB extension. We may need to
rethink this if/when other query engines support dynamic expressions,
but it works well enough for now.

Interesting potential follow up:
* If we run ahead in the background and open all files, then we get file
stats.
* IFF we have a dynamic filter, we can assume the bound will tighten
during execution. Therefore we should re-order our files/splits to make
the most of the dynamic filter, i.e. for ORDER BY event_time DESC, we'd
scan files in reverse order in an attempt to populate the highest
event_time first.

---------

Signed-off-by: Nicholas Gates <nick@nickgates.com>

Run Details

74 of 285 new or added lines in 4 files covered. (25.96%)

2 existing lines in 1 file now uncovered.

42008 of 51741 relevant lines covered (81.19%)

170908.17 hits per line

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

6.75

/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 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.
#[derive(Clone, Debug)]
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.default == other.default
            && self.operator == other.operator
            && self.lhs.eq(&other.lhs)
            && Arc::ptr_eq(&self.rhs.value, &other.rhs.value)
            && self.rhs.dtype == other.rhs.dtype
    }
}
impl Eq for DynamicComparisonExpr {}

impl Hash for DynamicComparisonExpr {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.default.hash(state);
        self.operator.hash(state);
        self.lhs.hash(state);
        Arc::as_ptr(&self.rhs.value).hash(state);
        self.rhs.dtype.hash(state);
    }
}

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

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,
            }),
            default,
        }
    }

    pub fn scalar(&self) -> Option<Scalar> {
        (self.rhs.value)().map(|v| Scalar::new(self.rhs.dtype.clone(), v))
    }
}

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]>,
    // Track the latest observed versions of each dynamic expression, along with a version counter.
    prev_versions: Mutex<(u64, Vec<Option<Scalar>>)>,
}

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.value)().map(|v| Scalar::new(expr.rhs.dtype.clone(), v)))
            .collect();

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

    pub fn version(&self) -> u64 {
        let mut guard = self.prev_versions.lock();

        let mut updated = false;
        for (i, expr) in self.exprs.iter().enumerate() {
            let current = expr.scalar();
            if current != guard.1[i] {
                // 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;
                guard.1[i] = current;
            }
        }

        if updated {
            guard.0 += 1;
        }

        guard.0
    }
}

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

vortex-data / vortex / 16446615099

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