diff --git a/dataframe-operations.cabal b/dataframe-operations.cabal
--- a/dataframe-operations.cabal
+++ b/dataframe-operations.cabal
@@ -1,6 +1,6 @@
 cabal-version:      3.4
 name:               dataframe-operations
-version:            2.3.0.0
+version:            2.4.0.0
 synopsis:           Column operations, expression DSL, and statistics for the dataframe ecosystem.
 description:
     Untyped column operations (select, filter, sort, join, groupBy,
@@ -58,7 +58,7 @@
     build-depends:      base >= 4 && < 5,
                         bytestring >= 0.11 && < 0.14,
                         containers >= 0.6.7 && < 0.10,
-                        dataframe-core >= 2.3 && < 2.4,
+                        dataframe-core >= 2.4 && < 2.5,
                         dataframe-parsing >= 2.2 && < 2.3,
                         random >= 1.2 && < 2,
                         regex-tdfa >= 1.3.0 && < 2,
diff --git a/src/DataFrame/Functions.hs b/src/DataFrame/Functions.hs
--- a/src/DataFrame/Functions.hs
+++ b/src/DataFrame/Functions.hs
@@ -35,6 +35,7 @@
 import qualified Data.Maybe as Maybe
 import qualified Data.Text as T
 import Data.Time
+import Data.Type.Equality (testEquality, type (:~:) (Refl))
 import qualified Data.Vector as V
 import qualified Data.Vector.Unboxed as VU
 
@@ -47,6 +48,7 @@
  )
 import DataFrame.Operators
 import Text.Regex.TDFA
+import Type.Reflection (typeRep)
 import Prelude hiding (maximum, minimum)
 import Prelude as P
 
@@ -328,6 +330,49 @@
 {-# SPECIALIZE mean :: Expr Int32 -> Expr Double #-}
 {-# SPECIALIZE mean :: Expr Int64 -> Expr Double #-}
 {-# INLINEABLE mean #-}
+
+-- | The k largest values per group.
+topK :: (Columnable a, Ord a) => Int -> Expr a -> Expr [a]
+topK = kExtremes "topK" (>)
+{-# INLINEABLE topK #-}
+
+-- | The k smallest values per group.
+bottomK :: (Columnable a, Ord a) => Int -> Expr a -> Expr [a]
+bottomK = kExtremes "bottomK" (<)
+{-# INLINEABLE bottomK #-}
+
+{- | Top k values after applying an ordering function. Floating-point NaNs
+are dropped: they have no place in a total order, and admitting them would
+make the merge non-associative (results would depend on chunk boundaries in
+the lazy executor).
+-}
+kExtremes ::
+    forall a.
+    (Columnable a, Ord a) =>
+    T.Text -> (a -> a -> Bool) -> Int -> Expr a -> Expr [a]
+kExtremes opName wins k =
+    Agg
+        ( MergeAgg
+            (opName <> "_" <> T.pack (show k))
+            []
+            step
+            (L.foldl' step)
+            id
+        )
+  where
+    step acc x
+        | isNaNLike x = acc
+        | otherwise = insert x acc
+    insert x = P.take k . go
+      where
+        go (z : zs) | wins z x = z : go zs
+        go zs = x : zs
+
+    isNaNLike :: a -> Bool
+    isNaNLike x
+        | Just Refl <- testEquality (typeRep @a) (typeRep @Double) = isNaN x
+        | Just Refl <- testEquality (typeRep @a) (typeRep @Float) = isNaN x
+        | otherwise = False
 
 meanMaybe :: forall a. (Columnable a, Real a) => Expr (Maybe a) -> Expr Double
 meanMaybe = Agg (CollectAgg "meanMaybe" (mean' . optionalToDoubleVector))
diff --git a/src/DataFrame/Operations/Core.hs b/src/DataFrame/Operations/Core.hs
--- a/src/DataFrame/Operations/Core.hs
+++ b/src/DataFrame/Operations/Core.hs
@@ -71,7 +71,9 @@
     columnTypeString,
     fromList,
     fromVector,
+    materializeMerged,
     materializePacked,
+    mergedHead,
     toDoubleVector,
     toFloatVector,
     toIntVector,
@@ -571,6 +573,7 @@
                         columnType
                         : acc
     go acc i col@(PackedText _ _) = go acc i (materializePacked col)
+    go acc i col@(MergedColumn _ _) = go acc i (materializeMerged col)
 
 nulls :: Column -> Int
 nulls (BoxedColumn (Just bm) xs) =
@@ -957,4 +960,11 @@
         Just (UnboxedColumn Nothing (_ :: VU.Vector a)) -> UExpr (Col @a name)
         Just (PackedText (Just _) _) -> UExpr (Col @(Maybe T.Text) name)
         Just (PackedText Nothing _) -> UExpr (Col @T.Text name)
+        Just c@(MergedColumn _ _) -> case mergedHead c of
+            BoxedColumn (Just _) (_ :: V.Vector a) -> UExpr (Col @(Maybe a) name)
+            BoxedColumn Nothing (_ :: V.Vector a) -> UExpr (Col @a name)
+            _ ->
+                error $
+                    "showDerivedExpressions: merged column did not materialize boxed: "
+                        ++ T.unpack name
         Nothing -> error $ "showDerivedExpressions: column not found: " ++ T.unpack name
diff --git a/src/DataFrame/Operations/Permutation.hs b/src/DataFrame/Operations/Permutation.hs
--- a/src/DataFrame/Operations/Permutation.hs
+++ b/src/DataFrame/Operations/Permutation.hs
@@ -28,7 +28,12 @@
 import Data.Type.Equality (testEquality, (:~:) (Refl))
 import Data.Vector.Internal.Check (HasCallStack)
 import DataFrame.Errors (DataFrameException (..))
-import DataFrame.Internal.Column (Column (..), Columnable, atIndicesStable)
+import DataFrame.Internal.Column (
+    Column (..),
+    Columnable,
+    atIndicesStable,
+    materializeMerged,
+ )
 import DataFrame.Internal.DataFrame (
     DataFrame (..),
     columnNames,
@@ -132,6 +137,11 @@
                     (aj, oj, lj) = packedSlice p j
                  in sliceCmpBytes ai oi li aj oj lj
             Nothing -> \_ _ -> EQ
+        c@(MergedColumn _ _) -> case materializeMerged c of
+            BoxedColumn _ (v :: V.Vector b) -> case testEquality (typeRep @a) (typeRep @b) of
+                Just Refl -> \i j -> compare (v `V.unsafeIndex` i) (v `V.unsafeIndex` j)
+                Nothing -> \_ _ -> EQ
+            _ -> \_ _ -> EQ
 sortOrderComparator (Desc (Col name :: Expr a)) df =
     case unsafeGetColumn name df of
         BoxedColumn _ (v :: V.Vector b) -> case testEquality (typeRep @a) (typeRep @b) of
@@ -146,6 +156,11 @@
                     (aj, oj, lj) = packedSlice p j
                  in sliceCmpBytes aj oj lj ai oi li
             Nothing -> \_ _ -> EQ
+        c@(MergedColumn _ _) -> case materializeMerged c of
+            BoxedColumn _ (v :: V.Vector b) -> case testEquality (typeRep @a) (typeRep @b) of
+                Just Refl -> \i j -> compare (v `V.unsafeIndex` j) (v `V.unsafeIndex` i)
+                Nothing -> \_ _ -> EQ
+            _ -> \_ _ -> EQ
 sortOrderComparator _ _ = error "Sorting on compound column"
 
 -- | Sort row indices using a comparator function.
diff --git a/src/DataFrame/Operations/Statistics.hs b/src/DataFrame/Operations/Statistics.hs
--- a/src/DataFrame/Operations/Statistics.hs
+++ b/src/DataFrame/Operations/Statistics.hs
@@ -272,6 +272,7 @@
         Just Refl -> VG.sum column
         Nothing -> 0
     Just (PackedText _ _) -> 0
+    Just (MergedColumn _ _) -> 0 -- matches the old eager These column (type never Num)
 sum expr df = case interpret df expr of
     Left e -> throw e
     Right (TColumn xs) -> case toVector @a @V.Vector xs of
diff --git a/src/DataFrame/Operations/Subset.hs b/src/DataFrame/Operations/Subset.hs
--- a/src/DataFrame/Operations/Subset.hs
+++ b/src/DataFrame/Operations/Subset.hs
@@ -178,6 +178,8 @@
             ColumnsNotFoundException [filterColumnName] "filter" (M.keys $ columnIndices df)
     Just c@(PackedText _ _) ->
         filter e condition (insertColumn filterColumnName (materializePacked c) df)
+    Just c@(MergedColumn _ _) ->
+        filter e condition (insertColumn filterColumnName (materializeMerged c) df)
     Just _col@(BoxedColumn bm (column :: V.Vector b)) ->
         case testEquality (typeRep @a) (typeRep @b) of
             Just Refl -> filterByVector filterColumnName column condition df
@@ -518,6 +520,7 @@
 
 -- | Convert any Column to a vector of Text labels (one per row).
 columnToTextVec :: Column -> V.Vector T.Text
+columnToTextVec c@(MergedColumn _ _) = columnToTextVec (materializeMerged c)
 columnToTextVec (BoxedColumn bm (col' :: V.Vector a)) =
     case bm of
         Nothing -> case testEquality (typeRep @a) (typeRep @T.Text) of