16992684502

Committed 15 Aug 2025 02:56PM UTC coverage: 87.875% (+0.2%) from 87.72%

Build # 16992684502

Build Type

Pull #2456

github

Committed by

web-flow

Commit Message

Merge 2d540e578 into 4a23f65b3

Pull Request Pull Request #2456: feat: basic BoolBuffer / BoolBufferMut

Run Details

1275 of 1428 new or added lines in 110 files covered. (89.29%)

334 existing lines in 31 files now uncovered.

57169 of 65057 relevant lines covered (87.88%)

658056.52 hits per line

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

93.75

/vortex-datafusion/src/convert/exprs.rs

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

use std::sync::Arc;

use arrow_schema::{DataType, Schema};
use datafusion_expr::Operator as DFOperator;
use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef};
use datafusion_physical_plan::expressions as df_expr;
use vortex::error::{VortexResult, vortex_bail, vortex_err};
use vortex::expr::{BinaryExpr, ExprRef, LikeExpr, Operator, and, get_item, lit, root};
use vortex::scalar::Scalar;

use crate::convert::{FromDataFusion, TryFromDataFusion};

const SUPPORTED_BINARY_OPS: &[DFOperator] = &[
    DFOperator::Eq,
    DFOperator::NotEq,
    DFOperator::Gt,
    DFOperator::GtEq,
    DFOperator::Lt,
    DFOperator::LtEq,
];

/// Tries to convert the expressions into a vortex conjunction. Will return Ok(None) iff the input conjunction is empty.
pub(crate) fn make_vortex_predicate(
    predicate: &[&Arc<dyn PhysicalExpr>],
) -> VortexResult<Option<ExprRef>> {
    let exprs = predicate
        .iter()
        .map(|e| ExprRef::try_from_df(e.as_ref()))
        .collect::<VortexResult<Vec<_>>>()?;

    Ok(exprs.into_iter().reduce(and))
}

// TODO(joe): Don't return an error when we have an unsupported node, bubble up "TRUE" as in keep
//  for that node, up to any `and` or `or` node.
impl TryFromDataFusion<dyn PhysicalExpr> for ExprRef {
    fn try_from_df(df: &dyn PhysicalExpr) -> VortexResult<Self> {
        if let Some(binary_expr) = df.as_any().downcast_ref::<df_expr::BinaryExpr>() {
            let left = ExprRef::try_from_df(binary_expr.left().as_ref())?;
            let right = ExprRef::try_from_df(binary_expr.right().as_ref())?;
            let operator = Operator::try_from_df(binary_expr.op())?;

            return Ok(BinaryExpr::new_expr(left, operator, right));
        }

        if let Some(col_expr) = df.as_any().downcast_ref::<df_expr::Column>() {
            return Ok(get_item(col_expr.name().to_owned(), root()));
        }

        if let Some(like) = df.as_any().downcast_ref::<df_expr::LikeExpr>() {
            let child = ExprRef::try_from_df(like.expr().as_ref())?;
            let pattern = ExprRef::try_from_df(like.pattern().as_ref())?;
            return Ok(LikeExpr::new_expr(
                child,
                pattern,
                like.negated(),
                like.case_insensitive(),
            ));
        }

        if let Some(literal) = df.as_any().downcast_ref::<df_expr::Literal>() {
            let value = Scalar::from_df(literal.value());
            return Ok(lit(value));
        }

        vortex_bail!("Couldn't convert DataFusion physical {df} expression to a vortex expression")
    }
}

impl TryFromDataFusion<DFOperator> for Operator {
    fn try_from_df(value: &DFOperator) -> VortexResult<Self> {
        match value {
            DFOperator::Eq => Ok(Operator::Eq),
            DFOperator::NotEq => Ok(Operator::NotEq),
            DFOperator::Lt => Ok(Operator::Lt),
            DFOperator::LtEq => Ok(Operator::Lte),
            DFOperator::Gt => Ok(Operator::Gt),
            DFOperator::GtEq => Ok(Operator::Gte),
            DFOperator::And => Ok(Operator::And),
            DFOperator::Or => Ok(Operator::Or),
            DFOperator::IsDistinctFrom
            | DFOperator::IsNotDistinctFrom
            | DFOperator::RegexMatch
            | DFOperator::RegexIMatch
            | DFOperator::RegexNotMatch
            | DFOperator::RegexNotIMatch
            | DFOperator::LikeMatch
            | DFOperator::ILikeMatch
            | DFOperator::NotLikeMatch
            | DFOperator::NotILikeMatch
            | DFOperator::BitwiseAnd
            | DFOperator::BitwiseOr
            | DFOperator::BitwiseXor
            | DFOperator::BitwiseShiftRight
            | DFOperator::BitwiseShiftLeft
            | DFOperator::StringConcat
            | DFOperator::AtArrow
            | DFOperator::ArrowAt
            | DFOperator::Plus
            | DFOperator::Minus
            | DFOperator::Multiply
            | DFOperator::Divide
            | DFOperator::Modulo
            | DFOperator::Arrow
            | DFOperator::LongArrow
            | DFOperator::HashArrow
            | DFOperator::HashLongArrow
            | DFOperator::AtAt
            | DFOperator::IntegerDivide
            | DFOperator::HashMinus
            | DFOperator::AtQuestion
            | DFOperator::Question
            | DFOperator::QuestionAnd
            | DFOperator::QuestionPipe => {
                Err(vortex_err!("Unsupported datafusion operator {value}"))
            }
        }
    }
}

pub(crate) fn can_be_pushed_down(expr: &PhysicalExprRef, schema: &Schema) -> bool {
    let expr = expr.as_any();
    if let Some(binary) = expr.downcast_ref::<df_expr::BinaryExpr>() {
        can_binary_be_pushed_down(binary, schema)
    } else if let Some(col) = expr.downcast_ref::<df_expr::Column>() {
        schema
            .field_with_name(col.name())
            .ok()
            .is_some_and(|field| supported_data_types(field.data_type()))
    } else if let Some(like) = expr.downcast_ref::<df_expr::LikeExpr>() {
        can_be_pushed_down(like.expr(), schema) && can_be_pushed_down(like.pattern(), schema)
    } else if let Some(lit) = expr.downcast_ref::<df_expr::Literal>() {
        supported_data_types(&lit.value().data_type())
    } else {
        log::debug!("DataFusion expression can't be pushed down: {expr:?}");
        false
    }
}

fn can_binary_be_pushed_down(binary: &df_expr::BinaryExpr, schema: &Schema) -> bool {
    let is_op_supported =
        binary.op().is_logic_operator() || SUPPORTED_BINARY_OPS.contains(binary.op());
    is_op_supported
        && can_be_pushed_down(binary.left(), schema)
        && can_be_pushed_down(binary.right(), schema)
}

fn supported_data_types(dt: &DataType) -> bool {
    use DataType::*;
    let is_supported = dt.is_null()
        || dt.is_numeric()
        || matches!(
            dt,
            Boolean
                | Utf8
                | Utf8View
                | Binary
                | BinaryView
                | Date32
                | Date64
                | Timestamp(_, _)
                | Time32(_)
                | Time64(_)
        );

    if !is_supported {
        log::debug!("DataFusion data type {dt:?} is not supported");
    }

    is_supported
}

1	// SPDX-License-Identifier: Apache-2.0
2	// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4	use std::sync::Arc;
5
6	use arrow_schema::{DataType, Schema};
7	use datafusion_expr::Operator as DFOperator;
8	use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef};
9	use datafusion_physical_plan::expressions as df_expr;
10	use vortex::error::{VortexResult, vortex_bail, vortex_err};
11	use vortex::expr::{BinaryExpr, ExprRef, LikeExpr, Operator, and, get_item, lit, root};
12	use vortex::scalar::Scalar;
13
14	use crate::convert::{FromDataFusion, TryFromDataFusion};
15
16	const SUPPORTED_BINARY_OPS: &[DFOperator] = &[
17	DFOperator::Eq,
18	DFOperator::NotEq,
19	DFOperator::Gt,
20	DFOperator::GtEq,
21	DFOperator::Lt,
22	DFOperator::LtEq,
23	];
24
25	/// Tries to convert the expressions into a vortex conjunction. Will return Ok(None) iff the input conjunction is empty.
26	pub(crate) fn make_vortex_predicate(	284✔
27	predicate: &[&Arc<dyn PhysicalExpr>],	284✔
28	) -> VortexResult<Option<ExprRef>> {	284✔
29	let exprs = predicate	284✔
30	.iter()	284✔
31	.map(\|e\| ExprRef::try_from_df(e.as_ref()))	538✔
32	.collect::<VortexResult<Vec<_>>>()?;	284✔
33
34	Ok(exprs.into_iter().reduce(and))	284✔
35	}	284✔
36
37	// TODO(joe): Don't return an error when we have an unsupported node, bubble up "TRUE" as in keep
38	// for that node, up to any `and` or `or` node.
39	impl TryFromDataFusion<dyn PhysicalExpr> for ExprRef {
40	fn try_from_df(df: &dyn PhysicalExpr) -> VortexResult<Self> {	2,072✔
41	if let Some(binary_expr) = df.as_any().downcast_ref::<df_expr::BinaryExpr>() {	2,072✔
42	let left = ExprRef::try_from_df(binary_expr.left().as_ref())?;	756✔
43	let right = ExprRef::try_from_df(binary_expr.right().as_ref())?;	756✔
44	let operator = Operator::try_from_df(binary_expr.op())?;	756✔
45
46	return Ok(BinaryExpr::new_expr(left, operator, right));	756✔
47	}	1,316✔
48
49	if let Some(col_expr) = df.as_any().downcast_ref::<df_expr::Column>() {	1,316✔
50	return Ok(get_item(col_expr.name().to_owned(), root()));	730✔
51	}	586✔
52
53	if let Some(like) = df.as_any().downcast_ref::<df_expr::LikeExpr>() {	586✔
54	let child = ExprRef::try_from_df(like.expr().as_ref())?;	12✔
55	let pattern = ExprRef::try_from_df(like.pattern().as_ref())?;	12✔
56	return Ok(LikeExpr::new_expr(	12✔
57	child,	12✔
58	pattern,	12✔
59	like.negated(),	12✔
60	like.case_insensitive(),	12✔
61	));	12✔
62	}	574✔
63
64	if let Some(literal) = df.as_any().downcast_ref::<df_expr::Literal>() {	574✔
65	let value = Scalar::from_df(literal.value());	574✔
66	return Ok(lit(value));	574✔
UNCOV 67	}	×
68
UNCOV 69	vortex_bail!("Couldn't convert DataFusion physical {df} expression to a vortex expression")	×
70	}	2,072✔
71	}
72
73	impl TryFromDataFusion<DFOperator> for Operator {
74	fn try_from_df(value: &DFOperator) -> VortexResult<Self> {	756✔
75	match value {	756✔
76	DFOperator::Eq => Ok(Operator::Eq),	134✔
77	DFOperator::NotEq => Ok(Operator::NotEq),	2✔
78	DFOperator::Lt => Ok(Operator::Lt),	138✔
79	DFOperator::LtEq => Ok(Operator::Lte),	98✔
80	DFOperator::Gt => Ok(Operator::Gt),	82✔
81	DFOperator::GtEq => Ok(Operator::Gte),	186✔
82	DFOperator::And => Ok(Operator::And),	48✔
83	DFOperator::Or => Ok(Operator::Or),	68✔
84	DFOperator::IsDistinctFrom
85	\| DFOperator::IsNotDistinctFrom
86	\| DFOperator::RegexMatch
87	\| DFOperator::RegexIMatch
88	\| DFOperator::RegexNotMatch
89	\| DFOperator::RegexNotIMatch
90	\| DFOperator::LikeMatch
91	\| DFOperator::ILikeMatch
92	\| DFOperator::NotLikeMatch
93	\| DFOperator::NotILikeMatch
94	\| DFOperator::BitwiseAnd
95	\| DFOperator::BitwiseOr
96	\| DFOperator::BitwiseXor
97	\| DFOperator::BitwiseShiftRight
98	\| DFOperator::BitwiseShiftLeft
99	\| DFOperator::StringConcat
100	\| DFOperator::AtArrow
101	\| DFOperator::ArrowAt
102	\| DFOperator::Plus
103	\| DFOperator::Minus
104	\| DFOperator::Multiply
105	\| DFOperator::Divide
106	\| DFOperator::Modulo
107	\| DFOperator::Arrow
108	\| DFOperator::LongArrow
109	\| DFOperator::HashArrow
110	\| DFOperator::HashLongArrow
111	\| DFOperator::AtAt
112	\| DFOperator::IntegerDivide
113	\| DFOperator::HashMinus
114	\| DFOperator::AtQuestion
115	\| DFOperator::Question
116	\| DFOperator::QuestionAnd
117	\| DFOperator::QuestionPipe => {
UNCOV 118	Err(vortex_err!("Unsupported datafusion operator {value}"))	×
119	}
120	}
121	}	756✔
122	}
123
124	pub(crate) fn can_be_pushed_down(expr: &PhysicalExprRef, schema: &Schema) -> bool {	2,548✔
125	let expr = expr.as_any();	2,548✔
126	if let Some(binary) = expr.downcast_ref::<df_expr::BinaryExpr>() {	2,548✔
127	can_binary_be_pushed_down(binary, schema)	918✔
128	} else if let Some(col) = expr.downcast_ref::<df_expr::Column>() {	1,630✔
129	schema	888✔
130	.field_with_name(col.name())	888✔
131	.ok()	888✔
132	.is_some_and(\|field\| supported_data_types(field.data_type()))	888✔
133	} else if let Some(like) = expr.downcast_ref::<df_expr::LikeExpr>() {	742✔
134	can_be_pushed_down(like.expr(), schema) && can_be_pushed_down(like.pattern(), schema)	24✔
135	} else if let Some(lit) = expr.downcast_ref::<df_expr::Literal>() {	718✔
136	supported_data_types(&lit.value().data_type())	710✔
137	} else {
138	log::debug!("DataFusion expression can't be pushed down: {expr:?}");	8✔
139	false	8✔
140	}
141	}	2,548✔
142
143	fn can_binary_be_pushed_down(binary: &df_expr::BinaryExpr, schema: &Schema) -> bool {	918✔
144	let is_op_supported =	918✔
145	binary.op().is_logic_operator() \|\| SUPPORTED_BINARY_OPS.contains(binary.op());	918✔
146	is_op_supported	918✔
147	&& can_be_pushed_down(binary.left(), schema)	918✔
148	&& can_be_pushed_down(binary.right(), schema)	910✔
149	}	918✔
150
151	fn supported_data_types(dt: &DataType) -> bool {	1,596✔
152	use DataType::*;
153	let is_supported = dt.is_null()	1,596✔
154	\|\| dt.is_numeric()	1,596✔
UNCOV 155	\|\| matches!(	×
156	dt,	1,240✔
157	Boolean
158	\| Utf8
159	\| Utf8View
160	\| Binary
161	\| BinaryView
162	\| Date32
163	\| Date64
164	\| Timestamp(_, _)
165	\| Time32(_)
166	\| Time64(_)
167	);
168
169	if !is_supported {	1,596✔
UNCOV 170	log::debug!("DataFusion data type {dt:?} is not supported");	×
171	}	1,596✔
172
173	is_supported	1,596✔
174	}	1,596✔

vortex-data / vortex / 16992684502

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