support intersect

Author: simonvandel

src/ast/query.rs

@@ -2695,7 +2695,16 @@ pub enum PipeOperator {
     /// Syntax: `|> UNION [ALL|DISTINCT] (<query>), (<query>), ...`
     ///
     /// See more at <https://cloud.google.com/bigquery/docs/reference/standard-sql/pipe-syntax#union_pipe_operator>
-    Union { 
+    Union {
+        set_quantifier: SetQuantifier,
+        queries: Vec<Box<Query>>,
+    },
+    /// Returns only the rows that are present in both the input table and the specified tables.
+    ///
+    /// Syntax: `|> INTERSECT [DISTINCT] (<query>), (<query>), ...`
+    ///
+    /// See more at <https://cloud.google.com/bigquery/docs/reference/standard-sql/pipe-syntax#intersect_pipe_operator>
+    Intersect {
         set_quantifier: SetQuantifier,
         queries: Vec<Box<Query>>,
     },
@@ -2757,12 +2766,37 @@ impl fmt::Display for PipeOperator {
             PipeOperator::Rename { mappings } => {
                 write!(f, "RENAME {}", display_comma_separated(mappings))
             }
-            PipeOperator::Union { set_quantifier, queries } => {
+            PipeOperator::Union {
+                set_quantifier,
+                queries,
+            } => {
                 write!(f, "UNION")?;
                 match set_quantifier {
                     SetQuantifier::All => write!(f, " ALL")?,
                     SetQuantifier::Distinct => write!(f, " DISTINCT")?,
-                    SetQuantifier::None => {},
+                    SetQuantifier::None => {}
+                    _ => {
+                        write!(f, " {}", set_quantifier)?;
+                    }
+                }
+                write!(f, " ")?;
+                for (i, query) in queries.iter().enumerate() {
+                    if i > 0 {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "({})", query)?;
+                }
+                Ok(())
+            }
+            PipeOperator::Intersect {
+                set_quantifier,
+                queries,
+            } => {
+                write!(f, "INTERSECT")?;
+                match set_quantifier {
+                    SetQuantifier::All => write!(f, " ALL")?,
+                    SetQuantifier::Distinct => write!(f, " DISTINCT")?,
+                    SetQuantifier::None => {}
                     _ => {
                         write!(f, " {}", set_quantifier)?;
                     }

src/parser/mod.rs

@@ -11085,6 +11085,7 @@ impl<'a> Parser<'a> {
                 Keyword::TABLESAMPLE,
                 Keyword::RENAME,
                 Keyword::UNION,
+                Keyword::INTERSECT,
             ])?;
             match kw {
                 Keyword::SELECT => {
@@ -11168,6 +11169,19 @@ impl<'a> Parser<'a> {
                     })?;
                     pipe_operators.push(PipeOperator::Union { set_quantifier, queries });
                 }
+                Keyword::INTERSECT => {
+                    // Reuse existing set quantifier parser for consistent modifier support
+                    let set_quantifier = self.parse_set_quantifier(&Some(SetOperator::Intersect));
+                    // BigQuery INTERSECT pipe operator requires parentheses around queries
+                    // Parse comma-separated list of parenthesized queries
+                    let queries = self.parse_comma_separated(|parser| {
+                        parser.expect_token(&Token::LParen)?;
+                        let query = parser.parse_query()?;
+                        parser.expect_token(&Token::RParen)?;
+                        Ok(query)
+                    })?;
+                    pipe_operators.push(PipeOperator::Intersect { set_quantifier, queries });
+                }
                 unhandled => {
                     return Err(ParserError::ParserError(format!(
                     "`expect_one_of_keywords` further up allowed unhandled keyword: {unhandled:?}"

tests/sqlparser_common.rs

@@ -15219,6 +15219,22 @@ fn parse_pipeline_operator() {
     // union pipe operator with BY NAME and multiple queries
     dialects.verified_stmt("SELECT * FROM users |> UNION BY NAME (SELECT * FROM admins), (SELECT * FROM guests)");
 
+    // intersect pipe operator (BigQuery does not support ALL modifier for INTERSECT)
+    dialects.verified_stmt("SELECT * FROM users |> INTERSECT (SELECT * FROM admins)");
+    dialects.verified_stmt("SELECT * FROM users |> INTERSECT DISTINCT (SELECT * FROM admins)");
+    
+    // intersect pipe operator with BY NAME modifier
+    dialects.verified_stmt("SELECT * FROM users |> INTERSECT BY NAME (SELECT * FROM admins)");
+    dialects.verified_stmt("SELECT * FROM users |> INTERSECT DISTINCT BY NAME (SELECT * FROM admins)");
+    
+    // intersect pipe operator with multiple queries
+    dialects.verified_stmt("SELECT * FROM users |> INTERSECT (SELECT * FROM admins), (SELECT * FROM guests)");
+    dialects.verified_stmt("SELECT * FROM users |> INTERSECT DISTINCT (SELECT * FROM admins), (SELECT * FROM guests)");
+    
+    // intersect pipe operator with BY NAME and multiple queries
+    dialects.verified_stmt("SELECT * FROM users |> INTERSECT BY NAME (SELECT * FROM admins), (SELECT * FROM guests)");
+    dialects.verified_stmt("SELECT * FROM users |> INTERSECT DISTINCT BY NAME (SELECT * FROM admins), (SELECT * FROM guests)");
+
     // many pipes
     dialects.verified_stmt(
         "SELECT * FROM CustomerOrders |> AGGREGATE SUM(cost) AS total_cost GROUP BY customer_id, state, item_type |> EXTEND COUNT(*) OVER (PARTITION BY customer_id) AS num_orders |> WHERE num_orders > 1 |> AGGREGATE AVG(total_cost) AS average GROUP BY state DESC, item_type ASC",