Merging upstream version 9.0.1.

Signed-off-by: Daniel Baumann <daniel@debian.org>

tests/dataframe/integration/dataframe_validator.py

@@ -0,0 +1,149 @@
+import typing as t
+import unittest
+import warnings
+
+import sqlglot
+from tests.helpers import SKIP_INTEGRATION
+
+if t.TYPE_CHECKING:
+    from pyspark.sql import DataFrame as SparkDataFrame
+
+
+@unittest.skipIf(SKIP_INTEGRATION, "Skipping Integration Tests since `SKIP_INTEGRATION` is set")
+class DataFrameValidator(unittest.TestCase):
+    spark = None
+    sqlglot = None
+    df_employee = None
+    df_store = None
+    df_district = None
+    spark_employee_schema = None
+    sqlglot_employee_schema = None
+    spark_store_schema = None
+    sqlglot_store_schema = None
+    spark_district_schema = None
+    sqlglot_district_schema = None
+
+    @classmethod
+    def setUpClass(cls):
+        from pyspark import SparkConf
+        from pyspark.sql import SparkSession, types
+
+        from sqlglot.dataframe.sql import types as sqlglotSparkTypes
+        from sqlglot.dataframe.sql.session import SparkSession as SqlglotSparkSession
+
+        # This is for test `test_branching_root_dataframes`
+        config = SparkConf().setAll([("spark.sql.analyzer.failAmbiguousSelfJoin", "false")])
+        cls.spark = SparkSession.builder.master("local[*]").appName("Unit-tests").config(conf=config).getOrCreate()
+        cls.spark.sparkContext.setLogLevel("ERROR")
+        cls.sqlglot = SqlglotSparkSession()
+        cls.spark_employee_schema = types.StructType(
+            [
+                types.StructField("employee_id", types.IntegerType(), False),
+                types.StructField("fname", types.StringType(), False),
+                types.StructField("lname", types.StringType(), False),
+                types.StructField("age", types.IntegerType(), False),
+                types.StructField("store_id", types.IntegerType(), False),
+            ]
+        )
+        cls.sqlglot_employee_schema = sqlglotSparkTypes.StructType(
+            [
+                sqlglotSparkTypes.StructField("employee_id", sqlglotSparkTypes.IntegerType(), False),
+                sqlglotSparkTypes.StructField("fname", sqlglotSparkTypes.StringType(), False),
+                sqlglotSparkTypes.StructField("lname", sqlglotSparkTypes.StringType(), False),
+                sqlglotSparkTypes.StructField("age", sqlglotSparkTypes.IntegerType(), False),
+                sqlglotSparkTypes.StructField("store_id", sqlglotSparkTypes.IntegerType(), False),
+            ]
+        )
+        employee_data = [
+            (1, "Jack", "Shephard", 37, 1),
+            (2, "John", "Locke", 65, 1),
+            (3, "Kate", "Austen", 37, 2),
+            (4, "Claire", "Littleton", 27, 2),
+            (5, "Hugo", "Reyes", 29, 100),
+        ]
+        cls.df_employee = cls.spark.createDataFrame(data=employee_data, schema=cls.spark_employee_schema)
+        cls.dfs_employee = cls.sqlglot.createDataFrame(data=employee_data, schema=cls.sqlglot_employee_schema)
+        cls.df_employee.createOrReplaceTempView("employee")
+
+        cls.spark_store_schema = types.StructType(
+            [
+                types.StructField("store_id", types.IntegerType(), False),
+                types.StructField("store_name", types.StringType(), False),
+                types.StructField("district_id", types.IntegerType(), False),
+                types.StructField("num_sales", types.IntegerType(), False),
+            ]
+        )
+        cls.sqlglot_store_schema = sqlglotSparkTypes.StructType(
+            [
+                sqlglotSparkTypes.StructField("store_id", sqlglotSparkTypes.IntegerType(), False),
+                sqlglotSparkTypes.StructField("store_name", sqlglotSparkTypes.StringType(), False),
+                sqlglotSparkTypes.StructField("district_id", sqlglotSparkTypes.IntegerType(), False),
+                sqlglotSparkTypes.StructField("num_sales", sqlglotSparkTypes.IntegerType(), False),
+            ]
+        )
+        store_data = [
+            (1, "Hydra", 1, 37),
+            (2, "Arrow", 2, 2000),
+        ]
+        cls.df_store = cls.spark.createDataFrame(data=store_data, schema=cls.spark_store_schema)
+        cls.dfs_store = cls.sqlglot.createDataFrame(data=store_data, schema=cls.sqlglot_store_schema)
+        cls.df_store.createOrReplaceTempView("store")
+
+        cls.spark_district_schema = types.StructType(
+            [
+                types.StructField("district_id", types.IntegerType(), False),
+                types.StructField("district_name", types.StringType(), False),
+                types.StructField("manager_name", types.StringType(), False),
+            ]
+        )
+        cls.sqlglot_district_schema = sqlglotSparkTypes.StructType(
+            [
+                sqlglotSparkTypes.StructField("district_id", sqlglotSparkTypes.IntegerType(), False),
+                sqlglotSparkTypes.StructField("district_name", sqlglotSparkTypes.StringType(), False),
+                sqlglotSparkTypes.StructField("manager_name", sqlglotSparkTypes.StringType(), False),
+            ]
+        )
+        district_data = [
+            (1, "Temple", "Dogen"),
+            (2, "Lighthouse", "Jacob"),
+        ]
+        cls.df_district = cls.spark.createDataFrame(data=district_data, schema=cls.spark_district_schema)
+        cls.dfs_district = cls.sqlglot.createDataFrame(data=district_data, schema=cls.sqlglot_district_schema)
+        cls.df_district.createOrReplaceTempView("district")
+        sqlglot.schema.add_table("employee", cls.sqlglot_employee_schema)
+        sqlglot.schema.add_table("store", cls.sqlglot_store_schema)
+        sqlglot.schema.add_table("district", cls.sqlglot_district_schema)
+
+    def setUp(self) -> None:
+        warnings.filterwarnings("ignore", category=ResourceWarning)
+        self.df_spark_store = self.df_store.alias("df_store")  # type: ignore
+        self.df_spark_employee = self.df_employee.alias("df_employee")  # type: ignore
+        self.df_spark_district = self.df_district.alias("df_district")  # type: ignore
+        self.df_sqlglot_store = self.dfs_store.alias("store")  # type: ignore
+        self.df_sqlglot_employee = self.dfs_employee.alias("employee")  # type: ignore
+        self.df_sqlglot_district = self.dfs_district.alias("district")  # type: ignore
+
+    def compare_spark_with_sqlglot(
+        self, df_spark, df_sqlglot, no_empty=True, skip_schema_compare=False
+    ) -> t.Tuple["SparkDataFrame", "SparkDataFrame"]:
+        def compare_schemas(schema_1, schema_2):
+            for schema in [schema_1, schema_2]:
+                for struct_field in schema.fields:
+                    struct_field.metadata = {}
+            self.assertEqual(schema_1, schema_2)
+
+        for statement in df_sqlglot.sql():
+            actual_df_sqlglot = self.spark.sql(statement)  # type: ignore
+        df_sqlglot_results = actual_df_sqlglot.collect()
+        df_spark_results = df_spark.collect()
+        if not skip_schema_compare:
+            compare_schemas(df_spark.schema, actual_df_sqlglot.schema)
+        self.assertEqual(df_spark_results, df_sqlglot_results)
+        if no_empty:
+            self.assertNotEqual(len(df_spark_results), 0)
+            self.assertNotEqual(len(df_sqlglot_results), 0)
+        return df_spark, actual_df_sqlglot
+
+    @classmethod
+    def get_explain_plan(cls, df: "SparkDataFrame", mode: str = "extended") -> str:
+        return df._sc._jvm.PythonSQLUtils.explainString(df._jdf.queryExecution(), mode)  # type: ignore

tests/dataframe/integration/test_grouped_data.py

@@ -0,0 +1,71 @@
+from pyspark.sql import functions as F
+
+from sqlglot.dataframe.sql import functions as SF
+from tests.dataframe.integration.dataframe_validator import DataFrameValidator
+
+
+class TestDataframeFunc(DataFrameValidator):
+    def test_group_by(self):
+        df_employee = self.df_spark_employee.groupBy(self.df_spark_employee.age).agg(
+            F.min(self.df_spark_employee.employee_id)
+        )
+        dfs_employee = self.df_sqlglot_employee.groupBy(self.df_sqlglot_employee.age).agg(
+            SF.min(self.df_sqlglot_employee.employee_id)
+        )
+        self.compare_spark_with_sqlglot(df_employee, dfs_employee, skip_schema_compare=True)
+
+    def test_group_by_where_non_aggregate(self):
+        df_employee = (
+            self.df_spark_employee.groupBy(self.df_spark_employee.age)
+            .agg(F.min(self.df_spark_employee.employee_id).alias("min_employee_id"))
+            .where(F.col("age") > F.lit(50))
+        )
+        dfs_employee = (
+            self.df_sqlglot_employee.groupBy(self.df_sqlglot_employee.age)
+            .agg(SF.min(self.df_sqlglot_employee.employee_id).alias("min_employee_id"))
+            .where(SF.col("age") > SF.lit(50))
+        )
+        self.compare_spark_with_sqlglot(df_employee, dfs_employee)
+
+    def test_group_by_where_aggregate_like_having(self):
+        df_employee = (
+            self.df_spark_employee.groupBy(self.df_spark_employee.age)
+            .agg(F.min(self.df_spark_employee.employee_id).alias("min_employee_id"))
+            .where(F.col("min_employee_id") > F.lit(1))
+        )
+        dfs_employee = (
+            self.df_sqlglot_employee.groupBy(self.df_sqlglot_employee.age)
+            .agg(SF.min(self.df_sqlglot_employee.employee_id).alias("min_employee_id"))
+            .where(SF.col("min_employee_id") > SF.lit(1))
+        )
+        self.compare_spark_with_sqlglot(df_employee, dfs_employee)
+
+    def test_count(self):
+        df = self.df_spark_employee.groupBy(self.df_spark_employee.age).count()
+        dfs = self.df_sqlglot_employee.groupBy(self.df_sqlglot_employee.age).count()
+        self.compare_spark_with_sqlglot(df, dfs)
+
+    def test_mean(self):
+        df = self.df_spark_employee.groupBy().mean("age", "store_id")
+        dfs = self.df_sqlglot_employee.groupBy().mean("age", "store_id")
+        self.compare_spark_with_sqlglot(df, dfs)
+
+    def test_avg(self):
+        df = self.df_spark_employee.groupBy("age").avg("store_id")
+        dfs = self.df_sqlglot_employee.groupBy("age").avg("store_id")
+        self.compare_spark_with_sqlglot(df, dfs)
+
+    def test_max(self):
+        df = self.df_spark_employee.groupBy("age").max("store_id")
+        dfs = self.df_sqlglot_employee.groupBy("age").max("store_id")
+        self.compare_spark_with_sqlglot(df, dfs)
+
+    def test_min(self):
+        df = self.df_spark_employee.groupBy("age").min("store_id")
+        dfs = self.df_sqlglot_employee.groupBy("age").min("store_id")
+        self.compare_spark_with_sqlglot(df, dfs)
+
+    def test_sum(self):
+        df = self.df_spark_employee.groupBy("age").sum("store_id")
+        dfs = self.df_sqlglot_employee.groupBy("age").sum("store_id")
+        self.compare_spark_with_sqlglot(df, dfs)

tests/dataframe/integration/test_session.py

@@ -0,0 +1,28 @@
+from pyspark.sql import functions as F
+
+from sqlglot.dataframe.sql import functions as SF
+from tests.dataframe.integration.dataframe_validator import DataFrameValidator
+
+
+class TestSessionFunc(DataFrameValidator):
+    def test_sql_simple_select(self):
+        query = "SELECT fname, lname FROM employee"
+        df = self.spark.sql(query)
+        dfs = self.sqlglot.sql(query)
+        self.compare_spark_with_sqlglot(df, dfs)
+
+    def test_sql_with_join(self):
+        query = """
+        SELECT
+            e.employee_id
+            , s.store_id    
+        FROM
+            employee e
+            INNER JOIN
+            store s
+            ON
+                e.store_id = s.store_id
+        """
+        df = self.spark.sql(query).groupBy(F.col("store_id")).agg(F.countDistinct(F.col("employee_id")))
+        dfs = self.sqlglot.sql(query).groupBy(SF.col("store_id")).agg(SF.countDistinct(SF.col("employee_id")))
+        self.compare_spark_with_sqlglot(df, dfs, skip_schema_compare=True)

tests/dataframe/unit/dataframe_sql_validator.py

@@ -0,0 +1,35 @@
+import typing as t
+import unittest
+
+from sqlglot.dataframe.sql import types
+from sqlglot.dataframe.sql.dataframe import DataFrame
+from sqlglot.dataframe.sql.session import SparkSession
+
+
+class DataFrameSQLValidator(unittest.TestCase):
+    def setUp(self) -> None:
+        self.spark = SparkSession()
+        self.employee_schema = types.StructType(
+            [
+                types.StructField("employee_id", types.IntegerType(), False),
+                types.StructField("fname", types.StringType(), False),
+                types.StructField("lname", types.StringType(), False),
+                types.StructField("age", types.IntegerType(), False),
+                types.StructField("store_id", types.IntegerType(), False),
+            ]
+        )
+        employee_data = [
+            (1, "Jack", "Shephard", 37, 1),
+            (2, "John", "Locke", 65, 1),
+            (3, "Kate", "Austen", 37, 2),
+            (4, "Claire", "Littleton", 27, 2),
+            (5, "Hugo", "Reyes", 29, 100),
+        ]
+        self.df_employee = self.spark.createDataFrame(data=employee_data, schema=self.employee_schema)
+
+    def compare_sql(self, df: DataFrame, expected_statements: t.Union[str, t.List[str]], pretty=False):
+        actual_sqls = df.sql(pretty=pretty)
+        expected_statements = [expected_statements] if isinstance(expected_statements, str) else expected_statements
+        self.assertEqual(len(expected_statements), len(actual_sqls))
+        for expected, actual in zip(expected_statements, actual_sqls):
+            self.assertEqual(expected, actual)

tests/dataframe/unit/test_column.py

@@ -0,0 +1,167 @@
+import datetime
+import unittest
+
+from sqlglot.dataframe.sql import functions as F
+from sqlglot.dataframe.sql.window import Window
+
+
+class TestDataframeColumn(unittest.TestCase):
+    def test_eq(self):
+        self.assertEqual("cola = 1", (F.col("cola") == 1).sql())
+
+    def test_neq(self):
+        self.assertEqual("cola <> 1", (F.col("cola") != 1).sql())
+
+    def test_gt(self):
+        self.assertEqual("cola > 1", (F.col("cola") > 1).sql())
+
+    def test_lt(self):
+        self.assertEqual("cola < 1", (F.col("cola") < 1).sql())
+
+    def test_le(self):
+        self.assertEqual("cola <= 1", (F.col("cola") <= 1).sql())
+
+    def test_ge(self):
+        self.assertEqual("cola >= 1", (F.col("cola") >= 1).sql())
+
+    def test_and(self):
+        self.assertEqual(
+            "cola = colb AND colc = cold", ((F.col("cola") == F.col("colb")) & (F.col("colc") == F.col("cold"))).sql()
+        )
+
+    def test_or(self):
+        self.assertEqual(
+            "cola = colb OR colc = cold", ((F.col("cola") == F.col("colb")) | (F.col("colc") == F.col("cold"))).sql()
+        )
+
+    def test_mod(self):
+        self.assertEqual("cola % 2", (F.col("cola") % 2).sql())
+
+    def test_add(self):
+        self.assertEqual("cola + 1", (F.col("cola") + 1).sql())
+
+    def test_sub(self):
+        self.assertEqual("cola - 1", (F.col("cola") - 1).sql())
+
+    def test_mul(self):
+        self.assertEqual("cola * 2", (F.col("cola") * 2).sql())
+
+    def test_div(self):
+        self.assertEqual("cola / 2", (F.col("cola") / 2).sql())
+
+    def test_radd(self):
+        self.assertEqual("1 + cola", (1 + F.col("cola")).sql())
+
+    def test_rsub(self):
+        self.assertEqual("1 - cola", (1 - F.col("cola")).sql())
+
+    def test_rmul(self):
+        self.assertEqual("1 * cola", (1 * F.col("cola")).sql())
+
+    def test_rdiv(self):
+        self.assertEqual("1 / cola", (1 / F.col("cola")).sql())
+
+    def test_pow(self):
+        self.assertEqual("POWER(cola, 2)", (F.col("cola") ** 2).sql())
+
+    def test_rpow(self):
+        self.assertEqual("POWER(2, cola)", (2 ** F.col("cola")).sql())
+
+    def test_invert(self):
+        self.assertEqual("NOT cola", (~F.col("cola")).sql())
+
+    def test_startswith(self):
+        self.assertEqual("STARTSWITH(cola, 'test')", F.col("cola").startswith("test").sql())
+
+    def test_endswith(self):
+        self.assertEqual("ENDSWITH(cola, 'test')", F.col("cola").endswith("test").sql())
+
+    def test_rlike(self):
+        self.assertEqual("cola RLIKE 'foo'", F.col("cola").rlike("foo").sql())
+
+    def test_like(self):
+        self.assertEqual("cola LIKE 'foo%'", F.col("cola").like("foo%").sql())
+
+    def test_ilike(self):
+        self.assertEqual("cola ILIKE 'foo%'", F.col("cola").ilike("foo%").sql())
+
+    def test_substring(self):
+        self.assertEqual("SUBSTRING(cola, 2, 3)", F.col("cola").substr(2, 3).sql())
+
+    def test_isin(self):
+        self.assertEqual("cola IN (1, 2, 3)", F.col("cola").isin([1, 2, 3]).sql())
+        self.assertEqual("cola IN (1, 2, 3)", F.col("cola").isin(1, 2, 3).sql())
+
+    def test_asc(self):
+        self.assertEqual("cola", F.col("cola").asc().sql())
+
+    def test_desc(self):
+        self.assertEqual("cola DESC", F.col("cola").desc().sql())
+
+    def test_asc_nulls_first(self):
+        self.assertEqual("cola", F.col("cola").asc_nulls_first().sql())
+
+    def test_asc_nulls_last(self):
+        self.assertEqual("cola NULLS LAST", F.col("cola").asc_nulls_last().sql())
+
+    def test_desc_nulls_first(self):
+        self.assertEqual("cola DESC NULLS FIRST", F.col("cola").desc_nulls_first().sql())
+
+    def test_desc_nulls_last(self):
+        self.assertEqual("cola DESC", F.col("cola").desc_nulls_last().sql())
+
+    def test_when_otherwise(self):
+        self.assertEqual("CASE WHEN cola = 1 THEN 2 END", F.when(F.col("cola") == 1, 2).sql())
+        self.assertEqual("CASE WHEN cola = 1 THEN 2 END", F.col("cola").when(F.col("cola") == 1, 2).sql())
+        self.assertEqual(
+            "CASE WHEN cola = 1 THEN 2 WHEN colb = 2 THEN 3 END",
+            (F.when(F.col("cola") == 1, 2).when(F.col("colb") == 2, 3)).sql(),
+        )
+        self.assertEqual(
+            "CASE WHEN cola = 1 THEN 2 WHEN colb = 2 THEN 3 END",
+            F.col("cola").when(F.col("cola") == 1, 2).when(F.col("colb") == 2, 3).sql(),
+        )
+        self.assertEqual(
+            "CASE WHEN cola = 1 THEN 2 WHEN colb = 2 THEN 3 ELSE 4 END",
+            F.when(F.col("cola") == 1, 2).when(F.col("colb") == 2, 3).otherwise(4).sql(),
+        )
+
+    def test_is_null(self):
+        self.assertEqual("cola IS NULL", F.col("cola").isNull().sql())
+
+    def test_is_not_null(self):
+        self.assertEqual("NOT cola IS NULL", F.col("cola").isNotNull().sql())
+
+    def test_cast(self):
+        self.assertEqual("CAST(cola AS INT)", F.col("cola").cast("INT").sql())
+
+    def test_alias(self):
+        self.assertEqual("cola AS new_name", F.col("cola").alias("new_name").sql())
+
+    def test_between(self):
+        self.assertEqual("cola BETWEEN 1 AND 3", F.col("cola").between(1, 3).sql())
+        self.assertEqual("cola BETWEEN 10.1 AND 12.1", F.col("cola").between(10.1, 12.1).sql())
+        self.assertEqual(
+            "cola BETWEEN TO_DATE('2022-01-01') AND TO_DATE('2022-03-01')",
+            F.col("cola").between(datetime.date(2022, 1, 1), datetime.date(2022, 3, 1)).sql(),
+        )
+        self.assertEqual(
+            "cola BETWEEN CAST('2022-01-01 01:01:01' AS TIMESTAMP) " "AND CAST('2022-03-01 01:01:01' AS TIMESTAMP)",
+            F.col("cola").between(datetime.datetime(2022, 1, 1, 1, 1, 1), datetime.datetime(2022, 3, 1, 1, 1, 1)).sql(),
+        )
+
+    def test_over(self):
+        over_rows = F.sum("cola").over(
+            Window.partitionBy("colb").orderBy("colc").rowsBetween(1, Window.unboundedFollowing)
+        )
+        self.assertEqual(
+            "SUM(cola) OVER (PARTITION BY colb ORDER BY colc ROWS BETWEEN 1 PRECEDING AND UNBOUNDED FOLLOWING)",
+            over_rows.sql(),
+        )
+        over_range = F.sum("cola").over(
+            Window.partitionBy("colb").orderBy("colc").rangeBetween(1, Window.unboundedFollowing)
+        )
+        self.assertEqual(
+            "SUM(cola) OVER (PARTITION BY colb ORDER BY colc RANGE BETWEEN 1 PRECEDING AND UNBOUNDED FOLLOWING)",
+            over_range.sql(),
+        )

tests/dataframe/unit/test_dataframe.py

@@ -0,0 +1,39 @@
+from sqlglot import expressions as exp
+from sqlglot.dataframe.sql.dataframe import DataFrame
+from tests.dataframe.unit.dataframe_sql_validator import DataFrameSQLValidator
+
+
+class TestDataframe(DataFrameSQLValidator):
+    def test_hash_select_expression(self):
+        expression = exp.select("cola").from_("table")
+        self.assertEqual("t17051", DataFrame._create_hash_from_expression(expression))
+
+    def test_columns(self):
+        self.assertEqual(["employee_id", "fname", "lname", "age", "store_id"], self.df_employee.columns)
+
+    def test_cache(self):
+        df = self.df_employee.select("fname").cache()
+        expected_statements = [
+            "DROP VIEW IF EXISTS t11623",
+            "CACHE LAZY TABLE t11623 OPTIONS('storageLevel' = 'MEMORY_AND_DISK') AS SELECT CAST(`a1`.`fname` AS string) AS `fname` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)",
+            "SELECT `t11623`.`fname` AS `fname` FROM `t11623` AS `t11623`",
+        ]
+        self.compare_sql(df, expected_statements)
+
+    def test_persist_default(self):
+        df = self.df_employee.select("fname").persist()
+        expected_statements = [
+            "DROP VIEW IF EXISTS t11623",
+            "CACHE LAZY TABLE t11623 OPTIONS('storageLevel' = 'MEMORY_AND_DISK_SER') AS SELECT CAST(`a1`.`fname` AS string) AS `fname` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)",
+            "SELECT `t11623`.`fname` AS `fname` FROM `t11623` AS `t11623`",
+        ]
+        self.compare_sql(df, expected_statements)
+
+    def test_persist_storagelevel(self):
+        df = self.df_employee.select("fname").persist("DISK_ONLY_2")
+        expected_statements = [
+            "DROP VIEW IF EXISTS t11623",
+            "CACHE LAZY TABLE t11623 OPTIONS('storageLevel' = 'DISK_ONLY_2') AS SELECT CAST(`a1`.`fname` AS string) AS `fname` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)",
+            "SELECT `t11623`.`fname` AS `fname` FROM `t11623` AS `t11623`",
+        ]
+        self.compare_sql(df, expected_statements)

tests/dataframe/unit/test_dataframe_writer.py

@@ -0,0 +1,86 @@
+from unittest import mock
+
+import sqlglot
+from sqlglot.schema import MappingSchema
+from tests.dataframe.unit.dataframe_sql_validator import DataFrameSQLValidator
+
+
+class TestDataFrameWriter(DataFrameSQLValidator):
+    def test_insertInto_full_path(self):
+        df = self.df_employee.write.insertInto("catalog.db.table_name")
+        expected = "INSERT INTO catalog.db.table_name SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_insertInto_db_table(self):
+        df = self.df_employee.write.insertInto("db.table_name")
+        expected = "INSERT INTO db.table_name SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_insertInto_table(self):
+        df = self.df_employee.write.insertInto("table_name")
+        expected = "INSERT INTO table_name SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_insertInto_overwrite(self):
+        df = self.df_employee.write.insertInto("table_name", overwrite=True)
+        expected = "INSERT OVERWRITE TABLE table_name SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    @mock.patch("sqlglot.schema", MappingSchema())
+    def test_insertInto_byName(self):
+        sqlglot.schema.add_table("table_name", {"employee_id": "INT"})
+        df = self.df_employee.write.byName.insertInto("table_name")
+        expected = "INSERT INTO table_name SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_insertInto_cache(self):
+        df = self.df_employee.cache().write.insertInto("table_name")
+        expected_statements = [
+            "DROP VIEW IF EXISTS t35612",
+            "CACHE LAZY TABLE t35612 OPTIONS('storageLevel' = 'MEMORY_AND_DISK') AS SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)",
+            "INSERT INTO table_name SELECT `t35612`.`employee_id` AS `employee_id`, `t35612`.`fname` AS `fname`, `t35612`.`lname` AS `lname`, `t35612`.`age` AS `age`, `t35612`.`store_id` AS `store_id` FROM `t35612` AS `t35612`",
+        ]
+        self.compare_sql(df, expected_statements)
+
+    def test_saveAsTable_format(self):
+        with self.assertRaises(NotImplementedError):
+            self.df_employee.write.saveAsTable("table_name", format="parquet").sql(pretty=False)[0]
+
+    def test_saveAsTable_append(self):
+        df = self.df_employee.write.saveAsTable("table_name", mode="append")
+        expected = "INSERT INTO table_name SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_saveAsTable_overwrite(self):
+        df = self.df_employee.write.saveAsTable("table_name", mode="overwrite")
+        expected = "CREATE OR REPLACE TABLE table_name AS SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_saveAsTable_error(self):
+        df = self.df_employee.write.saveAsTable("table_name", mode="error")
+        expected = "CREATE TABLE table_name AS SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_saveAsTable_ignore(self):
+        df = self.df_employee.write.saveAsTable("table_name", mode="ignore")
+        expected = "CREATE TABLE IF NOT EXISTS table_name AS SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_mode_standalone(self):
+        df = self.df_employee.write.mode("ignore").saveAsTable("table_name")
+        expected = "CREATE TABLE IF NOT EXISTS table_name AS SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_mode_override(self):
+        df = self.df_employee.write.mode("ignore").saveAsTable("table_name", mode="overwrite")
+        expected = "CREATE OR REPLACE TABLE table_name AS SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)"
+        self.compare_sql(df, expected)
+
+    def test_saveAsTable_cache(self):
+        df = self.df_employee.cache().write.saveAsTable("table_name")
+        expected_statements = [
+            "DROP VIEW IF EXISTS t35612",
+            "CACHE LAZY TABLE t35612 OPTIONS('storageLevel' = 'MEMORY_AND_DISK') AS SELECT CAST(`a1`.`employee_id` AS int) AS `employee_id`, CAST(`a1`.`fname` AS string) AS `fname`, CAST(`a1`.`lname` AS string) AS `lname`, CAST(`a1`.`age` AS int) AS `age`, CAST(`a1`.`store_id` AS int) AS `store_id` FROM (VALUES (1, 'Jack', 'Shephard', 37, 1), (2, 'John', 'Locke', 65, 1), (3, 'Kate', 'Austen', 37, 2), (4, 'Claire', 'Littleton', 27, 2), (5, 'Hugo', 'Reyes', 29, 100)) AS `a1`(`employee_id`, `fname`, `lname`, `age`, `store_id`)",
+            "CREATE TABLE table_name AS SELECT `t35612`.`employee_id` AS `employee_id`, `t35612`.`fname` AS `fname`, `t35612`.`lname` AS `lname`, `t35612`.`age` AS `age`, `t35612`.`store_id` AS `store_id` FROM `t35612` AS `t35612`",
+        ]
+        self.compare_sql(df, expected_statements)

tests/dataframe/unit/test_session.py

@@ -0,0 +1,114 @@
+from unittest import mock
+
+import sqlglot
+from sqlglot.dataframe.sql import functions as F
+from sqlglot.dataframe.sql import types
+from sqlglot.dataframe.sql.session import SparkSession
+from sqlglot.schema import MappingSchema
+from tests.dataframe.unit.dataframe_sql_validator import DataFrameSQLValidator
+
+
+class TestDataframeSession(DataFrameSQLValidator):
+    def test_cdf_one_row(self):
+        df = self.spark.createDataFrame([[1, 2]], ["cola", "colb"])
+        expected = "SELECT `a2`.`cola` AS `cola`, `a2`.`colb` AS `colb` FROM (VALUES (1, 2)) AS `a2`(`cola`, `colb`)"
+        self.compare_sql(df, expected)
+
+    def test_cdf_multiple_rows(self):
+        df = self.spark.createDataFrame([[1, 2], [3, 4], [None, 6]], ["cola", "colb"])
+        expected = "SELECT `a2`.`cola` AS `cola`, `a2`.`colb` AS `colb` FROM (VALUES (1, 2), (3, 4), (NULL, 6)) AS `a2`(`cola`, `colb`)"
+        self.compare_sql(df, expected)
+
+    def test_cdf_no_schema(self):
+        df = self.spark.createDataFrame([[1, 2], [3, 4], [None, 6]])
+        expected = (
+            "SELECT `a2`.`_1` AS `_1`, `a2`.`_2` AS `_2` FROM (VALUES (1, 2), (3, 4), (NULL, 6)) AS `a2`(`_1`, `_2`)"
+        )
+        self.compare_sql(df, expected)
+
+    def test_cdf_row_mixed_primitives(self):
+        df = self.spark.createDataFrame([[1, 10.1, "test", False, None]])
+        expected = "SELECT `a2`.`_1` AS `_1`, `a2`.`_2` AS `_2`, `a2`.`_3` AS `_3`, `a2`.`_4` AS `_4`, `a2`.`_5` AS `_5` FROM (VALUES (1, 10.1, 'test', FALSE, NULL)) AS `a2`(`_1`, `_2`, `_3`, `_4`, `_5`)"
+        self.compare_sql(df, expected)
+
+    def test_cdf_dict_rows(self):
+        df = self.spark.createDataFrame([{"cola": 1, "colb": "test"}, {"cola": 2, "colb": "test2"}])
+        expected = "SELECT `a2`.`cola` AS `cola`, `a2`.`colb` AS `colb` FROM (VALUES (1, 'test'), (2, 'test2')) AS `a2`(`cola`, `colb`)"
+        self.compare_sql(df, expected)
+
+    def test_cdf_str_schema(self):
+        df = self.spark.createDataFrame([[1, "test"]], "cola: INT, colb: STRING")
+        expected = "SELECT CAST(`a2`.`cola` AS INT) AS `cola`, CAST(`a2`.`colb` AS STRING) AS `colb` FROM (VALUES (1, 'test')) AS `a2`(`cola`, `colb`)"
+        self.compare_sql(df, expected)
+
+    def test_typed_schema_basic(self):
+        schema = types.StructType(
+            [
+                types.StructField("cola", types.IntegerType()),
+                types.StructField("colb", types.StringType()),
+            ]
+        )
+        df = self.spark.createDataFrame([[1, "test"]], schema)
+        expected = "SELECT CAST(`a2`.`cola` AS int) AS `cola`, CAST(`a2`.`colb` AS string) AS `colb` FROM (VALUES (1, 'test')) AS `a2`(`cola`, `colb`)"
+        self.compare_sql(df, expected)
+
+    def test_typed_schema_nested(self):
+        schema = types.StructType(
+            [
+                types.StructField(
+                    "cola",
+                    types.StructType(
+                        [
+                            types.StructField("sub_cola", types.IntegerType()),
+                            types.StructField("sub_colb", types.StringType()),
+                        ]
+                    ),
+                )
+            ]
+        )
+        df = self.spark.createDataFrame([[{"sub_cola": 1, "sub_colb": "test"}]], schema)
+        expected = "SELECT CAST(`a2`.`cola` AS struct<sub_cola:int, sub_colb:string>) AS `cola` FROM (VALUES (STRUCT(1 AS `sub_cola`, 'test' AS `sub_colb`))) AS `a2`(`cola`)"
+        self.compare_sql(df, expected)
+
+    @mock.patch("sqlglot.schema", MappingSchema())
+    def test_sql_select_only(self):
+        # TODO: Do exact matches once CTE names are deterministic
+        query = "SELECT cola, colb FROM table"
+        sqlglot.schema.add_table("table", {"cola": "string", "colb": "string"})
+        df = self.spark.sql(query)
+        self.assertIn(
+            "SELECT `table`.`cola` AS `cola`, `table`.`colb` AS `colb` FROM `table` AS `table`", df.sql(pretty=False)
+        )
+
+    @mock.patch("sqlglot.schema", MappingSchema())
+    def test_sql_with_aggs(self):
+        # TODO: Do exact matches once CTE names are deterministic
+        query = "SELECT cola, colb FROM table"
+        sqlglot.schema.add_table("table", {"cola": "string", "colb": "string"})
+        df = self.spark.sql(query).groupBy(F.col("cola")).agg(F.sum("colb"))
+        result = df.sql(pretty=False, optimize=False)[0]
+        self.assertIn("SELECT cola, colb FROM table", result)
+        self.assertIn("SUM(colb)", result)
+        self.assertIn("GROUP BY cola", result)
+
+    @mock.patch("sqlglot.schema", MappingSchema())
+    def test_sql_create(self):
+        query = "CREATE TABLE new_table AS WITH t1 AS (SELECT cola, colb FROM table) SELECT cola, colb, FROM t1"
+        sqlglot.schema.add_table("table", {"cola": "string", "colb": "string"})
+        df = self.spark.sql(query)
+        expected = "CREATE TABLE new_table AS SELECT `table`.`cola` AS `cola`, `table`.`colb` AS `colb` FROM `table` AS `table`"
+        self.compare_sql(df, expected)
+
+    @mock.patch("sqlglot.schema", MappingSchema())
+    def test_sql_insert(self):
+        query = "WITH t1 AS (SELECT cola, colb FROM table) INSERT INTO new_table SELECT cola, colb FROM t1"
+        sqlglot.schema.add_table("table", {"cola": "string", "colb": "string"})
+        df = self.spark.sql(query)
+        expected = (
+            "INSERT INTO new_table SELECT `table`.`cola` AS `cola`, `table`.`colb` AS `colb` FROM `table` AS `table`"
+        )
+        self.compare_sql(df, expected)
+
+    def test_session_create_builder_patterns(self):
+        spark = SparkSession()
+        self.assertEqual(spark.builder.appName("abc").getOrCreate(), spark)

tests/dataframe/unit/test_types.py

@@ -0,0 +1,70 @@
+import unittest
+
+from sqlglot.dataframe.sql import types
+
+
+class TestDataframeTypes(unittest.TestCase):
+    def test_string(self):
+        self.assertEqual("string", types.StringType().simpleString())
+
+    def test_char(self):
+        self.assertEqual("char(100)", types.CharType(100).simpleString())
+
+    def test_varchar(self):
+        self.assertEqual("varchar(65)", types.VarcharType(65).simpleString())
+
+    def test_binary(self):
+        self.assertEqual("binary", types.BinaryType().simpleString())
+
+    def test_boolean(self):
+        self.assertEqual("boolean", types.BooleanType().simpleString())
+
+    def test_date(self):
+        self.assertEqual("date", types.DateType().simpleString())
+
+    def test_timestamp(self):
+        self.assertEqual("timestamp", types.TimestampType().simpleString())
+
+    def test_timestamp_ntz(self):
+        self.assertEqual("timestamp_ntz", types.TimestampNTZType().simpleString())
+
+    def test_decimal(self):
+        self.assertEqual("decimal(10, 3)", types.DecimalType(10, 3).simpleString())
+
+    def test_double(self):
+        self.assertEqual("double", types.DoubleType().simpleString())
+
+    def test_float(self):
+        self.assertEqual("float", types.FloatType().simpleString())
+
+    def test_byte(self):
+        self.assertEqual("tinyint", types.ByteType().simpleString())
+
+    def test_integer(self):
+        self.assertEqual("int", types.IntegerType().simpleString())
+
+    def test_long(self):
+        self.assertEqual("bigint", types.LongType().simpleString())
+
+    def test_short(self):
+        self.assertEqual("smallint", types.ShortType().simpleString())
+
+    def test_array(self):
+        self.assertEqual("array<int>", types.ArrayType(types.IntegerType()).simpleString())
+
+    def test_map(self):
+        self.assertEqual("map<int, string>", types.MapType(types.IntegerType(), types.StringType()).simpleString())
+
+    def test_struct_field(self):
+        self.assertEqual("cola:int", types.StructField("cola", types.IntegerType()).simpleString())
+
+    def test_struct_type(self):
+        self.assertEqual(
+            "struct<cola:int, colb:string>",
+            types.StructType(
+                [
+                    types.StructField("cola", types.IntegerType()),
+                    types.StructField("colb", types.StringType()),
+                ]
+            ).simpleString(),
+        )

tests/dataframe/unit/test_window.py

@@ -0,0 +1,60 @@
+import unittest
+
+from sqlglot.dataframe.sql import functions as F
+from sqlglot.dataframe.sql.window import Window, WindowSpec
+
+
+class TestDataframeWindow(unittest.TestCase):
+    def test_window_spec_partition_by(self):
+        partition_by = WindowSpec().partitionBy(F.col("cola"), F.col("colb"))
+        self.assertEqual("OVER (PARTITION BY cola, colb)", partition_by.sql())
+
+    def test_window_spec_order_by(self):
+        order_by = WindowSpec().orderBy("cola", "colb")
+        self.assertEqual("OVER (ORDER BY cola, colb)", order_by.sql())
+
+    def test_window_spec_rows_between(self):
+        rows_between = WindowSpec().rowsBetween(3, 5)
+        self.assertEqual("OVER ( ROWS BETWEEN 3 PRECEDING AND 5 FOLLOWING)", rows_between.sql())
+
+    def test_window_spec_range_between(self):
+        range_between = WindowSpec().rangeBetween(3, 5)
+        self.assertEqual("OVER ( RANGE BETWEEN 3 PRECEDING AND 5 FOLLOWING)", range_between.sql())
+
+    def test_window_partition_by(self):
+        partition_by = Window.partitionBy(F.col("cola"), F.col("colb"))
+        self.assertEqual("OVER (PARTITION BY cola, colb)", partition_by.sql())
+
+    def test_window_order_by(self):
+        order_by = Window.orderBy("cola", "colb")
+        self.assertEqual("OVER (ORDER BY cola, colb)", order_by.sql())
+
+    def test_window_rows_between(self):
+        rows_between = Window.rowsBetween(3, 5)
+        self.assertEqual("OVER ( ROWS BETWEEN 3 PRECEDING AND 5 FOLLOWING)", rows_between.sql())
+
+    def test_window_range_between(self):
+        range_between = Window.rangeBetween(3, 5)
+        self.assertEqual("OVER ( RANGE BETWEEN 3 PRECEDING AND 5 FOLLOWING)", range_between.sql())
+
+    def test_window_rows_unbounded(self):
+        rows_between_unbounded_start = Window.rowsBetween(Window.unboundedPreceding, 2)
+        self.assertEqual("OVER ( ROWS BETWEEN UNBOUNDED PRECEDING AND 2 FOLLOWING)", rows_between_unbounded_start.sql())
+        rows_between_unbounded_end = Window.rowsBetween(1, Window.unboundedFollowing)
+        self.assertEqual("OVER ( ROWS BETWEEN 1 PRECEDING AND UNBOUNDED FOLLOWING)", rows_between_unbounded_end.sql())
+        rows_between_unbounded_both = Window.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
+        self.assertEqual(
+            "OVER ( ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)", rows_between_unbounded_both.sql()
+        )
+
+    def test_window_range_unbounded(self):
+        range_between_unbounded_start = Window.rangeBetween(Window.unboundedPreceding, 2)
+        self.assertEqual(
+            "OVER ( RANGE BETWEEN UNBOUNDED PRECEDING AND 2 FOLLOWING)", range_between_unbounded_start.sql()
+        )
+        range_between_unbounded_end = Window.rangeBetween(1, Window.unboundedFollowing)
+        self.assertEqual("OVER ( RANGE BETWEEN 1 PRECEDING AND UNBOUNDED FOLLOWING)", range_between_unbounded_end.sql())
+        range_between_unbounded_both = Window.rangeBetween(Window.unboundedPreceding, Window.unboundedFollowing)
+        self.assertEqual(
+            "OVER ( RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)", range_between_unbounded_both.sql()
+        )