Repository: tobymao/sqlglot

Issue:
Redshift - add support for oracle style join markers ` (+)`
**Fully reproducible code snippet**
```python
from sqlglot import parse_one
sample_sql_stmt = "select a.foo, b.bar, a.baz from a, b where a.baz = b.baz (+)"
parse_one(sample_sql_stmt, read="redshift").sql(dialect="redshift")
```
expected:
`SELECT a.foo, b.bar, a.baz FROM a, b WHERE a.baz = b.baz (+)`

actual:
`SELECT a.foo, b.bar, a.baz FROM a, b WHERE a.baz = b.BAZ()`

**Official Documentation**
[Oracle-Style outer joins in the WHERE clause](https://docs.aws.amazon.com/redshift/latest/dg/r_WHERE_oracle_outer.html)
>For Oracle compatibility, Amazon Redshift supports the Oracle outer-join operator (+) in WHERE clause join conditions


Discussion:

Patch:
diff --git a/sqlglot/dialects/postgres.py b/sqlglot/dialects/postgres.py
index 25a02b0d..7ad1f90e 100644
--- a/sqlglot/dialects/postgres.py
+++ b/sqlglot/dialects/postgres.py
@@ -8,6 +8,7 @@ from sqlglot.dialects.dialect import (
     Dialect,
     JSON_EXTRACT_TYPE,
     any_value_to_max_sql,
+    binary_from_function,
     bool_xor_sql,
     datestrtodate_sql,
     build_formatted_time,
@@ -329,6 +330,7 @@ class Postgres(Dialect):
             "REGTYPE": TokenType.OBJECT_IDENTIFIER,
             "FLOAT": TokenType.DOUBLE,
         }
+        KEYWORDS.pop("DIV")
 
         SINGLE_TOKENS = {
             **tokens.Tokenizer.SINGLE_TOKENS,
@@ -347,6 +349,9 @@ class Postgres(Dialect):
         FUNCTIONS = {
             **parser.Parser.FUNCTIONS,
             "DATE_TRUNC": build_timestamp_trunc,
+            "DIV": lambda args: exp.cast(
+                binary_from_function(exp.IntDiv)(args), exp.DataType.Type.DECIMAL
+            ),
             "GENERATE_SERIES": _build_generate_series,
             "JSON_EXTRACT_PATH": build_json_extract_path(exp.JSONExtract),
             "JSON_EXTRACT_PATH_TEXT": build_json_extract_path(exp.JSONExtractScalar),
@@ -494,6 +499,7 @@ class Postgres(Dialect):
             exp.DateSub: _date_add_sql("-"),
             exp.Explode: rename_func("UNNEST"),
             exp.GroupConcat: _string_agg_sql,
+            exp.IntDiv: rename_func("DIV"),
             exp.JSONExtract: _json_extract_sql("JSON_EXTRACT_PATH", "->"),
             exp.JSONExtractScalar: _json_extract_sql("JSON_EXTRACT_PATH_TEXT", "->>"),
             exp.JSONBExtract: lambda self, e: self.binary(e, "#>"),
@@ -621,3 +627,12 @@ class Postgres(Dialect):
                     return f"{self.expressions(expression, flat=True)}[{values}]"
                 return "ARRAY"
             return super().datatype_sql(expression)
+
+        def cast_sql(self, expression: exp.Cast, safe_prefix: t.Optional[str] = None) -> str:
+            this = expression.this
+
+            # Postgres casts DIV() to decimal for transpilation but when roundtripping it's superfluous
+            if isinstance(this, exp.IntDiv) and expression.to == exp.DataType.build("decimal"):
+                return self.sql(this)
+
+            return super().cast_sql(expression, safe_prefix=safe_prefix)
diff --git a/sqlglot/dialects/redshift.py b/sqlglot/dialects/redshift.py
index df6985df..0dd372e7 100644
--- a/sqlglot/dialects/redshift.py
+++ b/sqlglot/dialects/redshift.py
@@ -115,6 +115,12 @@ class Redshift(Postgres):
             self._retreat(index)
             return None
 
+        def _parse_column(self) -> t.Optional[exp.Expression]:
+            column = super()._parse_column()
+            if column:
+                column.set("join_mark", self._match(TokenType.JOIN_MARKER))
+            return column
+
     class Tokenizer(Postgres.Tokenizer):
         BIT_STRINGS = []
         HEX_STRINGS = []
@@ -128,6 +134,7 @@ class Redshift(Postgres):
             "UNLOAD": TokenType.COMMAND,
             "VARBYTE": TokenType.VARBINARY,
             "MINUS": TokenType.EXCEPT,
+            "(+)": TokenType.JOIN_MARKER,
         }
         KEYWORDS.pop("VALUES")
 
@@ -148,6 +155,7 @@ class Redshift(Postgres):
         HEX_FUNC = "TO_HEX"
         # Redshift doesn't have `WITH` as part of their with_properties so we remove it
         WITH_PROPERTIES_PREFIX = " "
+        COLUMN_JOIN_MARKS_SUPPORTED = True
 
         TYPE_MAPPING = {
             **Postgres.Generator.TYPE_MAPPING,
diff --git a/sqlglot/dialects/spark.py b/sqlglot/dialects/spark.py
index e828b9bb..0212352d 100644
--- a/sqlglot/dialects/spark.py
+++ b/sqlglot/dialects/spark.py
@@ -41,6 +41,21 @@ def _build_datediff(args: t.List) -> exp.Expression:
     )
 
 
+def _build_dateadd(args: t.List) -> exp.Expression:
+    expression = seq_get(args, 1)
+
+    if len(args) == 2:
+        # DATE_ADD(startDate, numDays INTEGER)
+        # https://docs.databricks.com/en/sql/language-manual/functions/date_add.html
+        return exp.TsOrDsAdd(
+            this=seq_get(args, 0), expression=expression, unit=exp.Literal.string("DAY")
+        )
+
+    # DATE_ADD / DATEADD / TIMESTAMPADD(unit, value integer, expr)
+    # https://docs.databricks.com/en/sql/language-manual/functions/date_add3.html
+    return exp.TimestampAdd(this=seq_get(args, 2), expression=expression, unit=seq_get(args, 0))
+
+
 def _normalize_partition(e: exp.Expression) -> exp.Expression:
     """Normalize the expressions in PARTITION BY (<expression>, <expression>, ...)"""
     if isinstance(e, str):
@@ -50,6 +65,30 @@ def _normalize_partition(e: exp.Expression) -> exp.Expression:
     return e
 
 
+def _dateadd_sql(self: Spark.Generator, expression: exp.TsOrDsAdd | exp.TimestampAdd) -> str:
+    if not expression.unit or (
+        isinstance(expression, exp.TsOrDsAdd) and expression.text("unit").upper() == "DAY"
+    ):
+        # Coming from Hive/Spark2 DATE_ADD or roundtripping the 2-arg version of Spark3/DB
+        return self.func("DATE_ADD", expression.this, expression.expression)
+
+    this = self.func(
+        "DATE_ADD",
+        unit_to_var(expression),
+        expression.expression,
+        expression.this,
+    )
+
+    if isinstance(expression, exp.TsOrDsAdd):
+        # The 3 arg version of DATE_ADD produces a timestamp in Spark3/DB but possibly not
+        # in other dialects
+        return_type = expression.return_type
+        if not return_type.is_type(exp.DataType.Type.TIMESTAMP, exp.DataType.Type.DATETIME):
+            this = f"CAST({this} AS {return_type})"
+
+    return this
+
+
 class Spark(Spark2):
     class Tokenizer(Spark2.Tokenizer):
         RAW_STRINGS = [
@@ -62,6 +101,9 @@ class Spark(Spark2):
         FUNCTIONS = {
             **Spark2.Parser.FUNCTIONS,
             "ANY_VALUE": _build_with_ignore_nulls(exp.AnyValue),
+            "DATE_ADD": _build_dateadd,
+            "DATEADD": _build_dateadd,
+            "TIMESTAMPADD": _build_dateadd,
             "DATEDIFF": _build_datediff,
             "TIMESTAMP_LTZ": _build_as_cast("TIMESTAMP_LTZ"),
             "TIMESTAMP_NTZ": _build_as_cast("TIMESTAMP_NTZ"),
@@ -111,9 +153,8 @@ class Spark(Spark2):
             exp.PartitionedByProperty: lambda self,
             e: f"PARTITIONED BY {self.wrap(self.expressions(sqls=[_normalize_partition(e) for e in e.this.expressions], skip_first=True))}",
             exp.StartsWith: rename_func("STARTSWITH"),
-            exp.TimestampAdd: lambda self, e: self.func(
-                "DATEADD", unit_to_var(e), e.expression, e.this
-            ),
+            exp.TsOrDsAdd: _dateadd_sql,
+            exp.TimestampAdd: _dateadd_sql,
             exp.TryCast: lambda self, e: (
                 self.trycast_sql(e) if e.args.get("safe") else self.cast_sql(e)
             ),


Test patch:
diff --git a/tests/dialects/test_bigquery.py b/tests/dialects/test_bigquery.py
index bfaf0091..888071de 100644
--- a/tests/dialects/test_bigquery.py
+++ b/tests/dialects/test_bigquery.py
@@ -619,9 +619,9 @@ LANGUAGE js AS
             'SELECT TIMESTAMP_ADD(TIMESTAMP "2008-12-25 15:30:00+00", INTERVAL 10 MINUTE)',
             write={
                 "bigquery": "SELECT TIMESTAMP_ADD(CAST('2008-12-25 15:30:00+00' AS TIMESTAMP), INTERVAL 10 MINUTE)",
-                "databricks": "SELECT DATEADD(MINUTE, 10, CAST('2008-12-25 15:30:00+00' AS TIMESTAMP))",
+                "databricks": "SELECT DATE_ADD(MINUTE, 10, CAST('2008-12-25 15:30:00+00' AS TIMESTAMP))",
                 "mysql": "SELECT DATE_ADD(TIMESTAMP('2008-12-25 15:30:00+00'), INTERVAL 10 MINUTE)",
-                "spark": "SELECT DATEADD(MINUTE, 10, CAST('2008-12-25 15:30:00+00' AS TIMESTAMP))",
+                "spark": "SELECT DATE_ADD(MINUTE, 10, CAST('2008-12-25 15:30:00+00' AS TIMESTAMP))",
             },
         )
         self.validate_all(
diff --git a/tests/dialects/test_postgres.py b/tests/dialects/test_postgres.py
index 74753beb..50ba6053 100644
--- a/tests/dialects/test_postgres.py
+++ b/tests/dialects/test_postgres.py
@@ -724,6 +724,28 @@ class TestPostgres(Validator):
         self.validate_identity("cast(a as FLOAT8)", "CAST(a AS DOUBLE PRECISION)")
         self.validate_identity("cast(a as FLOAT4)", "CAST(a AS REAL)")
 
+        self.validate_all(
+            "1 / DIV(4, 2)",
+            read={
+                "postgres": "1 / DIV(4, 2)",
+            },
+            write={
+                "sqlite": "1 / CAST(CAST(CAST(4 AS REAL) / 2 AS INTEGER) AS REAL)",
+                "duckdb": "1 / CAST(4 // 2 AS DECIMAL)",
+                "bigquery": "1 / CAST(DIV(4, 2) AS NUMERIC)",
+            },
+        )
+        self.validate_all(
+            "CAST(DIV(4, 2) AS DECIMAL(5, 3))",
+            read={
+                "duckdb": "CAST(4 // 2 AS DECIMAL(5, 3))",
+            },
+            write={
+                "duckdb": "CAST(CAST(4 // 2 AS DECIMAL) AS DECIMAL(5, 3))",
+                "postgres": "CAST(DIV(4, 2) AS DECIMAL(5, 3))",
+            },
+        )
+
     def test_ddl(self):
         # Checks that user-defined types are parsed into DataType instead of Identifier
         self.parse_one("CREATE TABLE t (a udt)").this.expressions[0].args["kind"].assert_is(
diff --git a/tests/dialects/test_redshift.py b/tests/dialects/test_redshift.py
index 844fe461..69793c73 100644
--- a/tests/dialects/test_redshift.py
+++ b/tests/dialects/test_redshift.py
@@ -281,6 +281,9 @@ class TestRedshift(Validator):
                 "redshift": "SELECT DATEADD(MONTH, 18, '2008-02-28')",
                 "snowflake": "SELECT DATEADD(MONTH, 18, CAST('2008-02-28' AS TIMESTAMP))",
                 "tsql": "SELECT DATEADD(MONTH, 18, CAST('2008-02-28' AS DATETIME2))",
+                "spark": "SELECT DATE_ADD(MONTH, 18, '2008-02-28')",
+                "spark2": "SELECT ADD_MONTHS('2008-02-28', 18)",
+                "databricks": "SELECT DATE_ADD(MONTH, 18, '2008-02-28')",
             },
         )
         self.validate_all(
@@ -585,3 +588,9 @@ FROM (
         self.assertEqual(
             ast.sql("redshift"), "SELECT * FROM x AS a, a.b AS c, c.d.e AS f, f.g.h.i.j.k AS l"
         )
+
+    def test_join_markers(self):
+        self.validate_identity(
+            "select a.foo, b.bar, a.baz from a, b where a.baz = b.baz (+)",
+            "SELECT a.foo, b.bar, a.baz FROM a, b WHERE a.baz = b.baz (+)",
+        )
diff --git a/tests/dialects/test_spark.py b/tests/dialects/test_spark.py
index ecc152f2..bff91bf8 100644
--- a/tests/dialects/test_spark.py
+++ b/tests/dialects/test_spark.py
@@ -563,6 +563,7 @@ TBLPROPERTIES (
             "SELECT DATE_ADD(my_date_column, 1)",
             write={
                 "spark": "SELECT DATE_ADD(my_date_column, 1)",
+                "spark2": "SELECT DATE_ADD(my_date_column, 1)",
                 "bigquery": "SELECT DATE_ADD(CAST(CAST(my_date_column AS DATETIME) AS DATE), INTERVAL 1 DAY)",
             },
         )
@@ -675,6 +676,16 @@ TBLPROPERTIES (
                 "spark": "SELECT ARRAY_SORT(x)",
             },
         )
+        self.validate_all(
+            "SELECT DATE_ADD(MONTH, 20, col)",
+            read={
+                "spark": "SELECT TIMESTAMPADD(MONTH, 20, col)",
+            },
+            write={
+                "spark": "SELECT DATE_ADD(MONTH, 20, col)",
+                "databricks": "SELECT DATE_ADD(MONTH, 20, col)",
+            },
+        )
 
     def test_bool_or(self):
         self.validate_all(