Repository: prql/prql
Pull request #503 (2022-05-16 20:59:39)

Distinct


Patch:
diff --git a/prql-compiler/src/ast/ast_fold.rs b/prql-compiler/src/ast/ast_fold.rs
index 1a01563f..a59a529c 100644
--- a/prql-compiler/src/ast/ast_fold.rs
+++ b/prql-compiler/src/ast/ast_fold.rs
@@ -70,15 +70,15 @@ pub trait AstFold {
             .map(|c| self.fold_column_sort(c))
             .try_collect()
     }
-    fn fold_select(&mut self, select: Select) -> Result<Select> {
-        fold_select(self, select)
-    }
     fn fold_join_filter(&mut self, f: JoinFilter) -> Result<JoinFilter> {
         fold_join_filter(self, f)
     }
     fn fold_type(&mut self, t: Type) -> Result<Type> {
         fold_type(self, t)
     }
+    fn fold_windowed(&mut self, windowed: Windowed) -> Result<Windowed> {
+        fold_windowed(self, windowed)
+    }
 }
 
 pub fn fold_item<T: ?Sized + AstFold>(fold: &mut T, item: Item) -> Result<Item> {
@@ -110,15 +110,7 @@ pub fn fold_item<T: ?Sized + AstFold>(fold: &mut T, item: Item) -> Result<Item>
         Item::FuncDef(func) => Item::FuncDef(fold.fold_func_def(func)?),
         Item::FuncCall(func_call) => Item::FuncCall(fold.fold_func_call(func_call)?),
         Item::Table(table) => Item::Table(fold.fold_table(table)?),
-        Item::Windowed(window) => Item::Windowed(Windowed {
-            expr: Box::new(fold.fold_node(*window.expr)?),
-            group: fold.fold_nodes(window.group)?,
-            sort: fold.fold_column_sorts(window.sort)?,
-            window: {
-                let (kind, range) = window.window;
-                (kind, fold_range(fold, range)?)
-            },
-        }),
+        Item::Windowed(window) => Item::Windowed(fold.fold_windowed(window)?),
         Item::Type(t) => Item::Type(fold.fold_type(t)?),
         // None of these capture variables, so we don't need to replace
         // them.
@@ -126,7 +118,19 @@ pub fn fold_item<T: ?Sized + AstFold>(fold: &mut T, item: Item) -> Result<Item>
     })
 }
 
-fn fold_range<F: ?Sized + AstFold>(fold: &mut F, Range { start, end }: Range) -> Result<Range> {
+pub fn fold_windowed<F: ?Sized + AstFold>(fold: &mut F, window: Windowed) -> Result<Windowed> {
+    Ok(Windowed {
+        expr: Box::new(fold.fold_node(*window.expr)?),
+        group: fold.fold_nodes(window.group)?,
+        sort: fold.fold_column_sorts(window.sort)?,
+        window: {
+            let (kind, range) = window.window;
+            (kind, fold_range(fold, range)?)
+        },
+    })
+}
+
+pub fn fold_range<F: ?Sized + AstFold>(fold: &mut F, Range { start, end }: Range) -> Result<Range> {
     Ok(Range {
         start: fold_optional_box(fold, start)?,
         end: fold_optional_box(fold, end)?,
@@ -174,47 +178,41 @@ pub fn fold_transform<T: ?Sized + AstFold>(
     fold: &mut T,
     transformation: Transform,
 ) -> Result<Transform> {
-    match transformation {
-        Transform::From(table) => Ok(Transform::From(fold.fold_table_ref(table)?)),
+    Ok(match transformation {
+        Transform::From(table) => Transform::From(fold.fold_table_ref(table)?),
 
-        Transform::Derive(assigns) => Ok(Transform::Derive(fold.fold_nodes(assigns)?)),
-        Transform::Select(assigns) => Ok(Transform::Select(fold.fold_nodes(assigns)?)),
-        Transform::Aggregate { assigns, by } => Ok(Transform::Aggregate {
+        Transform::Derive(assigns) => Transform::Derive(fold.fold_nodes(assigns)?),
+        Transform::Select(assigns) => Transform::Select(fold.fold_nodes(assigns)?),
+        Transform::Aggregate { assigns, by } => Transform::Aggregate {
             assigns: fold.fold_nodes(assigns)?,
             by: fold.fold_nodes(by)?,
-        }),
+        },
 
-        Transform::Filter(f) => Ok(Transform::Filter(Box::new(fold.fold_node(*f)?))),
-        Transform::Sort(items) => Ok(Transform::Sort(fold.fold_column_sorts(items)?)),
-        Transform::Join { side, with, filter } => Ok(Transform::Join {
+        Transform::Filter(f) => Transform::Filter(Box::new(fold.fold_node(*f)?)),
+        Transform::Sort(items) => Transform::Sort(fold.fold_column_sorts(items)?),
+        Transform::Join { side, with, filter } => Transform::Join {
             side,
             with: fold.fold_table_ref(with)?,
             filter: fold.fold_join_filter(filter)?,
-        }),
-        Transform::Group { by, pipeline } => Ok(Transform::Group {
+        },
+        Transform::Group { by, pipeline } => Transform::Group {
             by: fold.fold_nodes(by)?,
             pipeline: Box::new(fold.fold_node(*pipeline)?),
-        }),
+        },
         Transform::Window {
             kind,
             range,
             pipeline,
-        } => Ok(Transform::Window {
+        } => Transform::Window {
             range: fold_range(fold, range)?,
             kind,
             pipeline: Box::new(fold.fold_node(*pipeline)?),
-        }),
-        // TODO: generalize? Or this never changes?
-        Transform::Take(_) => Ok(transformation),
-    }
-}
-
-pub fn fold_select<T: ?Sized + AstFold>(fold: &mut T, select: Select) -> Result<Select> {
-    Ok(Select {
-        assigns: fold.fold_nodes(select.assigns)?,
-        group: fold.fold_nodes(select.group)?,
-        window: select.window.map(|x| fold.fold_nodes(x)).transpose()?,
-        sort: select.sort.map(|x| fold.fold_nodes(x)).transpose()?,
+        },
+        Transform::Take { by, range } => Transform::Take {
+            range: fold_range(fold, range)?,
+            by: fold.fold_nodes(by)?,
+        },
+        Transform::Unique => Transform::Unique,
     })
 }
 
diff --git a/prql-compiler/src/ast/item.rs b/prql-compiler/src/ast/item.rs
index 0a8c8a9a..1ca94ad9 100644
--- a/prql-compiler/src/ast/item.rs
+++ b/prql-compiler/src/ast/item.rs
@@ -1,7 +1,5 @@
-use std::{
-    collections::HashMap,
-    fmt::{Display, Write},
-};
+use std::collections::HashMap;
+use std::fmt::{Display, Write};
 
 use anyhow::anyhow;
 use enum_as_inner::EnumAsInner;
@@ -83,10 +81,12 @@ pub enum InterpolateItem {
     Expr(Box<Node>),
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
-pub struct Range {
-    pub start: Option<Box<Node>>,
-    pub end: Option<Box<Node>>,
+/// Inclusive-inclusive range.
+/// Missing bound means unbounded range.
+#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
+pub struct Range<T = Box<Node>> {
+    pub start: Option<T>,
+    pub end: Option<T>,
 }
 
 impl Range {
@@ -102,6 +102,17 @@ impl Range {
         let end = end.map(|x| Box::new(Node::from(Item::Literal(Literal::Integer(x)))));
         Range { start, end }
     }
+
+    pub fn into_int(self) -> Result<Range<i64>> {
+        fn cast_bound(bound: Node) -> Result<i64> {
+            Ok(bound.item.into_literal()?.into_integer()?)
+        }
+
+        Ok(Range {
+            start: self.start.map(|b| cast_bound(*b)).transpose()?,
+            end: self.end.map(|b| cast_bound(*b)).transpose()?,
+        })
+    }
 }
 
 // I could imagine there being a wrapper of this to represent "2 days 3 hours".
diff --git a/prql-compiler/src/ast/mod.rs b/prql-compiler/src/ast/mod.rs
index 52cb3e88..5f65df38 100644
--- a/prql-compiler/src/ast/mod.rs
+++ b/prql-compiler/src/ast/mod.rs
@@ -35,6 +35,10 @@ pub struct Node {
     pub span: Option<Span>,
     #[serde(skip)]
     pub declared_at: Option<usize>,
+
+    /// Is true when containing window functions
+    #[serde(skip)]
+    pub is_complex: bool,
 }
 
 impl Node {
@@ -111,6 +115,7 @@ impl From<Item> for Node {
             item,
             span: None,
             declared_at: None,
+            is_complex: false,
         }
     }
 }
diff --git a/prql-compiler/src/ast/query.rs b/prql-compiler/src/ast/query.rs
index af94d806..497cbe5f 100644
--- a/prql-compiler/src/ast/query.rs
+++ b/prql-compiler/src/ast/query.rs
@@ -40,7 +40,10 @@ pub enum Transform {
         by: Vec<Node>,
     },
     Sort(Vec<ColumnSort<Node>>),
-    Take(Range),
+    Take {
+        range: Range,
+        by: Vec<Node>,
+    },
     Join {
         side: JoinSide,
         with: TableRef,
@@ -55,6 +58,7 @@ pub enum Transform {
         range: Range,
         pipeline: Box<Node>,
     },
+    Unique, // internal only, can be expressed with group & take
 }
 
 #[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
@@ -62,25 +66,6 @@ pub enum WindowKind {
     Rows,
     Range,
 }
-#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
-pub struct Select {
-    pub assigns: Vec<Node>,
-    pub group: Vec<Node>,
-    pub window: Option<Vec<Node>>,
-    pub sort: Option<Vec<Node>>,
-}
-
-impl Select {
-    pub fn new(assigns: Vec<Node>) -> Self {
-        Select {
-            assigns,
-            group: Vec::new(),
-            window: None,
-            sort: None,
-        }
-    }
-}
-
 #[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
 pub struct TableRef {
     pub name: String,
diff --git a/prql-compiler/src/semantic/complexity.rs b/prql-compiler/src/semantic/complexity.rs
new file mode 100644
index 00000000..68898c12
--- /dev/null
+++ b/prql-compiler/src/semantic/complexity.rs
@@ -0,0 +1,69 @@
+use anyhow::Result;
+
+use super::*;
+use crate::ast::ast_fold::{fold_transform, AstFold};
+use crate::ast::*;
+
+/// sets `node.is_complex` of some nodes in AST
+pub fn determine_complexity(nodes: Vec<Node>, context: &Context) -> Vec<Node> {
+    let mut d = DetermineComplex {
+        is_complex: false,
+        context,
+    };
+
+    d.fold_nodes(nodes).unwrap()
+}
+
+struct DetermineComplex<'a> {
+    is_complex: bool,
+    context: &'a Context,
+}
+
+impl<'a> AstFold for DetermineComplex<'a> {
+    fn fold_node(&mut self, mut node: Node) -> Result<Node> {
+        node.item = self.fold_item(node.item)?;
+        node.is_complex = self.is_complex;
+
+        if !node.is_complex {
+            if let Some(declared_at) = node.declared_at {
+                let decl = self.context.declarations[declared_at].0.clone();
+                if let Declaration::Expression(expr) = decl {
+                    let expr = self.fold_node(*expr).unwrap();
+                    node.is_complex = expr.is_complex;
+                }
+            }
+        }
+
+        Ok(node)
+    }
+
+    fn fold_nodes(&mut self, nodes: Vec<Node>) -> Result<Vec<Node>> {
+        let res: Vec<_> = nodes
+            .into_iter()
+            .map(|node| {
+                self.is_complex = false;
+                self.fold_node(node).unwrap()
+            })
+            .collect();
+
+        self.is_complex = res.iter().any(|n| n.is_complex);
+        Ok(res)
+    }
+
+    fn fold_transform(&mut self, transform: Transform) -> Result<Transform> {
+        // fold only filter transforms (other don't need is_complex)
+        match transform {
+            Transform::Filter(_) => fold_transform(self, transform),
+            _ => Ok(transform),
+        }
+    }
+
+    fn fold_type(&mut self, t: Type) -> Result<Type> {
+        Ok(t)
+    }
+
+    fn fold_windowed(&mut self, windowed: Windowed) -> Result<Windowed> {
+        self.is_complex = true;
+        Ok(windowed)
+    }
+}
diff --git a/prql-compiler/src/semantic/frame.rs b/prql-compiler/src/semantic/frame.rs
index 5e5ffc83..de8b7b6e 100644
--- a/prql-compiler/src/semantic/frame.rs
+++ b/prql-compiler/src/semantic/frame.rs
@@ -117,7 +117,7 @@ impl Frame {
             Transform::Sort(sort) => {
                 self.sort = extract_sorts(sort)?;
             }
-            Transform::Filter(_) | Transform::Take(_) => {}
+            Transform::Filter(_) | Transform::Take { .. } | Transform::Unique => {}
         }
         Ok(())
     }
diff --git a/prql-compiler/src/semantic/mod.rs b/prql-compiler/src/semantic/mod.rs
index cfe6f013..5d586b09 100644
--- a/prql-compiler/src/semantic/mod.rs
+++ b/prql-compiler/src/semantic/mod.rs
@@ -1,3 +1,4 @@
+mod complexity;
 mod context;
 mod frame;
 mod reporting;
diff --git a/prql-compiler/src/semantic/resolver.rs b/prql-compiler/src/semantic/resolver.rs
index d712d5d9..f9f13cda 100644
--- a/prql-compiler/src/semantic/resolver.rs
+++ b/prql-compiler/src/semantic/resolver.rs
@@ -7,6 +7,7 @@ use crate::ast::ast_fold::*;
 use crate::error::{Error, Reason, WithErrorInfo};
 use crate::{ast::*, Declaration};
 
+use super::complexity::determine_complexity;
 use super::frame::extract_sorts;
 use super::transforms;
 use super::Context;
@@ -23,6 +24,8 @@ pub fn resolve(nodes: Vec<Node>, context: Option<Context>) -> Result<(Vec<Node>,
 
     let nodes = resolver.fold_nodes(nodes)?;
 
+    let nodes = determine_complexity(nodes, &resolver.context);
+
     Ok((nodes, resolver.context))
 }
 
@@ -332,24 +335,23 @@ impl Resolver {
             .validate_function_call(node.declared_at, func_call, within_curry)
             .with_span(node.span)?;
 
-        node.item = {
-            let return_type = func_def.return_type.as_ref();
-            if Some(&Type::frame()) <= return_type {
-                // cast if this is a transform
-                let transform = transforms::cast_transform(func_call, node.span)?;
+        let return_type = func_def.return_type.as_ref();
+        if Some(&Type::frame()) <= return_type {
+            // cast if this is a transform
+            let transform = transforms::cast_transform(func_call, node.span)?;
 
-                Item::Transform(self.fold_transform(transform)?)
-            } else {
-                let mut item = Item::FuncCall(self.fold_func_call(func_call)?);
+            node.item = Item::Transform(self.fold_transform(transform)?)
+        } else {
+            let func_call = Item::FuncCall(self.fold_func_call(func_call)?);
 
-                // wrap into windowed
-                if !self.within_aggregate && Some(&Type::column()) <= return_type {
-                    item = self.wrap_into_windowed(item, node.declared_at);
-                    node.declared_at = None;
-                }
-                item
+            // wrap into windowed
+            if !self.within_aggregate && Some(&Type::column()) <= return_type {
+                node.item = self.wrap_into_windowed(func_call, node.declared_at);
+                node.declared_at = None;
+            } else {
+                node.item = func_call;
             }
-        };
+        }
 
         Ok(node)
     }
@@ -394,7 +396,7 @@ impl Resolver {
             | Transform::Window { .. } => {
                 // ok
             }
-            Transform::Aggregate { by, .. } => {
+            Transform::Aggregate { by, .. } | Transform::Take { by, .. } => {
                 *by = (self.within_group)
                     .iter()
                     .map(|id| Node::new_ident("<ref>", *id))
@@ -666,38 +668,48 @@ mod tests {
 
     #[test]
     fn test_applying_group_context() {
-        let prql = r#"
+        assert_snapshot!(parse(r#"
         from employees
         group last_name (
             sort first_name
             take 1
         )
-        "#;
-        let result = parse(prql).and_then(resolve_and_translate);
-        assert!(result.is_err());
+        "#).and_then(resolve_and_translate).unwrap(), @r###"
+        SELECT
+          DISTINCT employees.*
+        FROM
+          employees
+        "###);
 
-        let prql = r#"
+        let res = parse(
+            r#"
+        from employees
+        group last_name (
+            group last_name ( | aaa )
+        )
+        "#,
+        )
+        .and_then(resolve_and_translate);
+        assert!(res.is_err());
+
+        assert_snapshot!(parse(r#"
         from employees
         group last_name (
             select first_name
         )
-        "#;
-        let result = parse(prql).and_then(resolve_and_translate).unwrap();
-        assert_snapshot!(result, @r###"
+        "#).and_then(resolve_and_translate).unwrap(), @r###"
         SELECT
           first_name
         FROM
           employees
         "###);
 
-        let prql = r#"
+        assert_snapshot!(parse(r#"
         from employees
         group last_name (
             aggregate count
         )
-        "#;
-        let result = parse(prql).and_then(resolve_and_translate).unwrap();
-        assert_snapshot!(result, @r###"
+        "#).and_then(resolve_and_translate).unwrap(), @r###"
         SELECT
           last_name,
           COUNT(*)
diff --git a/prql-compiler/src/semantic/transforms.rs b/prql-compiler/src/semantic/transforms.rs
index d2e915a8..4d768d46 100644
--- a/prql-compiler/src/semantic/transforms.rs
+++ b/prql-compiler/src/semantic/transforms.rs
@@ -118,13 +118,12 @@ pub fn cast_transform(func_call: FuncCall, span: Option<Span>) -> Result<Transfo
         "take" => {
             let ([expr], []) = unpack(func_call, [])?;
 
-            match expr.discard_name()?.item {
-                Item::Literal(Literal::Integer(n)) => {
-                    Transform::Take(Range::from_ints(None, Some(n)))
-                }
-                Item::Range(range) => Transform::Take(range),
+            let range = match expr.discard_name()?.item {
+                Item::Literal(Literal::Integer(n)) => Range::from_ints(None, Some(n)),
+                Item::Range(range) => range,
                 _ => unimplemented!(),
-            }
+            };
+            Transform::Take { range, by: vec![] }
         }
         "join" => {
             let ([with, filter], [side]) = unpack(func_call, ["side"])?;
@@ -327,15 +326,13 @@ mod tests {
         let query = parse(
             "
         from c_invoice
+        select invoice_no
         group invoice_no (
             take 1
         )
         ",
         )
         .unwrap();
-        // TODO: this test
-        assert!(resolve(query.nodes, context.clone()).is_err());
-        /*
         let (result, _) = resolve(query.nodes, context.clone()).unwrap();
         assert_yaml_snapshot!(result, @r###"
         ---
@@ -346,7 +343,10 @@ mod tests {
                   From:
                     name: c_invoice
                     alias: ~
-                    declared_at: 58
+                    declared_at: 78
+              - Transform:
+                  Select:
+                    - Ident: invoice_no
               - Transform:
                   Group:
                     by:
@@ -356,9 +356,15 @@ mod tests {
                         value: ~
                         functions:
                           - Transform:
-                              Take: 1
+                              Take:
+                                range:
+                                  start: ~
+                                  end:
+                                    Literal:
+                                      Integer: 1
+                                by:
+                                  - Ident: "<ref>"
         "###);
-        */
 
         // oops, two arguments #339
         let query = parse(
@@ -402,7 +408,7 @@ mod tests {
                   From:
                     name: c_invoice
                     alias: ~
-                    declared_at: 80
+                    declared_at: 78
               - Transform:
                   Group:
                     by:
@@ -448,7 +454,7 @@ mod tests {
                   From:
                     name: invoices
                     alias: ~
-                    declared_at: 80
+                    declared_at: 78
               - Transform:
                   Sort:
                     - direction: Asc
diff --git a/prql-compiler/src/sql/distinct.rs b/prql-compiler/src/sql/distinct.rs
new file mode 100644
index 00000000..6eed45f1
--- /dev/null
+++ b/prql-compiler/src/sql/distinct.rs
@@ -0,0 +1,49 @@
+use anyhow::Result;
+
+use crate::ast::{ast_fold::AstFold, *};
+
+pub fn take_to_distinct(nodes: Vec<Node>) -> Result<Vec<Node>> {
+    let mut d = DistinctMaker {};
+    d.fold_nodes(nodes)
+}
+/// Creates [Transform::Unique] from [Transform::Take]
+struct DistinctMaker {}
+
+impl AstFold for DistinctMaker {
+    fn fold_nodes(&mut self, nodes: Vec<Node>) -> Result<Vec<Node>> {
+        let mut res = Vec::new();
+
+        for node in nodes {
+            match node.item {
+                Item::Transform(Transform::Take { ref by, .. }) if by.is_empty() => {
+                    res.push(node);
+                }
+
+                Item::Transform(Transform::Take { range, .. }) => {
+                    let range = range.into_int()?;
+                    let take_only_first =
+                        range.start.unwrap_or(1) == 1 && matches!(range.end, Some(1));
+
+                    if take_only_first {
+                        // TODO: use distinct only if `by == all columns in frame`
+                        res.push(Item::Transform(Transform::Unique).into());
+                    }
+
+                    // TODO: else
+                    //  return `
+                    //    derive _rn = row number over (order by)
+                    //    filter (_rn | in range)
+                    //  `
+                }
+                _ => {
+                    res.push(node);
+                }
+            }
+        }
+        Ok(res)
+    }
+
+    fn fold_func_def(&mut self, function: FuncDef) -> Result<FuncDef> {
+        Ok(function)
+    }
+}
diff --git a/prql-compiler/src/sql/materializer.rs b/prql-compiler/src/sql/materializer.rs
index 664a6a48..3880fffb 100644
--- a/prql-compiler/src/sql/materializer.rs
+++ b/prql-compiler/src/sql/materializer.rs
@@ -878,10 +878,12 @@ take 20
               declared_at: 37
         - Transform:
             Take:
-              start: ~
-              end:
-                Literal:
-                  Integer: 20
+              range:
+                start: ~
+                end:
+                  Literal:
+                    Integer: 20
+              by: []
         "###);
         assert_yaml_snapshot!(frame.columns, @r###"
         ---
diff --git a/prql-compiler/src/sql/mod.rs b/prql-compiler/src/sql/mod.rs
index 3748dc42..7ad56f03 100644
--- a/prql-compiler/src/sql/mod.rs
+++ b/prql-compiler/src/sql/mod.rs
@@ -1,3 +1,4 @@
+mod distinct;
 mod materializer;
 mod translator;
 mod un_group;
diff --git a/prql-compiler/src/sql/stdlib.prql b/prql-compiler/src/sql/stdlib.prql
index 83aad4ae..0ab5d090 100644
--- a/prql-compiler/src/sql/stdlib.prql
+++ b/prql-compiler/src/sql/stdlib.prql
@@ -34,7 +34,7 @@ func first<column> offset column ->  s"FIRST_VALUE({column}, {offset})"
 func last<column> offset column ->  s"LAST_VALUE({column}, {offset})"
 func rank<column> ->  s"RANK()"
 func rank_dense<column> ->  s"DENSE_RANK()"
-func row_number<column> offset column ->  s"ROW_NUMBER({column}, {offset})"
+func row_number<column> ->  s"ROW_NUMBER()"
 
 # Logical functions
 func coalesce value default -> s"COALESCE({value}, {default})"
diff --git a/prql-compiler/src/sql/translator.rs b/prql-compiler/src/sql/translator.rs
index 6b68a22f..ebfca3f2 100644
--- a/prql-compiler/src/sql/translator.rs
+++ b/prql-compiler/src/sql/translator.rs
@@ -25,7 +25,7 @@ use crate::semantic::Context;
 use crate::utils::OrMap;
 
 use super::materializer::MaterializationContext;
-use super::{un_group, MaterializedFrame};
+use super::{distinct, un_group, MaterializedFrame};
 
 /// Translate a PRQL AST into a SQL string.
 pub fn translate(query: Query, context: Context) -> Result<String> {
@@ -141,9 +141,6 @@ fn table_factor_of_table_ref(table_ref: &TableRef) -> TableFactor {
 // impl Translator for
 // fn sql_query_of_atomic_table(table: AtomicTable, dialect: &Dialect) -> Result<sql_ast::Query> {
 fn sql_query_of_atomic_table(table: AtomicTable, dialect: &Dialect) -> Result<sql_ast::Query> {
-    // TODO: possibly do validation here? e.g. check there isn't more than one
-    // `from`? Or do we rely on the caller for that?
-
     let frame = table.frame.ok_or_else(|| anyhow!("frame not provided?"))?;
 
     let transforms = table.pipeline.into_transforms()?;
@@ -162,32 +159,7 @@ fn sql_query_of_atomic_table(table: AtomicTable, dialect: &Dialect) -> Result<sq
     let joins = transforms
         .iter()
         .filter(|t| matches!(t, Transform::Join { .. }))
-        .map(|t| match t {
-            Transform::Join { side, with, filter } => {
-                let constraint = match filter {
-                    JoinFilter::On(nodes) => Item::Expr(nodes.to_vec())
-                        .try_into()
-                        .map(JoinConstraint::On)?,
-                    JoinFilter::Using(nodes) => JoinConstraint::Using(
-                        nodes
-                            .iter()
-                            .map(|x| x.item.clone().try_into())
-                            .collect::<Result<Vec<_>>>()?,
-                    ),
-                };
-
-                Ok(Join {
-                    relation: table_factor_of_table_ref(with),
-                    join_operator: match *side {
-                        JoinSide::Inner => JoinOperator::Inner(constraint),
-                        JoinSide::Left => JoinOperator::LeftOuter(constraint),
-                        JoinSide::Right => JoinOperator::RightOuter(constraint),
-                        JoinSide::Full => JoinOperator::FullOuter(constraint),
-                    },
-                })
-            }
-            _ => unreachable!(),
-        })
+        .map(Join::try_from)
         .collect::<Result<Vec<_>>>()?;
     if !joins.is_empty() {
         if let Some(from) = from.last_mut() {
@@ -204,24 +176,6 @@ fn sql_query_of_atomic_table(table: AtomicTable, dialect: &Dialect) -> Result<sq
         .unwrap_or(transforms.len());
     let (before, after) = transforms.split_at(aggregate_position);
 
-    // Convert the filters in a pipeline into an Expr
-    #[allow(unstable_name_collisions)] // Same behavior as the std lib; we can remove this + itertools when that's released.
-    fn filter_of_pipeline(pipeline: &[Transform]) -> Result<Option<Expr>> {
-        let filters: Vec<Node> = pipeline
-            .iter()
-            .filter_map(|t| match t {
-                Transform::Filter(filter) => Some(*filter.clone()),
-                _ => None,
-            })
-            .intersperse(Node::from(Item::Operator("and".to_owned())))
-            .collect();
-
-        Ok(if !filters.is_empty() {
-            Some((Item::Expr(filters)).try_into()?)
-        } else {
-            None
-        })
-    }
     // Find the filters that come before the aggregation.
     let where_ = filter_of_pipeline(before)?;
     let having = filter_of_pipeline(after)?;
@@ -229,7 +183,7 @@ fn sql_query_of_atomic_table(table: AtomicTable, dialect: &Dialect) -> Result<sq
     let takes = transforms
         .iter()
         .filter_map(|t| match t {
-            Transform::Take(take) => Some(take.clone()),
+            Transform::Take { range, .. } => Some(range.clone()),
             _ => None,
         })
         .collect();
@@ -261,9 +215,11 @@ fn sql_query_of_atomic_table(table: AtomicTable, dialect: &Dialect) -> Result<sq
     };
     let dialect = dialect.handler();
 
+    let distinct = transforms.iter().any(|t| matches!(t, Transform::Unique));
+
     Ok(sql_ast::Query {
         body: SetExpr::Select(Box::new(Select {
-            distinct: false,
+            distinct,
             top: if dialect.use_top() {
                 limit.map(top_of_i64)
             } else {
@@ -302,7 +258,7 @@ fn atomic_pipelines_of_pipeline(pipeline: Pipeline) -> Result<Vec<AtomicTable>>
     // - joins (no limit),
     // - filters (for WHERE)
     // - aggregate (max 1x)
-    // - sort (max 1x)
+    // - sort (no limit)
     // - filters (for HAVING)
     // - take (no limit)
     //
@@ -310,7 +266,9 @@ fn atomic_pipelines_of_pipeline(pipeline: Pipeline) -> Result<Vec<AtomicTable>>
     //
     // So we loop through the Pipeline, and cut it into cte-sized pipelines,
     // which we'll then compose together.
-    let pipeline = un_group::un_group(pipeline.functions)?;
+    let pipeline = Ok(pipeline.functions)
+        .and_then(un_group::un_group)
+        .and_then(distinct::take_to_distinct)?;
 
     let mut counts: HashMap<&str, u32> = HashMap::new();
     let mut splits = vec![0];
@@ -330,7 +288,8 @@ fn atomic_pipelines_of_pipeline(pipeline: Pipeline) -> Result<Vec<AtomicTable>>
                     || counts.get("Sort").is_some()
                     || counts.get("Take").is_some()
             }
-            "Filter" | "Sort" => counts.get("Take").is_some(),
+            "Sort" => counts.get("Take").is_some(),
+            "Filter" => counts.get("Take").is_some() || function.is_complex,
 
             // There can be many takes, but they have to be consecutive
             // For example `take 100 | sort a | take 10` can't be one CTE.
@@ -412,23 +371,13 @@ fn prepend_with_from(pipeline: &mut Pipeline, table: Option<TableRef>) {
     }
 }
 
-type RangeI64 = core::ops::Range<Option<i64>>;
-
 /// Aggregate several ordered ranges into one, computing the intersection.
 ///
 /// Returns a tuple of `(start, end)`, where `end` is optional.
-fn range_of_ranges(ranges: Vec<Range>) -> Result<RangeI64> {
-    #[allow(clippy::boxed_local)]
-    fn cast_bound(bound: Box<Node>) -> Result<i64> {
-        Ok(bound.item.into_literal()?.into_integer()?)
-    }
-
-    let mut current = RangeI64::default();
+fn range_of_ranges(ranges: Vec<Range>) -> Result<Range<i64>> {
+    let mut current = Range::default();
     for range in ranges {
-        let mut range = RangeI64 {
-            start: range.start.map(cast_bound).transpose()?,
-            end: range.end.map(cast_bound).transpose()?,
-        };
+        let mut range = range.into_int()?;
 
         // b = b + a.start -1 (take care of 1-based index!)
         range.start = range.start.or_map(current.start, |a, b| a + b - 1);
@@ -450,6 +399,24 @@ fn range_of_ranges(ranges: Vec<Range>) -> Result<RangeI64> {
     Ok(current)
 }
 
+#[allow(unstable_name_collisions)] // Same behavior as the std lib; we can remove this + itertools when that's released.
+fn filter_of_pipeline(pipeline: &[Transform]) -> Result<Option<Expr>> {
+    let filters: Vec<Node> = pipeline
+        .iter()
+        .filter_map(|t| match t {
+            Transform::Filter(filter) => Some(*filter.clone()),
+            _ => None,
+        })
+        .intersperse(Node::from(Item::Operator("and".to_owned())))
+        .collect();
+
+    Ok(if !filters.is_empty() {
+        Some((Item::Expr(filters)).try_into()?)
+    } else {
+        None
+    })
+}
+
 fn expr_of_i64(number: i64) -> Expr {
     Expr::Value(Value::Number(
         number.to_string(),
@@ -717,6 +684,37 @@ impl TryFrom<ColumnSort> for OrderByExpr {
         })
     }
 }
+impl TryFrom<&Transform> for Join {
+    type Error = anyhow::Error;
+    fn try_from(t: &Transform) -> Result<Join> {
+        match t {
+            Transform::Join { side, with, filter } => {
+                let constraint = match filter {
+                    JoinFilter::On(nodes) => Item::Expr(nodes.to_vec())
+                        .try_into()
+                        .map(JoinConstraint::On)?,
+                    JoinFilter::Using(nodes) => JoinConstraint::Using(
+                        nodes
+                            .iter()
+                            .map(|x| x.item.clone().try_into())
+                            .collect::<Result<Vec<_>>>()?,
+                    ),
+                };
+
+                Ok(Join {
+                    relation: table_factor_of_table_ref(with),
+                    join_operator: match *side {
+                        JoinSide::Inner => JoinOperator::Inner(constraint),
+                        JoinSide::Left => JoinOperator::LeftOuter(constraint),
+                        JoinSide::Right => JoinOperator::RightOuter(constraint),
+                        JoinSide::Full => JoinOperator::FullOuter(constraint),
+                    },
+                })
+            }
+            _ => unreachable!(),
+        }
+    }
+}
 impl TryFrom<Item> for sql_ast::Ident {
     type Error = anyhow::Error;
     fn try_from(item: Item) -> Result<Self> {
@@ -1921,4 +1919,69 @@ take 20
 
         Ok(())
     }
+
+    #[test]
+    fn test_distinct() {
+        // window functions cannot materialize into where statement: CTE is needed
+        assert_display_snapshot!((resolve_and_translate(parse(r###"
+        from employees
+        derive rn = row_number
+        filter rn > 2
+        "###,
+        ).unwrap()).unwrap()), @r###"
+        WITH table_0 AS (
+          SELECT
+            employees.*,
+            ROW_NUMBER() OVER () AS rn
+          FROM
+            employees
+        )
+        SELECT
+          table_0.*
+        FROM
+          table_0
+        WHERE
+          rn > 2
+        "###);
+
+        // basic distinct
+        assert_display_snapshot!((resolve_and_translate(parse(r###"
+        from employees
+        select first_name
+        group first_name ( | take 1)
+        "###,
+        ).unwrap()).unwrap()), @r###"
+        SELECT
+          DISTINCT first_name
+        FROM
+          employees
+        "###);
+
+        // distinct on two columns
+        assert_display_snapshot!((resolve_and_translate(parse(r###"
+        from employees
+        select [first_name, last_name]
+        group [first_name, last_name] ( | take 1)
+        "###,
+        ).unwrap()).unwrap()), @r###"
+        SELECT
+          DISTINCT first_name,
+          last_name
+        FROM
+          employees
+        "###);
+
+        // TODO: this should not use DISTINCT but ROW_NUMBER and WHERE, because we want
+        // row  distinct only over first_name and last_name.
+        assert_display_snapshot!((resolve_and_translate(parse(r###"
+        from employees
+        group [first_name, last_name] ( | take 1)
+        "###,
+        ).unwrap()).unwrap()), @r###"
+        SELECT
+          DISTINCT employees.*
+        FROM
+          employees
+        "###);
+    }
 }
diff --git a/reference/src/SUMMARY.md b/reference/src/SUMMARY.md
index c1d68edb..dcd5e10a 100644
--- a/reference/src/SUMMARY.md
+++ b/reference/src/SUMMARY.md
@@ -17,8 +17,9 @@
 - [Language features](./language-features/README.md)
   - [Strings](./language-features/strings.md)
   - [F-Strings](./language-features/f-strings.md)
-  - [S-Strings](./language-features/s-strings.md)
-  - [Ranges](./language-features/ranges.md)
+  - [Distinct](./language-features/distinct.md)
+    <!-- - Ranges — `BETWEEN 1 AND 3` -> `in 1..3`? -->
+  - [Ranges]()
     <!-- - - Dates — `"2021-01-01"` -> `@2021-01-01`? And `DATE_TRUNC(foo_date, YEAR)` -> `foo_date.year`? Or -> `foo_date | as year`? Or `foo_date | to year`? -->
   - [Dates & Times](./language-features/dates_and_times.md)
     <!--   - Regex — `REGEX_MATCH(foo, "\\w{3}")` -> `foo ~ r"\w{3}"`? Or -> `regex foo r"\w{3}"`? -->
diff --git a/reference/src/language-features/distinct.md b/reference/src/language-features/distinct.md
new file mode 100644
index 00000000..22d27658
--- /dev/null
+++ b/reference/src/language-features/distinct.md
@@ -0,0 +1,58 @@
+# Distinct
+
+PRQL doesn't have a specific `distinct` keyword. Instead, use `group` and `take 1`:
+
+```prql
+from employees
+select department
+group department (
+  take 1
+)
+```
+
+or without a linebreak:
+```prql
+from employees
+select department
+group department ( | take 1)
+```
+
+> Note: `|` is here temporarily, until we finish work on function currying and how pipelines are treated.
+
+Note that `group` can contain `sort` to [select range of rows in each group](https://stackoverflow.com/questions/3800551/select-first-row-in-each-group-by-group):
+
+```prql_no_test
+# youngest employee from each department
+from employees
+group department (
+  sort age
+  take 1
+)
+```
+... which should produce ...
+```sql
+WITH table_0 = (
+  SELECT ROW_NUMBER() OVER(PARTITION BY department ORDER_BY age) as _row_number
+  FROM employees
+)
+SELECT *
+FROM table_0
+WHERE _row_number = 1
+```
+... or in Postgres dialect ...
+```sql
+SELECT DISTINCT ON (department) *
+FROM employees
+ORDER BY department, age
+```
+
+> Note: this transpilation is not yet implemented. For now you can use:
+
+```prql
+from employees
+group department (
+  sort age
+  derive rank = row_number
+)
+filter rank == 1
+```


Test patch:
diff --git a/prql-compiler/src/sql/snapshots/prql_compiler__sql__materializer__test__materialize_2.snap b/prql-compiler/src/sql/snapshots/prql_compiler__sql__materializer__test__materialize_2.snap
index 841fb2aa..6a46edce 100644
--- a/prql-compiler/src/sql/snapshots/prql_compiler__sql__materializer__test__materialize_2.snap
+++ b/prql-compiler/src/sql/snapshots/prql_compiler__sql__materializer__test__materialize_2.snap
@@ -111,8 +111,10 @@ expression: mat
             Integer: 200
 - Transform:
     Take:
-      start: ~
-      end:
-        Literal:
-          Integer: 20
+      range:
+        start: ~
+        end:
+          Literal:
+            Integer: 20
+      by: []
 
diff --git a/prql-compiler/src/sql/snapshots/prql_compiler__sql__materializer__test__replace_variables_2.snap b/prql-compiler/src/sql/snapshots/prql_compiler__sql__materializer__test__replace_variables_2.snap
index 9ee22e3c..ba9304f7 100644
--- a/prql-compiler/src/sql/snapshots/prql_compiler__sql__materializer__test__replace_variables_2.snap
+++ b/prql-compiler/src/sql/snapshots/prql_compiler__sql__materializer__test__replace_variables_2.snap
@@ -111,8 +111,10 @@ expression: "&mat"
             Integer: 200
 - Transform:
     Take:
-      start: ~
-      end:
-        Literal:
-          Integer: 20
+      range:
+        start: ~
+        end:
+          Literal:
+            Integer: 20
+      by: []
 
diff --git a/reference/tests/prql/language-features/distinct-0.prql b/reference/tests/prql/language-features/distinct-0.prql
new file mode 100644
index 00000000..ed8a37d1
--- /dev/null
+++ b/reference/tests/prql/language-features/distinct-0.prql
@@ -0,0 +1,5 @@
+from employees
+select department
+group department (
+  take 1
+)
diff --git a/reference/tests/prql/language-features/distinct-1.prql b/reference/tests/prql/language-features/distinct-1.prql
new file mode 100644
index 00000000..160f95fe
--- /dev/null
+++ b/reference/tests/prql/language-features/distinct-1.prql
@@ -0,0 +1,3 @@
+from employees
+select department
+group department ( | take 1)
diff --git a/reference/tests/prql/language-features/distinct-2.prql b/reference/tests/prql/language-features/distinct-2.prql
new file mode 100644
index 00000000..7b3428d3
--- /dev/null
+++ b/reference/tests/prql/language-features/distinct-2.prql
@@ -0,0 +1,6 @@
+from employees
+group department (
+  sort age
+  derive rank = row_number
+)
+filter rank == 1
diff --git a/reference/tests/snapshots/snapshot__run_reference_prql@distinct-0.prql.snap b/reference/tests/snapshots/snapshot__run_reference_prql@distinct-0.prql.snap
new file mode 100644
index 00000000..5e005ca3
--- /dev/null
+++ b/reference/tests/snapshots/snapshot__run_reference_prql@distinct-0.prql.snap
@@ -0,0 +1,9 @@
+---
+source: reference/tests/snapshot.rs
+expression: sql
+input_file: reference/tests/prql/language-features/distinct-0.prql
+---
+SELECT
+  DISTINCT department
+FROM
+  employees
diff --git a/reference/tests/snapshots/snapshot__run_reference_prql@distinct-1.prql.snap b/reference/tests/snapshots/snapshot__run_reference_prql@distinct-1.prql.snap
new file mode 100644
index 00000000..dfdc42d9
--- /dev/null
+++ b/reference/tests/snapshots/snapshot__run_reference_prql@distinct-1.prql.snap
@@ -0,0 +1,9 @@
+---
+source: reference/tests/snapshot.rs
+expression: sql
+input_file: reference/tests/prql/language-features/distinct-1.prql
+---
+SELECT
+  DISTINCT department
+FROM
+  employees
diff --git a/reference/tests/snapshots/snapshot__run_reference_prql@distinct-2.prql.snap b/reference/tests/snapshots/snapshot__run_reference_prql@distinct-2.prql.snap
new file mode 100644
index 00000000..5b025aaa
--- /dev/null
+++ b/reference/tests/snapshots/snapshot__run_reference_prql@distinct-2.prql.snap
@@ -0,0 +1,23 @@
+---
+source: reference/tests/snapshot.rs
+expression: sql
+input_file: reference/tests/prql/language-features/distinct-2.prql
+---
+WITH table_0 AS (
+  SELECT
+    employees.*,
+    ROW_NUMBER() OVER (
+      PARTITION BY department
+      ORDER BY
+        age ROWS BETWEEN UNBOUNDED PRECEDING
+        AND UNBOUNDED FOLLOWING
+    ) AS rank
+  FROM
+    employees
+)
+SELECT
+  table_0.*
+FROM
+  table_0
+WHERE
+  rank = 1