mir: Use filters with functions as join conditions

Modify `filters_to_join_keys` to pull filters with function
calls into join conditions. Injecting a projection node between
the join and its parents that evaluates the function.

Also, handle the edge case where we have a filter that couldn't
be pulled right after a cross-join. Cross joins add bogokeys to
themselves because all joins need conditions; however, the filter
doesn't expect that key and therefore receives one more column than
it expects.

One solution is to move the bogokey creation from the lowering
phase to the mir rewrite phase since we have a mir rewrite rule
that ensures all columns are pulled. We modify the existing
`add_bogokey` rewrite rule to add bogokeys to cross join. We also
change the rule's order to execute after pulling filters and before
pulling columns (the rule that makes sure that nodes account for their
parents' cols).

Fixes: REA-4801
Closes: #1372
Release-Note-Core: Readyset now pulls filter conditions with
  function calls into join keys.
Change-Id: Ib58e421e2138ef557fa45b8ccbe46d545378725c
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9084
Tested-by: Buildkite CI
Reviewed-by: Jason Brown <[email protected]>
Reviewed-by: Vassili Zarouba <[email protected]>

Author: MohamedAbdeen-rs

logictests/join_predicates_in_where_clause.test

@@ -45,3 +45,129 @@ select a.id, c.other from a join (select * from b) as c where a.b_id = c.id;
 30
 4
 50
+
+statement ok
+create table c(id int, name text);
+
+statement ok
+create table d(id int, name text);
+
+statement ok
+insert into c values (1, 'hello'), (2, 'WORLD'), (3, 'HELLO');
+
+statement ok
+insert into d values (1, 'hello'), (4, 'not world');
+
+query IT rowsort
+select c.id, d.name from c join d where c.name = d.name;
+----
+1
+hello
+
+query IT rowsort
+select c.id, d.name from c join d where lower(c.name) = d.name;
+----
+1
+hello
+3
+hello
+
+query IT rowsort
+select c.id, d.name from c join d where c.name = upper(d.name);
+----
+3
+hello
+
+query TI rowsort
+select c.name, d.id from c join d where lower(c.name) = substr(d.name, 5, 5);
+----
+WORLD
+4
+
+# Pushed function should not appear when doing select *
+query ITIT rowsort
+select * from c join d where lower(c.name) = substr(d.name, 5, 5);
+----
+2
+WORLD
+4
+not world
+
+
+query IT rowsort
+select c.id, d.name from c join d where lower(d.name) = lower(c.name);
+----
+1
+hello
+3
+hello
+
+query IT rowsort
+select c.id, d.name from c join d where upper(d.name) = c.name;
+----
+3
+hello
+
+# Test nested functions
+query IT rowsort
+select c.id, d.name from c join d where ascii(substr(lower(c.name), 1, 5)) = ascii(lower(upper(d.name)));
+----
+1
+hello
+3
+hello
+
+# Test with AND'ed conditions
+query IT rowsort
+select c.id, d.name from c join d where
+	lower(c.name) = lower(substr(d.name, 1, 5))
+	and c.id = d.id
+----
+1
+hello
+
+query IT rowsort
+select c.id, d.name from c join d
+where lower(c.name) = substr(d.name, 1, 5)
+  and ascii(substr(lower(c.name), 1, 1)) = ascii(substr(lower(d.name), 1, 1));
+----
+1
+hello
+3
+hello
+
+# A filter that can't be pushed
+query IT rowsort
+select c.id, d.name from c join d
+	where round(c.id) = d.id;
+----
+1
+hello
+
+# A filter that can be pushed, and another that can't
+query IT rowsort
+select c.id, d.name from c join d
+	where lower(c.name) = lower(substr(d.name, 1, 5))
+	and round(c.id) = d.id;
+----
+1
+hello
+
+# Test where filter is already a condition, but contains function calls
+query IT rowsort
+select c.id, d.name from c join d on lower(c.name) = lower(d.name);
+----
+1
+hello
+3
+hello
+
+# Filter uses one side of the join
+query IT rowsort
+select c.id, d.name from c join d
+	where lower(c.name) = substr(c.name, 1, 5);
+----
+1
+hello
+1
+not world

readyset-mir/src/rewrite/add_bogokey.rs

@@ -10,14 +10,7 @@ use crate::node::{MirNode, MirNodeInner};
 use crate::query::MirQuery;
 use crate::Column;
 
-/// If the given query has a Leaf but doesn't have any keys, create a key for it by adding a new
-/// node to the query that projects out a constant literal value (a "bogokey", from "bogus key") and
-/// making that the key for the query.
-///
-/// This pass will also handle ensuring that any topk or paginate nodes in leaf position in such
-/// queries have `group_by` columns, by lifting the bogokey project node over those nodes and adding
-/// the bogokey to their `group_by`
-pub(crate) fn add_bogokey_if_necessary(query: &mut MirQuery<'_>) -> ReadySetResult<()> {
+fn add_bogokey_leaf(query: &mut MirQuery<'_>) -> ReadySetResult<()> {
     match &query.leaf_node().inner {
         MirNodeInner::Leaf { keys, .. } if keys.is_empty() => {}
         _ => {
@@ -87,7 +80,83 @@ pub(crate) fn add_bogokey_if_necessary(query: &mut MirQuery<'_>) -> ReadySetResu
         &mut query.get_node_mut(node_to_insert_above).unwrap().inner
     {
         group_by.push(Column::named("bogokey"))
-    }
+    };
+
+    Ok(())
+}
+
+fn add_bogokey_join(query: &mut MirQuery<'_>) -> ReadySetResult<()> {
+    let cross_joins = query
+        .node_references()
+        .filter(|(_, node)| {
+            matches!(
+                node,
+                MirNode {
+                    inner: MirNodeInner::Join { on, .. },
+                    ..
+                } if on.is_empty()
+            )
+        })
+        .map(|(idx, _)| idx)
+        .collect::<Vec<_>>();
+
+    cross_joins
+        .iter()
+        .try_for_each(|idx| -> ReadySetResult<()> {
+            trace!(?idx, "Adding bogokey to cross join");
+
+            let ancestors = query.ancestors(*idx).unwrap();
+            for (i, ancestor) in ancestors.into_iter().enumerate() {
+                query.insert_below(
+                    ancestor,
+                    MirNode::new(
+                        format!("{}_bogo_project_{}", query.name().display_unquoted(), i).into(),
+                        MirNodeInner::Project {
+                            emit: query
+                                .graph
+                                .columns(ancestor)
+                                .into_iter()
+                                .map(ProjectExpr::Column)
+                                .chain(iter::once(ProjectExpr::Expr {
+                                    expr: Expr::Literal(0.into()),
+                                    alias: "bogokey".into(),
+                                }))
+                                .collect(),
+                        },
+                    ),
+                )?;
+            }
+
+            match &mut query.get_node_mut(*idx).unwrap().inner {
+                MirNodeInner::Join { on, project } => {
+                    on.push((Column::named("bogokey"), Column::named("bogokey")));
+                    project.push(Column::named("bogokey"));
+                }
+                _ => unreachable!(),
+            }
+
+            Ok(())
+        })?;
+
+    Ok(())
+}
+
+/// A few scenarios where a bogokey (from "bogus key") is needed:
+///
+/// If the given query has a Leaf but doesn't have any keys, create a key for it by adding a new
+/// node to the query that projects out a constant literal value (a "bogokey", from "bogus key") and
+/// making that the key for the query.
+///
+/// This pass will also handle ensuring that any topk or paginate nodes in leaf position in such
+/// queries have `group_by` columns, by lifting the bogokey project node over those nodes and adding
+/// the bogokey to their `group_by`
+///
+/// Consider a join node without a join condition (i.e. a cross-join); however, all join nodes
+/// require a join condition, so we add a bogokey projection to both of the join's parents and
+/// use that as a filter condition..
+pub(crate) fn add_bogokey_if_necessary(query: &mut MirQuery<'_>) -> ReadySetResult<()> {
+    add_bogokey_leaf(query)?;
+    add_bogokey_join(query)?;
 
     Ok(())
 }
@@ -313,4 +382,97 @@ mod tests {
             _ => panic!(),
         }
     }
+
+    #[test]
+    fn test_add_bogokey_to_cross_join_node() {
+        let query_name = Relation::from("query_needing_bogokey");
+        let mut mir_graph = MirGraph::new();
+
+        let left = mir_graph.add_node(MirNode::new(
+            "left_base".into(),
+            MirNodeInner::Base {
+                column_specs: vec![ColumnSpecification {
+                    column: ast::Column::from("a"),
+                    sql_type: SqlType::Int(None),
+                    generated: None,
+                    constraints: vec![],
+                    comment: None,
+                }],
+                primary_key: Some([Column::from("a")].into()),
+                unique_keys: Default::default(),
+            },
+        ));
+        mir_graph[left].add_owner(query_name.clone());
+
+        let right = mir_graph.add_node(MirNode::new(
+            "right_base".into(),
+            MirNodeInner::Base {
+                column_specs: vec![ColumnSpecification {
+                    column: ast::Column::from("b"),
+                    sql_type: SqlType::Int(None),
+                    generated: None,
+                    constraints: vec![],
+                    comment: None,
+                }],
+                primary_key: Some([Column::from("b")].into()),
+                unique_keys: Default::default(),
+            },
+        ));
+        mir_graph[right].add_owner(query_name.clone());
+
+        let join_node = mir_graph.add_node(MirNode::new(
+            "join_node".into(),
+            MirNodeInner::Join {
+                on: vec![], // Empty join condition (i.e. cross join)
+                project: vec![Column::named("a"), Column::named("b")],
+            },
+        ));
+        mir_graph[join_node].add_owner(query_name.clone());
+        mir_graph.add_edge(left, join_node, 0);
+        mir_graph.add_edge(right, join_node, 1);
+
+        let mut query = MirQuery::new(query_name.clone(), join_node, &mut mir_graph);
+
+        add_bogokey_if_necessary(&mut query).unwrap();
+
+        match &query.get_node(join_node).unwrap().inner {
+            MirNodeInner::Join { on, .. } => {
+                assert_eq!(on.len(), 1, "Expected a bogo key to be added");
+                assert_eq!(
+                    on[0].0.name, "bogokey",
+                    "Bogo key column name should be 'bogokey'"
+                );
+                assert_eq!(
+                    on[0].1.name, "bogokey",
+                    "Bogo key column name should be 'bogokey'"
+                );
+            }
+            _ => panic!("Leaf node is not a Join node"),
+        }
+
+        // Helper closure to validate parent projections
+        let check_projection_node = |parent_name: &str| {
+            let parent = mir_graph
+                .neighbors_directed(join_node, petgraph::Direction::Incoming)
+                .inspect(|p| println!("Parent: {:?}", mir_graph[*p]))
+                .find(|&n| mir_graph[n].name() == &Relation::from(parent_name))
+                .unwrap_or_else(|| panic!("Expected a projection node for {}", parent_name));
+
+            match &mir_graph[parent].inner {
+                MirNodeInner::Project { emit, .. } => {
+                    assert!(
+                        emit.iter().any(
+                            |c| matches!(c, ProjectExpr::Expr{ alias, ..} if alias == "bogokey")
+                        ),
+                        "{} projection should output bogokey",
+                        parent_name
+                    );
+                }
+                _ => panic!("{} parent is not a Projection node", parent_name),
+            }
+        };
+
+        check_projection_node(format!("{}_bogo_project_0", query_name.display_unquoted()).as_str());
+        check_projection_node(format!("{}_bogo_project_1", query_name.display_unquoted()).as_str());
+    }
 }