sql: Put DISTINCT before leaf project node

If a (non-injective) expression returns the same value given two
different values as input, we want only one row rather than two -
previously, since the DISTINCT node was placed *before* the leaf project
node, we'd do a DISTINCT on the input columns but not the actual result
values of the leaf projected expressions. Now, we always put DISTINCT
last, which fixes the issue.

Fixes: ENG-2959
Change-Id: I2b13f3ac692fb56f112d4726df2b6c72688f6ec8
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/4801
Tested-by: Buildkite CI
Reviewed-by: Dan Wilbanks <dan@readyset.io>

logictests/select_distinct.test

@@ -22,6 +22,9 @@ CREATE TABLE righttable (
   value INTEGER
 )
 
+statement ok
+CREATE TABLE date_table (val DATE);
+
 statement ok
 INSERT INTO distinct_test (value, number) VALUES (1, 4)
 
@@ -70,6 +73,9 @@ INSERT INTO righttable (lefttable_id, value) VALUES (1, 20)
 statement ok
 INSERT INTO righttable (lefttable_id, value) VALUES (2, 20)
 
+statement ok
+INSERT INTO date_table (val) VALUES ('2020-11-27'), ('2020-07-17');
+
 query II nosort
 SELECT DISTINCT value as v, number as n FROM distinct_test
 ----
@@ -143,8 +149,13 @@ SELECT DISTINCT number FROM delete_test
 statement ok
 INSERT INTO delete_test (id, number) VALUES (1, 7)
 
-query I nosort
+query I rowsort
 SELECT DISTINCT number FROM delete_test
 ----
 6
 7
+
+query nosort
+SELECT DISTINCT dayofweek(val) FROM date_table
+----
+6

readyset-server/src/controller/sql/mir/mod.rs

@@ -1772,6 +1772,19 @@ impl SqlToMirConverter {
                 }
             }
 
+            final_node = self.make_project_node(
+                query_name,
+                if leaf_behavior.should_make_leaf() {
+                    format!("q_{:x}_project", query_graph.signature().hash).into()
+                } else {
+                    query_name.clone()
+                },
+                final_node,
+                projected_columns.clone(),
+                projected_expressions,
+                projected_literals,
+            );
+
             if query_graph.distinct {
                 let name = if leaf_behavior.should_make_leaf() {
                     format!(
@@ -1788,28 +1801,13 @@ impl SqlToMirConverter {
                     )
                     .into()
                 };
-                let distinct_node = self.make_distinct_node(
-                    query_name,
-                    name,
-                    final_node,
-                    projected_columns.clone(),
-                );
-                final_node = distinct_node;
+                // This needs to go *after* the leaf project node, so that we get distinct values
+                // for the results of expressions (which might not be injective) rather than for the
+                // *inputs* to those expressions
+                final_node =
+                    self.make_distinct_node(query_name, name, final_node, self.columns(final_node));
             }
 
-            let leaf_project_node = self.make_project_node(
-                query_name,
-                if leaf_behavior.should_make_leaf() {
-                    format!("q_{:x}_project", query_graph.signature().hash).into()
-                } else {
-                    query_name.clone()
-                },
-                final_node,
-                projected_columns,
-                projected_expressions,
-                projected_literals,
-            );
-
             if leaf_behavior.should_make_leaf() {
                 // We are supposed to add a `Leaf` node keyed on the query parameters. For purely
                 // internal views (e.g., subqueries), this is not set.
@@ -1839,7 +1837,7 @@ impl SqlToMirConverter {
                 // projected by this projection, so we can then add another projection that returns
                 // the columns in the correct order
                 let mut project_order = Vec::with_capacity(returned_cols.len());
-                let parent_columns = self.mir_graph.columns(leaf_project_node);
+                let parent_columns = self.mir_graph.columns(final_node);
                 for col in returned_cols.iter() {
                     if let Some(c) = parent_columns.iter().find(|c| col.cmp(c).is_eq()) {
                         project_order.push(c.clone());
@@ -1861,7 +1859,7 @@ impl SqlToMirConverter {
                     } else {
                         query_name.clone()
                     },
-                    leaf_project_node,
+                    final_node,
                     project_order,
                     vec![],
                     vec![],
@@ -1929,12 +1927,12 @@ impl SqlToMirConverter {
                             format!("{}_view_key", query_name.display_unquoted()).into(),
                             MirNodeInner::ViewKey { key },
                         ),
-                        &[leaf_project_node],
+                        &[final_node],
                     )
                 } else if let Ok(placeholders) = unsupported_placeholders.try_into() {
                     return Err(ReadySetError::UnsupportedPlaceholders { placeholders });
                 } else {
-                    leaf_project_node
+                    final_node
                 }
             }
         };