fix(pandas): fix first and last over windows

fixes #5417

ibis/backends/pandas/execution/generic.py

@@ -772,13 +772,21 @@ def execute_reduction_series_mask(op, data, mask, aggcontext=None, **kwargs):
 @execute_node.register(ops.First, pd.Series, (pd.Series, type(None)))
 def execute_first_series_mask(op, data, mask, aggcontext=None, **kwargs):
     operand = data[mask] if mask is not None else data
-    return aggcontext.agg(operand, lambda x: x.iloc[0])
+
+    def _first(x):
+        return getattr(x, "iloc", x)[0]
+
+    return aggcontext.agg(operand, _first)
 
 
 @execute_node.register(ops.Last, pd.Series, (pd.Series, type(None)))
 def execute_last_series_mask(op, data, mask, aggcontext=None, **kwargs):
     operand = data[mask] if mask is not None else data
-    return aggcontext.agg(operand, lambda x: x.iloc[-1])
+
+    def _last(x):
+        return getattr(x, "iloc", x)[-1]
+
+    return aggcontext.agg(operand, _last)
 
 
 @execute_node.register(

ibis/backends/pandas/execution/window.py

@@ -87,9 +87,17 @@ def _post_process_order_by(
     assert order_by and not group_by
     indexed_parent = parent.set_index(order_by)
     index = indexed_parent.index
-    names = index.names
-    if len(names) > 1:
-        series = series.reorder_levels(names)
+
+    # get the names of the levels that will be in the result
+    series_index_names = frozenset(series.index.names)
+
+    # get the levels common to series.index, in the order that they occur in
+    # the parent's index
+    reordered_levels = [name for name in index.names if name in series_index_names]
+
+    if len(reordered_levels) > 1:
+        series = series.reorder_levels(reordered_levels)
+
     series = series.iloc[index.argsort(kind='mergesort')]
     return series
 

ibis/backends/pandas/tests/execution/test_window.py

@@ -118,6 +118,23 @@ def test_last(t, df):
     assert result == expected
 
 
+def test_first_and_last_over_window(t):
+    def simple_window_ops(table):
+        w = ibis.window(
+            order_by=[table.plain_uint64, table.dup_ints],
+            preceding=1,
+            following=0,
+        )
+        return table.mutate(
+            x_first=t.plain_uint64.first().over(w),
+            x_last=t.plain_uint64.last().over(w),
+            y_first=t.dup_ints.first().over(w),
+            y_last=t.dup_ints.last().over(w),
+        )
+
+    assert simple_window_ops(t).execute() is not None
+
+
 def test_group_by_mutate_analytic(t, df):
     gb = t.group_by(t.dup_strings)
     expr = gb.mutate(