readyset-server: Allow computed local predicates

This allows more predicates referencing a single table within join `ON`
clauses to be treated as local to that table, which causes them to be
turned into filters between the table and the join.

For example, this was previously unsupported because the length
predicate became a "global" predicate which is unsupported for the join:

```sql
SELECT * FROM t LEFT JOIN v ON t.id = v.id AND length(v.field) > 10;
```

This pattern is not yet allowed *everywhere*, but allows predicates
which are binary expressions only referencing a single table, even if
they contain computed expressions on columns in that table.

Change-Id: I5a8ef482ee5a7e670de6b98e4713abad6a6a6964
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/11069
Tested-by: Buildkite CI
Reviewed-by: Vassili Zarouba <[email protected]>

Author: mvzink

logictests/on_clause_length_filter.test

@@ -0,0 +1,73 @@
+statement ok
+create table foo (
+  id int primary key,
+  bar_id int,
+  category text
+);
+
+statement ok
+create table bar (
+  id int primary key,
+  text_value text
+);
+
+statement ok
+insert into bar (id, text_value) values
+  (1, 'ok'),
+  (2, null),
+  (3, 'toolong'),
+  (4, 'short'),
+  (5, 'tiny');
+
+statement ok
+insert into foo (id, bar_id, category) values
+  (1, 1, 'other'),
+  (2, 2, 'other'),
+  (3, 3, 'other'),
+  (4, 4, 'other'),
+  (5, 5, 'something_else');
+
+# Baseline: IS NOT NULL in ON clause should still attach rows with long values.
+query IT rowsort
+SELECT
+  foo.id,
+  bar.text_value
+FROM
+  foo
+LEFT JOIN bar
+  ON bar.id = foo.bar_id
+  AND bar.text_value IS NOT NULL
+WHERE
+  foo.category = 'other'
+----
+1
+ok
+2
+NULL
+3
+toolong
+4
+short
+
+# Regression: computed predicate in ON clause should be treated as a right-local filter.
+query IT rowsort
+SELECT
+  foo.id,
+  bar.text_value
+FROM
+  foo
+LEFT JOIN bar
+  ON bar.id = foo.bar_id
+  AND bar.text_value IS NOT NULL
+  AND length(bar.text_value) <= 5
+WHERE
+  foo.category = 'other'
+----
+1
+ok
+2
+NULL
+3
+NULL
+4
+short

readyset-psql/tests/integration.rs

@@ -2119,6 +2119,77 @@ async fn trunc_in_trx() {
     shutdown_tx.shutdown().await;
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn left_join_on_computed_predicate_filters_right_side() {
+    let (opts, _handle, shutdown_tx) = setup().await;
+    let conn = connect(opts).await;
+
+    conn.simple_query("CREATE TABLE foo (id INT PRIMARY KEY, bar_id INT, category TEXT);")
+        .await
+        .unwrap();
+    conn.simple_query("CREATE TABLE bar (id INT PRIMARY KEY, text_value TEXT);")
+        .await
+        .unwrap();
+
+    conn.simple_query(
+        "INSERT INTO bar (id, text_value) VALUES \
+            (1, 'ok'), \
+            (2, NULL), \
+            (3, 'toolong'), \
+            (4, 'short'), \
+            (5, 'tiny');",
+    )
+    .await
+    .unwrap();
+
+    conn.simple_query(
+        "INSERT INTO foo (id, bar_id, category) VALUES \
+            (1, 1, 'other'), \
+            (2, 2, 'other'), \
+            (3, 3, 'other'), \
+            (4, 4, 'other'), \
+            (5, 5, 'something_else');",
+    )
+    .await
+    .unwrap();
+
+    sleep().await;
+
+    let rows = conn
+        .query(
+            "SELECT \
+                foo.id, \
+                bar.text_value \
+             FROM foo \
+             LEFT JOIN bar \
+               ON bar.id = foo.bar_id \
+               AND bar.text_value IS NOT NULL \
+               AND length(bar.text_value) <= 5 \
+             WHERE foo.category = 'other' \
+             ORDER BY foo.id",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    let results: Vec<(i32, Option<String>)> = rows
+        .into_iter()
+        .map(|row| (row.get(0), row.get(1)))
+        .collect();
+
+    assert_eq!(
+        results,
+        vec![
+            (1, Some("ok".to_string())),
+            (2, None),
+            (3, None),
+            (4, Some("short".to_string()))
+        ]
+    );
+
+    shutdown_tx.shutdown().await;
+}
+
 mod http_tests {
     use super::*;
     #[tokio::test(flavor = "multi_thread")]

readyset-server/src/controller/sql/query_graph.rs

@@ -14,7 +14,7 @@ use readyset_errors::{
     unsupported, unsupported_err, ReadySetResult,
 };
 use readyset_sql::analysis::visit_mut::{walk_expr, VisitorMut};
-use readyset_sql::analysis::{is_aggregate, ReferredColumns};
+use readyset_sql::analysis::{is_aggregate, ReferredColumns, ReferredTables};
 use readyset_sql::ast::{
     self, BinaryOperator, Column, Expr, FieldDefinitionExpr, FieldReference, FunctionExpr, InValue,
     ItemPlaceholder, JoinConstraint, JoinOperator, JoinRightSide, LimitClause, Literal, NullOrder,
@@ -690,9 +690,18 @@ fn classify_conditionals(
                         local.entry(table.clone()).or_default().push(ce.clone());
                     }
 
-                    // Comparisons between computed columns and literals: Store globally
+                    // Any other binary predicate: if it mentions exactly one table, it is a local
+                    // predicate for that table; otherwise, it is a global predicate.
                     _ => {
-                        global.push(ce.clone());
+                        let mut tables = ce.referred_tables().into_iter();
+                        match (tables.next(), tables.next()) {
+                            (Some(table), None) => {
+                                local.entry(table).or_default().push(ce.clone());
+                            }
+                            _ => {
+                                global.push(ce.clone());
+                            }
+                        }
                     }
                 }
             } else {
@@ -1838,6 +1847,45 @@ mod tests {
         }
     }
 
+    #[test]
+    fn computed_right_local_predicates_in_left_join() {
+        let qg = make_query_graph(
+            "SELECT foo.id, bar.text_value \
+             FROM foo \
+             LEFT JOIN bar \
+               ON bar.id = foo.bar_id \
+               AND bar.text_value IS NOT NULL \
+               AND length(bar.text_value) <= 5 \
+             WHERE foo.category = 'other'",
+        );
+
+        let join = qg.edges.get(&("foo".into(), "bar".into())).unwrap();
+
+        match join {
+            QueryGraphEdge::LeftJoin {
+                on,
+                left_local_preds,
+                right_local_preds,
+                global_preds,
+                params,
+            } => {
+                assert_eq!(
+                    *on,
+                    vec![JoinPredicate {
+                        left: Column::from("foo.bar_id"),
+                        right: Column::from("bar.id"),
+                    }]
+                );
+                assert!(left_local_preds.is_empty());
+                // Both the IS NOT NULL and the length() predicate should be treated as right-local.
+                assert_eq!(right_local_preds.len(), 2);
+                assert!(global_preds.is_empty());
+                assert!(params.is_empty());
+            }
+            QueryGraphEdge::Join { .. } => panic!("Expected left join, got {join:?}"),
+        }
+    }
+
     mod view_key {
         use super::*;