1use crate::LogicalPlan;
10use sqlparser::dialect::GenericDialect;
11use sqlparser::parser::Parser;
12
13use super::OptPass;
14
15pub struct PredicatePushdown;
19
20impl OptPass for PredicatePushdown {
21 fn name(&self) -> &'static str {
22 "PredicatePushdown"
23 }
24
25 fn apply(&self, plan: LogicalPlan) -> LogicalPlan {
26 push_predicate(plan)
27 }
28}
29
30pub(super) fn push_predicate(plan: LogicalPlan) -> LogicalPlan {
34 match plan {
35 LogicalPlan::Filter { input, predicate } => {
36 let inner = push_predicate(*input);
38
39 match inner {
40 LogicalPlan::Join {
46 left,
47 right,
48 on,
49 join_type,
50 ..
51 } => {
52 let can_push =
53 matches!(join_type, crate::JoinType::Inner | crate::JoinType::Cross);
54
55 let left_tables = collect_table_names(&left);
56 let right_tables = collect_table_names(&right);
57
58 let only_left =
59 can_push && predicate_refs_only(&predicate, &left_tables, &right_tables);
60 let only_right =
61 can_push && predicate_refs_only(&predicate, &right_tables, &left_tables);
62
63 if only_left {
64 let new_left = push_predicate(LogicalPlan::Filter {
65 input: left,
66 predicate,
67 });
68 LogicalPlan::Join {
69 left: Box::new(new_left),
70 right,
71 on,
72 join_type,
73 algo_hint: None,
74 }
75 } else if only_right {
76 let new_right = push_predicate(LogicalPlan::Filter {
77 input: right,
78 predicate,
79 });
80 LogicalPlan::Join {
81 left,
82 right: Box::new(new_right),
83 on,
84 join_type,
85 algo_hint: None,
86 }
87 } else {
88 LogicalPlan::Filter {
90 input: Box::new(LogicalPlan::Join {
91 left,
92 right,
93 on,
94 join_type,
95 algo_hint: None,
96 }),
97 predicate,
98 }
99 }
100 }
101
102 LogicalPlan::Project { input, columns } => {
105 let pred_cols = extract_identifier_names(&predicate);
106 let proj_cols: Vec<&str> = columns.iter().map(|c| c.as_str()).collect();
107 let all_in_scope = pred_cols
108 .iter()
109 .all(|pc| proj_cols.iter().any(|prc| prc.contains(pc.as_str())));
110
111 if all_in_scope {
112 LogicalPlan::Project {
113 input: Box::new(push_predicate(LogicalPlan::Filter {
114 input,
115 predicate,
116 })),
117 columns,
118 }
119 } else {
120 LogicalPlan::Filter {
121 input: Box::new(LogicalPlan::Project { input, columns }),
122 predicate,
123 }
124 }
125 }
126
127 other => LogicalPlan::Filter {
129 input: Box::new(other),
130 predicate,
131 },
132 }
133 }
134
135 LogicalPlan::Project { input, columns } => LogicalPlan::Project {
137 input: Box::new(push_predicate(*input)),
138 columns,
139 },
140 LogicalPlan::Join {
141 left,
142 right,
143 on,
144 join_type,
145 algo_hint,
146 } => LogicalPlan::Join {
147 left: Box::new(push_predicate(*left)),
148 right: Box::new(push_predicate(*right)),
149 on,
150 join_type,
151 algo_hint,
152 },
153 LogicalPlan::Aggregate {
154 input,
155 group_by,
156 aggregates,
157 } => LogicalPlan::Aggregate {
158 input: Box::new(push_predicate(*input)),
159 group_by,
160 aggregates,
161 },
162 LogicalPlan::Sort { input, order_by } => LogicalPlan::Sort {
163 input: Box::new(push_predicate(*input)),
164 order_by,
165 },
166 LogicalPlan::Limit {
167 input,
168 count,
169 offset,
170 } => LogicalPlan::Limit {
171 input: Box::new(push_predicate(*input)),
172 count,
173 offset,
174 },
175 other => other,
177 }
178}
179
180fn collect_table_names(plan: &LogicalPlan) -> Vec<String> {
184 let mut names = Vec::new();
185 collect_table_names_inner(plan, &mut names);
186 names
187}
188
189fn collect_table_names_inner(plan: &LogicalPlan, out: &mut Vec<String>) {
190 match plan {
191 LogicalPlan::Scan { table, alias, .. } => {
192 out.push(table.clone());
193 if let Some(a) = alias {
194 out.push(a.clone());
195 }
196 }
197 LogicalPlan::Filter { input, .. }
198 | LogicalPlan::Project { input, .. }
199 | LogicalPlan::Aggregate { input, .. }
200 | LogicalPlan::Sort { input, .. }
201 | LogicalPlan::Limit { input, .. } => collect_table_names_inner(input, out),
202 LogicalPlan::Join { left, right, .. } => {
203 collect_table_names_inner(left, out);
204 collect_table_names_inner(right, out);
205 }
206 LogicalPlan::SetOp { left, right, .. } => {
207 collect_table_names_inner(left, out);
208 collect_table_names_inner(right, out);
209 }
210 LogicalPlan::Cte { query, .. } => collect_table_names_inner(query, out),
211 LogicalPlan::Window { input, .. } => collect_table_names_inner(input, out),
212 LogicalPlan::Subquery { query, .. }
213 | LogicalPlan::Exists {
214 subquery: query, ..
215 }
216 | LogicalPlan::InSubquery {
217 subquery: query, ..
218 } => {
219 collect_table_names_inner(query, out);
220 }
221 LogicalPlan::CteRef { .. } | LogicalPlan::Values { .. } | LogicalPlan::Empty => {}
222 }
223}
224
225fn predicate_refs_only(predicate: &str, owns: &[String], excludes: &[String]) -> bool {
228 let refs_any_exclude = excludes.iter().any(|name| {
229 predicate.contains(&format!("{name}.")) || predicate_has_bare_word(predicate, name)
230 });
231 if refs_any_exclude {
232 return false;
233 }
234 owns.iter().any(|name| {
235 predicate.contains(&format!("{name}.")) || predicate_has_bare_word(predicate, name)
236 })
237}
238
239fn predicate_has_bare_word(text: &str, word: &str) -> bool {
241 if word.is_empty() {
242 return false;
243 }
244 let bytes = text.as_bytes();
245 let word_bytes = word.as_bytes();
246 let wlen = word_bytes.len();
247 let mut start = 0;
248 while start + wlen <= bytes.len() {
249 if &bytes[start..start + wlen] == word_bytes {
250 let before_ok = start == 0
251 || !bytes
252 .get(start - 1)
253 .map(|b| b.is_ascii_alphanumeric() || *b == b'_')
254 .unwrap_or(false);
255 let after_ok = (start + wlen) >= bytes.len()
256 || !bytes
257 .get(start + wlen)
258 .map(|b| b.is_ascii_alphanumeric() || *b == b'_')
259 .unwrap_or(false);
260 if before_ok && after_ok {
261 return true;
262 }
263 }
264 start += 1;
265 }
266 false
267}
268
269fn extract_identifier_names(predicate: &str) -> Vec<String> {
271 let dialect = GenericDialect {};
272 let result = Parser::new(&dialect)
273 .try_with_sql(predicate)
274 .ok()
275 .and_then(|mut p| p.parse_expr().ok());
276
277 let mut names = Vec::new();
278 if let Some(expr) = result {
279 collect_identifiers_from_expr(&expr, &mut names);
280 }
281 names
282}
283
284fn collect_identifiers_from_expr(expr: &sqlparser::ast::Expr, out: &mut Vec<String>) {
285 use sqlparser::ast::Expr;
286 match expr {
287 Expr::Identifier(id) => out.push(id.value.clone()),
288 Expr::CompoundIdentifier(parts) => {
289 for part in parts {
290 out.push(part.value.clone());
291 }
292 }
293 Expr::BinaryOp { left, right, .. } => {
294 collect_identifiers_from_expr(left, out);
295 collect_identifiers_from_expr(right, out);
296 }
297 Expr::UnaryOp { expr, .. } => collect_identifiers_from_expr(expr, out),
298 Expr::Nested(inner) => collect_identifiers_from_expr(inner, out),
299 Expr::Between {
300 expr, low, high, ..
301 } => {
302 collect_identifiers_from_expr(expr, out);
303 collect_identifiers_from_expr(low, out);
304 collect_identifiers_from_expr(high, out);
305 }
306 Expr::InList { expr, list, .. } => {
307 collect_identifiers_from_expr(expr, out);
308 for item in list {
309 collect_identifiers_from_expr(item, out);
310 }
311 }
312 _ => {}
313 }
314}