16973175259

Committed 14 Aug 2025 06:13PM UTC coverage: 87.744% (+0.02%) from 87.72%

Build # 16973175259

Build Type

Pull #4180

github

Committed by

web-flow

Commit Message

Merge 0a6ed3f05 into f0147c092

Pull Request Pull Request #4180: feat: Support more datafusion features

Run Details

227 of 271 new or added lines in 5 files covered. (83.76%)

2 existing lines in 1 file now uncovered.

56401 of 64279 relevant lines covered (87.74%)

628978.31 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 datafusion::arrow::datatypes::{DataType, Schema};
use datafusion::logical_expr::Operator as DFOperator;
use datafusion::physical_expr::{PhysicalExpr, PhysicalExprRef, expressions};
use vortex::error::{VortexResult, vortex_bail, vortex_err};
use vortex::expr::{ExprRef, Operator, and};
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> {
        use vortex::expr::{BinaryExpr, ExprRef, LikeExpr, get_item, lit, root};

        if let Some(binary_expr) = df.as_any().downcast_ref::<expressions::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::<expressions::Column>() {
            return Ok(get_item(col_expr.name().to_owned(), root()));
        }

        if let Some(like) = df.as_any().downcast_ref::<expressions::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::<expressions::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 {
    use datafusion::physical_plan::expressions::{BinaryExpr, Column, LikeExpr, Literal};

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

vortex-data / vortex / 16973175259

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