feat(table): implement pivot_wider API

ibis/backends/tests/test_generic.py

@@ -1003,6 +1003,23 @@ def test_pivot_longer(backend):
     assert not df.empty
 
 
+@pytest.mark.notyet(["datafusion"], raises=com.OperationNotDefinedError)
+def test_pivot_wider(backend):
+    diamonds = backend.diamonds
+    expr = (
+        diamonds.group_by(["cut", "color"])
+        .agg(carat=_.carat.mean())
+        .pivot_wider(
+            names_from="cut", values_from="carat", names_sort=True, values_agg="mean"
+        )
+    )
+    df = expr.execute()
+    assert set(df.columns) == {"color"} | set(
+        diamonds[["cut"]].distinct().cut.execute()
+    )
+    assert len(df) == diamonds.color.nunique().execute()
+
+
 @pytest.mark.parametrize(
     "on",
     [

ibis/expr/analysis.py

@@ -846,3 +846,13 @@ def finder(node):
         )
 
     return g.traverse(finder, nodes, filter=ops.Node)
+
+
+def find_toplevel_aggs(nodes: Iterable[ops.Node]) -> Iterator[ops.Table]:
+    def finder(node):
+        return (
+            isinstance(node, ops.Value),
+            node if isinstance(node, ops.Reduction) else None,
+        )
+
+    return g.traverse(finder, nodes, filter=ops.Node)

ibis/expr/selectors.py

@@ -94,7 +94,7 @@ def expand(self, table: ir.Table) -> Sequence[ir.Value]:
         return [col for column in table.columns if self.predicate(col := table[column])]
 
     def __and__(self, other: Selector) -> Predicate:
-        """Compute the conjunction of two `Selectors`.
+        """Compute the conjunction of two `Selector`s.
 
         Parameters
         ----------
@@ -104,7 +104,7 @@ def __and__(self, other: Selector) -> Predicate:
         return self.__class__(lambda col: self.predicate(col) and other.predicate(col))
 
     def __or__(self, other: Selector) -> Predicate:
-        """Compute the disjunction of two `Selectors`.
+        """Compute the disjunction of two `Selector`s.
 
         Parameters
         ----------
@@ -114,7 +114,7 @@ def __or__(self, other: Selector) -> Predicate:
         return self.__class__(lambda col: self.predicate(col) or other.predicate(col))
 
     def __invert__(self) -> Predicate:
-        """Compute the logical negation of two `Selectors`."""
+        """Compute the logical negation of two `Selector`s."""
         return self.__class__(lambda col: not self.predicate(col))
 
 
@@ -457,3 +457,8 @@ def last() -> Predicate:
 def all() -> Predicate:
     """Return every column from a table."""
     return r[:]
+
+
+def _to_selector(obj: str | Selector) -> Selector:
+    """Convert an object to a `Selector`."""
+    return c(obj) if isinstance(obj, str) else obj