Repository: tobymao/sqlglot
Issue category: Feature Request

Add `MINUS` keyword support for `Spark` and related dialects
`MINUS` is a supported keyword in Spark and Databricks, but fails in sqlglot:


```python
from sqlglot import parse as parse_sql

def test_parses_minus_statement(migration_index):
    query = "SELECT * FROM old.things MINUS SELECT * FROM old.stuff"
    statements = parse_sql(query, read='databricks')
```

Throws:
```
        if self.error_level == ErrorLevel.IMMEDIATE:
>           raise error
E           sqlglot.errors.ParseError: Invalid expression / Unexpected token. Line 1, Col: 37.
E             SELECT * FROM old.things MINUS SELECT * FROM old.stuff
```



Functions to edit:
sqlglot/dialects/spark.py:Spark

Patch:
diff --git a/sqlglot/dialects/spark.py b/sqlglot/dialects/spark.py
index 410c12e010..d7c788f5c0 100644
--- a/sqlglot/dialects/spark.py
+++ b/sqlglot/dialects/spark.py
@@ -7,6 +7,7 @@
 from sqlglot.dialects.hive import _build_with_ignore_nulls
 from sqlglot.dialects.spark2 import Spark2, temporary_storage_provider, _build_as_cast
 from sqlglot.helper import ensure_list, seq_get
+from sqlglot.tokens import TokenType
 from sqlglot.transforms import (
     ctas_with_tmp_tables_to_create_tmp_view,
     remove_unique_constraints,
@@ -101,6 +102,13 @@ class Tokenizer(Spark2.Tokenizer):
             for prefix in ("r", "R")
         ]
 
+        KEYWORDS = {
+            **Spark2.Tokenizer.KEYWORDS,
+            # as per https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-setops.html
+            # MINUS is an alias for EXCEPT
+            "MINUS": TokenType.EXCEPT,
+        }
+
     class Parser(Spark2.Parser):
         FUNCTIONS = {
             **Spark2.Parser.FUNCTIONS,

Test patch: diff --git a/tests/dialects/test_spark.py b/tests/dialects/test_spark.py
index dac2ead3f9..03ee228981 100644
--- a/tests/dialects/test_spark.py
+++ b/tests/dialects/test_spark.py
@@ -574,7 +574,8 @@ def test_spark(self):
         )
 
         self.validate_all(
-            "CAST(x AS TIMESTAMP)", read={"trino": "CAST(x AS TIMESTAMP(6) WITH TIME ZONE)"}
+            "CAST(x AS TIMESTAMP)",
+            read={"trino": "CAST(x AS TIMESTAMP(6) WITH TIME ZONE)"},
         )
         self.validate_all(
             "SELECT DATE_ADD(my_date_column, 1)",
@@ -822,8 +823,22 @@ def test_strip_modifiers(self):
                     self.assertEqual(query.sql(name), without_modifiers)
 
     def test_schema_binding_options(self):
-        for schema_binding in ("BINDING", "COMPENSATION", "TYPE EVOLUTION", "EVOLUTION"):
+        for schema_binding in (
+            "BINDING",
+            "COMPENSATION",
+            "TYPE EVOLUTION",
+            "EVOLUTION",
+        ):
             with self.subTest(f"Test roundtrip of VIEW schema binding {schema_binding}"):
                 self.validate_identity(
                     f"CREATE VIEW emp_v WITH SCHEMA {schema_binding} AS SELECT * FROM emp"
                 )
+
+    def test_minus(self):
+        self.validate_all(
+            "SELECT * FROM db.table1 MINUS SELECT * FROM db.table2",
+            write={
+                "spark": "SELECT * FROM db.table1 EXCEPT SELECT * FROM db.table2",
+                "databricks": "SELECT * FROM db.table1 EXCEPT SELECT * FROM db.table2",
+            },
+        )