Repository: tobymao/sqlglot
Pull request: fix(spark)!: distinguish STORED AS from USING
Fixes #5317

This PR distinguishes `STORED AS` from `USING`. 

Spark and Databricks supports hive format when creating a table. This means that if we only use the previous approach, generating `STORED AS` with `USING` the generated SQL can be wrong. 

Example: 
```CREATE TABLE student (id INT, name STRING, age INT) STORED AS TEXTFILE```

**DOCS**
[Hive Format (STORED AS)](https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-create-table-hiveformat)

Problem statement:
Spark incorrectly roundtrips `STORED AS` -> `USING`
**Before you file an issue**
- Make sure you specify the "read" dialect eg. `parse_one(sql, read="spark")`
- Make sure you specify the "write" dialect eg. `ast.sql(dialect="duckdb")`
- Check if the issue still exists on main
we are going from spark to spark in this case

**Fully reproducible code snippet**
Please include a fully reproducible code snippet or the input sql, dialect, and expected output.
```
>>> sql = "CREATE TABLE student (id INT, name STRING, age INT) STORED AS ORC;"
>>> print(sqlglot.parse_one(sql, dialect="spark").sql(pretty=True, dialect="spark"))
CREATE TABLE student (
  id INT,
  name STRING,
  age INT
)
USING ORC
```
**Official Documentation**
Please include links to official SQL documentation related to your issue.
I would just expect the round trip to keep the SQL as is, since now it is invalid

Hints from issue discussion:
Looks like the both get parsed as the same AST but they probably shouldnt?
```
>>> sqlglot.parse_one('CREATE TABLE student (id INT) STORED AS ORC')
Create(
  this=Schema(
    this=Table(
      this=Identifier(this=student, quoted=False)),
    expressions=[
      ColumnDef(
        this=Identifier(this=id, quoted=False),
        kind=DataType(this=Type.INT, nested=False))]),
  kind=TABLE,
  properties=Properties(
    expressions=[
      FileFormatProperty(
        this=Var(this=ORC))]))
>>> sqlglot.parse_one('CREATE TABLE student (id INT) USING ORC')
Create(
  this=Schema(
    this=Table(
      this=Identifier(this=student, quoted=False)),
    expressions=[
      ColumnDef(
        this=Identifier(this=id, quoted=False),
        kind=DataType(this=Type.INT, nested=False))]),
  kind=TABLE,
  properties=Properties(
    expressions=[
      FileFormatProperty(
        this=Var(this=ORC))]))
```

Commit message: add input output format

Patch to review:
diff --git a/sqlglot/dialects/spark2.py b/sqlglot/dialects/spark2.py
index afb1a1c537..b1a771e025 100644
--- a/sqlglot/dialects/spark2.py
+++ b/sqlglot/dialects/spark2.py
@@ -284,7 +284,10 @@ class Generator(Hive.Generator):
             # (DAY_OF_WEEK(datetime) % 7) + 1 is equivalent to DAYOFWEEK_ISO(datetime)
             exp.DayOfWeekIso: lambda self, e: f"(({self.func('DAYOFWEEK', e.this)} % 7) + 1)",
             exp.DayOfYear: rename_func("DAYOFYEAR"),
-            exp.FileFormatProperty: lambda self, e: f"USING {e.name.upper()}",
+            exp.FileFormatProperty: lambda self,
+            e: f"STORED AS {self.sql(e, 'this') if isinstance(e.this, exp.InputOutputFormat) else e.name.upper()}"
+            if e.args.get("hive_format")
+            else f"USING {e.name.upper()}",
             exp.From: transforms.preprocess([_unalias_pivot]),
             exp.FromTimeZone: lambda self, e: self.func(
                 "TO_UTC_TIMESTAMP", e.this, e.args.get("zone")
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index 021d43d0a8..a878274447 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -2856,8 +2856,9 @@ class FallbackProperty(Property):
     arg_types = {"no": True, "protection": False}
 
 
+# https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-create-table-hiveformat
 class FileFormatProperty(Property):
-    arg_types = {"this": False, "expressions": False}
+    arg_types = {"this": False, "expressions": False, "hive_format": False}
 
 
 class CredentialsProperty(Property):
diff --git a/sqlglot/parser.py b/sqlglot/parser.py
index 775ff5a1a3..0538d2fbf7 100644
--- a/sqlglot/parser.py
+++ b/sqlglot/parser.py
@@ -2237,6 +2237,7 @@ def _parse_stored(self) -> t.Union[exp.FileFormatProperty, exp.StorageHandlerPro
                 if input_format or output_format
                 else self._parse_var_or_string() or self._parse_number() or self._parse_id_var()
             ),
+            hive_format=True,
         )
 
     def _parse_unquoted_field(self) -> t.Optional[exp.Expression]:
diff --git a/tests/dialects/test_hive.py b/tests/dialects/test_hive.py
index de084e363b..532145c62f 100644
--- a/tests/dialects/test_hive.py
+++ b/tests/dialects/test_hive.py
@@ -163,7 +163,16 @@ def test_ddl(self):
                 "duckdb": "CREATE TABLE test AS SELECT 1",
                 "presto": "CREATE TABLE test WITH (format='parquet', x='1', Z='2') AS SELECT 1",
                 "hive": "CREATE TABLE test STORED AS PARQUET TBLPROPERTIES ('x'='1', 'Z'='2') AS SELECT 1",
-                "spark": "CREATE TABLE test USING PARQUET TBLPROPERTIES ('x'='1', 'Z'='2') AS SELECT 1",
+                "spark": "CREATE TABLE test STORED AS PARQUET TBLPROPERTIES ('x'='1', 'Z'='2') AS SELECT 1",
+            },
+        )
+
+        self.validate_all(
+            "CREATE TABLE test STORED AS INPUTFORMAT 'foo1' OUTPUTFORMAT 'foo2'",
+            write={
+                "hive": "CREATE TABLE test STORED AS INPUTFORMAT 'foo1' OUTPUTFORMAT 'foo2'",
+                "spark": "CREATE TABLE test STORED AS INPUTFORMAT 'foo1' OUTPUTFORMAT 'foo2'",
+                "databricks": "CREATE TABLE test STORED AS INPUTFORMAT 'foo1' OUTPUTFORMAT 'foo2'",
             },
         )
 
diff --git a/tests/dialects/test_presto.py b/tests/dialects/test_presto.py
index 9e87ac0cb5..f113a1dca3 100644
--- a/tests/dialects/test_presto.py
+++ b/tests/dialects/test_presto.py
@@ -511,7 +511,7 @@ def test_ddl(self):
                 "duckdb": "CREATE TABLE test AS SELECT 1",
                 "presto": "CREATE TABLE test WITH (format='PARQUET') AS SELECT 1",
                 "hive": "CREATE TABLE test STORED AS PARQUET AS SELECT 1",
-                "spark": "CREATE TABLE test USING PARQUET AS SELECT 1",
+                "spark": "CREATE TABLE test STORED AS PARQUET AS SELECT 1",
             },
         )
         self.validate_all(
diff --git a/tests/dialects/test_spark.py b/tests/dialects/test_spark.py
index 4ad03fdf55..cfa7159e7c 100644
--- a/tests/dialects/test_spark.py
+++ b/tests/dialects/test_spark.py
@@ -69,11 +69,11 @@ def test_ddl(self):
                 "trino": "CREATE TABLE test WITH (format='PARQUET') AS SELECT 1",
                 "athena": "CREATE TABLE test WITH (format='PARQUET') AS SELECT 1",  # note: lowercase format property is important for Athena
                 "hive": "CREATE TABLE test STORED AS PARQUET AS SELECT 1",
-                "spark": "CREATE TABLE test USING PARQUET AS SELECT 1",
+                "spark": "CREATE TABLE test STORED AS PARQUET AS SELECT 1",
             },
         )
         self.validate_all(
-            """CREATE TABLE blah (col_a INT) COMMENT "Test comment: blah" PARTITIONED BY (date STRING) STORED AS ICEBERG TBLPROPERTIES('x' = '1')""",
+            """CREATE TABLE blah (col_a INT) COMMENT "Test comment: blah" PARTITIONED BY (date STRING) USING ICEBERG TBLPROPERTIES('x' = '1')""",
             write={
                 "duckdb": """CREATE TABLE blah (
   col_a INT


Code review comments:
diff_hunk: @@ -284,7 +284,10 @@ class Generator(Hive.Generator):
             # (DAY_OF_WEEK(datetime) % 7) + 1 is equivalent to DAYOFWEEK_ISO(datetime)
             exp.DayOfWeekIso: lambda self, e: f"(({self.func('DAYOFWEEK', e.this)} % 7) + 1)",
             exp.DayOfYear: rename_func("DAYOFYEAR"),
-            exp.FileFormatProperty: lambda self, e: f"USING {e.name.upper()}",
+            exp.FileFormatProperty: lambda self,
+            e: f"STORED AS {self.sql(e, 'this') if isinstance(e.this, exp.InputOutputFormat) else e.name.upper()}"
+            if e.args.get("hive_format")
+            else f"USING {e.name.upper()}",
path: sqlglot/dialects/spark2.py
text: @user1:
Can we clean this up a bit? Putting this logic in a single lambda is unreadable. Consider defining a `fileformatproperty_sql` for Hive so that you don't have to override the `TRANSFORMS` entry here, and simply override the method instead.

@author:
yep, will change it, thanks.