feat(pyspark): topk support

Author: cpcloud

ibis/backends/tests/test_aggregation.py

@@ -365,7 +365,7 @@ def test_topk_op(backend, alltypes, df, result_fn, expected_fn):
         )
     ],
 )
-@mark.notimpl(["datafusion", "pandas", "dask", "pyspark"])
+@mark.notimpl(["datafusion", "pandas", "dask"])
 def test_topk_filter_op(alltypes, df, result_fn, expected_fn):
     # TopK expression will order rows by "count" but each backend
     # can have different result for that.

ibis/tests/expr/test_analytics.py

@@ -15,7 +15,6 @@
 import pytest
 
 import ibis
-import ibis.expr.operations as ops
 import ibis.expr.types as ir
 from ibis.tests.expr.mocks import MockBackend
 from ibis.tests.util import assert_equal
@@ -111,7 +110,7 @@ def test_topk_analysis_bug(airlines):
 
     filtered = t.filter([delay_filter])
 
-    assert delay_filter._to_semi_join(t).equals(filtered)
+    assert delay_filter._to_semi_join(t)[t].equals(filtered)
 
 
 def test_topk_function_late_bind(airlines):