feat(pyspark): implement Distinct for pyspark

Author: cpcloud

ibis/backends/pyspark/compiler.py

@@ -1176,6 +1176,12 @@ def compile_join(t, expr, scope, timecontext, *, how):
     return left_df.join(right_df, pred_columns, how)
 
 
+@compiles(ops.Distinct)
+def compile_distinct(t, expr, scope, timecontext):
+    op = expr.op()
+    return t.translate(op.table, scope, timecontext).distinct()
+
+
 def _canonicalize_interval(t, interval, scope, timecontext, **kwargs):
     """Convert interval to integer timestamp of second
 

ibis/tests/sql/test_compiler.py

@@ -71,7 +71,7 @@ def test_table_distinct(con):
 
 def test_column_distinct(con):
     t = con.table('functional_alltypes')
-    expr = t[["string_col"]].distinct()
+    expr = t[t.string_col].distinct()
 
     result = Compiler.to_sql(expr)
     expected = """SELECT DISTINCT `string_col`

ibis/tests/sql/test_sqlalchemy.py

@@ -656,7 +656,7 @@ def test_not_exists(con, not_exists):
             ).distinct(),
         ),
         (
-            lambda t: t[["string_col"]].distinct(),
+            lambda t: t[t.string_col].distinct(),
             lambda sat: sa.select([sat.c.string_col.distinct()]),
         ),
         (