dataframe-core 2.4.0.0 → 2.5.0.0

raw patch · 62 files changed

+11479/−7216 lines, 62 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• dataframe-core.cabal +38/−23
• src-internal/DataFrame/Errors.hs +2/−0
• src-internal/DataFrame/Internal/AggKernel.hs +0/−260
• src-internal/DataFrame/Internal/AggKernelDirect.hs +0/−338
• src-internal/DataFrame/Internal/AggKernelPar.hs +0/−391
• src-internal/DataFrame/Internal/AggPlan.hs +0/−299
• src-internal/DataFrame/Internal/Aggregation/Kernel/Dense.hs +832/−0
• src-internal/DataFrame/Internal/Aggregation/Kernel/Fused.hs +757/−0
• src-internal/DataFrame/Internal/Aggregation/Kernel/Moments.hs +389/−0
• src-internal/DataFrame/Internal/Aggregation/Kernel/Scatter.hs +818/−0
• src-internal/DataFrame/Internal/Aggregation/Plan.hs +222/−0
• src-internal/DataFrame/Internal/Aggregation/Reduction.hs +44/−0
• src-internal/DataFrame/Internal/Algorithms/Hash.hs +113/−0
• src-internal/DataFrame/Internal/Algorithms/Rank/Radix.hs +101/−0
• src-internal/DataFrame/Internal/Algorithms/Sort/Radix/Parallel.hs +332/−0
• src-internal/DataFrame/Internal/Column.hs +26/−1835
• src-internal/DataFrame/Internal/Column/Base.hs +333/−0
• src-internal/DataFrame/Internal/Column/Bitmap.hs +188/−0
• src-internal/DataFrame/Internal/Column/Builder.hs +306/−0
• src-internal/DataFrame/Internal/Column/Conversion.hs +441/−0
• src-internal/DataFrame/Internal/Column/Encode.hs +185/−0
• src-internal/DataFrame/Internal/Column/Merge.hs +138/−0
• src-internal/DataFrame/Internal/Column/Operations.hs +1369/−0
• src-internal/DataFrame/Internal/Column/Properties.hs +153/−0
• src-internal/DataFrame/Internal/Column/Types.hs +161/−0
• src-internal/DataFrame/Internal/ColumnBuilder.hs +0/−299
• src-internal/DataFrame/Internal/ColumnMerge.hs +0/−185
• src-internal/DataFrame/Internal/Control/Concurrent.hs +177/−0
• src-internal/DataFrame/Internal/Data/HashTable.hs +108/−0
• src-internal/DataFrame/Internal/Data/PackedText.hs +361/−0
• src-internal/DataFrame/Internal/Data/PackedText/Utf8.hs +95/−0
• src-internal/DataFrame/Internal/DataFrame.hs +100/−46
• src-internal/DataFrame/Internal/DictEncode.hs +0/−171
• src-internal/DataFrame/Internal/Display/Pretty.hs +129/−0
• src-internal/DataFrame/Internal/Expression.hs +1/−1
• src-internal/DataFrame/Internal/Expression/Operators.hs +425/−0
• src-internal/DataFrame/Internal/Expression/Operators/Nullable.hs +467/−0
• src-internal/DataFrame/Internal/Expression/Simplify.hs +417/−0
• src-internal/DataFrame/Internal/Grouping.hs +667/−78
• src-internal/DataFrame/Internal/Grouping/Direct.hs +859/−0
• src-internal/DataFrame/Internal/Grouping/Partitioned.hs +406/−0
• src-internal/DataFrame/Internal/GroupingDirect.hs +0/−234
• src-internal/DataFrame/Internal/GroupingPar.hs +0/−308
• src-internal/DataFrame/Internal/Hash.hs +0/−113
• src-internal/DataFrame/Internal/HashTable.hs +0/−94
• src-internal/DataFrame/Internal/Interpreter.hs +47/−5
• src-internal/DataFrame/Internal/Nullable.hs +0/−467
• src-internal/DataFrame/Internal/PackedText.hs +0/−238
• src-internal/DataFrame/Internal/ParRadixSort.hs +0/−272
• src-internal/DataFrame/Internal/Pretty.hs +0/−129
• src-internal/DataFrame/Internal/RadixRank.hs +0/−101
• src-internal/DataFrame/Internal/Row.hs +17/−5
• src-internal/DataFrame/Internal/Row/RowHash.hs +240/−0
• src-internal/DataFrame/Internal/RowHash.hs +0/−223
• src-internal/DataFrame/Internal/Simplify.hs +0/−417
• src-internal/DataFrame/Internal/Types.hs +0/−161
• src-internal/DataFrame/Internal/Utf8.hs +0/−95
• src-internal/DataFrame/Operators.hs +0/−425
• src/DataFrame/Core.hs +1/−1
• src/DataFrame/Expression/Operators.hs +12/−0
• src/DataFrame/Typed/Schema.hs +1/−1
• src/DataFrame/Typed/Types.hs +1/−1

@@ -1,6 +1,6 @@ cabal-version:      3.4 name:               dataframe-core-version:            2.4.0.0+version:            2.5.0.0 synopsis:           Core data structures for the dataframe library. description:     Minimal interchange-format types for the @dataframe@ ecosystem:@@ -32,42 +32,57 @@     import:             warnings     exposed-modules:                         DataFrame.Core+                        DataFrame.Errors+                        DataFrame.Expression.Operators                         DataFrame.Typed.Freeze                         DataFrame.Typed.Generic                         DataFrame.Typed.Record                         DataFrame.Typed.Schema                         DataFrame.Typed.Types                         DataFrame.Typed.Util-                        DataFrame.Errors-                        DataFrame.Operators                         DataFrame.Display.Terminal.Colours                         DataFrame.Display.Terminal.PrettyPrint-                        DataFrame.Internal.AggKernel-                        DataFrame.Internal.AggKernelDirect-                        DataFrame.Internal.AggKernelPar-                        DataFrame.Internal.AggPlan-                        DataFrame.Internal.GroupingDirect+                        -- Concurrency+                        DataFrame.Internal.Control.Concurrent+                        -- Aggregation kernels and planning.+                        DataFrame.Internal.Aggregation.Kernel.Dense+                        DataFrame.Internal.Aggregation.Kernel.Fused+                        DataFrame.Internal.Aggregation.Kernel.Moments+                        DataFrame.Internal.Aggregation.Kernel.Scatter+                        DataFrame.Internal.Aggregation.Plan+                        DataFrame.Internal.Aggregation.Reduction+                        -- Shared algorithms.+                        DataFrame.Internal.Algorithms.Hash+                        DataFrame.Internal.Algorithms.Rank.Radix+                        DataFrame.Internal.Algorithms.Sort.Radix.Parallel+                        -- The column representation.                         DataFrame.Internal.Column-                        DataFrame.Internal.ColumnBuilder-                        DataFrame.Internal.ColumnMerge+                        DataFrame.Internal.Column.Base+                        DataFrame.Internal.Column.Bitmap+                        DataFrame.Internal.Column.Builder+                        DataFrame.Internal.Column.Conversion+                        DataFrame.Internal.Column.Encode+                        DataFrame.Internal.Column.Merge+                        DataFrame.Internal.Column.Operations+                        DataFrame.Internal.Column.Properties+                        DataFrame.Internal.Column.Types+                        -- Backing data structures.+                        DataFrame.Internal.Data.HashTable+                        DataFrame.Internal.Data.PackedText+                        DataFrame.Internal.Data.PackedText.Utf8                         DataFrame.Internal.DataFrame-                        DataFrame.Internal.DictEncode+                        DataFrame.Internal.Display.Pretty+                        -- Expressions: syntax, operators, simplification.                         DataFrame.Internal.Expression+                        DataFrame.Internal.Expression.Operators+                        DataFrame.Internal.Expression.Operators.Nullable+                        DataFrame.Internal.Expression.Simplify                         DataFrame.Internal.Grouping-                        DataFrame.Internal.GroupingPar-                        DataFrame.Internal.Hash-                        DataFrame.Internal.HashTable+                        DataFrame.Internal.Grouping.Direct+                        DataFrame.Internal.Grouping.Partitioned                         DataFrame.Internal.Interpreter-                        DataFrame.Internal.Nullable-                        DataFrame.Internal.PackedText-                        DataFrame.Internal.ParRadixSort-                        DataFrame.Internal.Pretty-                        DataFrame.Internal.RadixRank-                        DataFrame.Internal.RowHash                         DataFrame.Internal.Row-                        DataFrame.Internal.Simplify-                        DataFrame.Internal.Types-                        DataFrame.Internal.Utf8+                        DataFrame.Internal.Row.RowHash     build-depends:      base >= 4 && < 5,                         containers >= 0.6.7 && < 0.10,                         primitive >= 0.7 && < 0.11,

@@ -34,6 +34,7 @@     ColumnsNotFoundException :: [T.Text] -> T.Text -> [T.Text] -> DataFrameException     EmptyDataSetException :: T.Text -> DataFrameException     InternalException :: T.Text -> DataFrameException+    ExpectedNonNullableException :: DataFrameException     NonColumnReferenceException :: T.Text -> DataFrameException     UnaggregatedException :: T.Text -> DataFrameException     WrongQuantileNumberException :: Int -> DataFrameException@@ -42,6 +43,7 @@  instance Show DataFrameException where     show :: DataFrameException -> String+    show ExpectedNonNullableException = "Expected non-nullable column"     show (TypeMismatchException context) =         let             errorString =

@@ -1,260 +0,0 @@-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE ExplicitNamespaces #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeApplications #-}---- | Vectorized scatter-accumulate aggregation kernel.-module DataFrame.Internal.AggKernel (-    Reduction (..),-    scatterReduce,-    scatterColumnToDouble,-) where--import Data.Type.Equality (TestEquality (..), type (:~:) (Refl))-import qualified Data.Vector.Unboxed as VU-import qualified Data.Vector.Unboxed.Mutable as VUM--import Control.Monad (when)-import Control.Monad.ST (ST, runST)-import DataFrame.Internal.Column (-    Column (..),-    Columnable,-    fromUnboxedVector,-    materializePacked,- )-import Type.Reflection (typeRep)--{- | A recognised fast-path reduction over a single value column. The element-type (Int vs Double) is resolved at scatter time; sum/min/max preserve the-column's element type, everything else produces a Double column.--}-data Reduction-    = RSum-    | RCount-    | RMin-    | RMax-    | RMean-    | RStd-    | RVar-    | RTop2Sum-    deriving (Eq, Show)--{- | Coerce an unboxed Int or Double column to an unboxed Double vector for the-moment/mean/sd/median family. Returns 'Nothing' for boxed, nullable, or other-element types (the caller then falls back to the interpreter).--}-scatterColumnToDouble :: Column -> Maybe (VU.Vector Double)-scatterColumnToDouble = \case-    UnboxedColumn Nothing (v :: VU.Vector a) ->-        case testEquality (typeRep @a) (typeRep @Double) of-            Just Refl -> Just v-            Nothing -> case testEquality (typeRep @a) (typeRep @Int) of-                Just Refl -> Just (VU.map fromIntegral v)-                Nothing -> Nothing-    p@(PackedText _ _) -> scatterColumnToDouble (materializePacked p)-    _ -> Nothing--scatterReduce ::-    Reduction -> VU.Vector Int -> Int -> Column -> Maybe Column-scatterReduce red g nGroups col = case col of-    UnboxedColumn Nothing (v :: VU.Vector a) ->-        case testEquality (typeRep @a) (typeRep @Int) of-            Just Refl -> Just (reduceTyped red g nGroups v intIdent)-            Nothing -> case testEquality (typeRep @a) (typeRep @Double) of-                Just Refl -> Just (reduceTyped red g nGroups v dblIdent)-                Nothing -> Nothing-    p@(PackedText _ _) -> scatterReduce red g nGroups (materializePacked p)-    _ -> Nothing-{-# INLINEABLE scatterReduce #-}---- | Per-type seed identities for the order-preserving reductions.-data Idents a = Idents {minSeed :: !a, maxSeed :: !a}--intIdent :: Idents Int-intIdent = Idents maxBound minBound--dblIdent :: Idents Double-dblIdent = Idents (1 / 0) (negate (1 / 0))--reduceTyped ::-    forall a.-    (Columnable a, VU.Unbox a, Num a, Ord a, Real a) =>-    Reduction -> VU.Vector Int -> Int -> VU.Vector a -> Idents a -> Column-reduceTyped red g nGroups v idents = case red of-    RCount -> fromUnboxedVector (countScatter g nGroups)-    RSum -> fromUnboxedVector (sumScatter g nGroups v)-    RMin -> fromUnboxedVector (extremaScatter min (minSeed idents) g nGroups v)-    RMax -> fromUnboxedVector (extremaScatter max (maxSeed idents) g nGroups v)-    RMean -> fromUnboxedVector (meanScatter g nGroups v)-    RVar -> fromUnboxedVector (varScatter False g nGroups v)-    RStd -> fromUnboxedVector (varScatter True g nGroups v)-    RTop2Sum -> fromUnboxedVector (top2Scatter g nGroups v)-{-# INLINE reduceTyped #-}--countScatter :: VU.Vector Int -> Int -> VU.Vector Int-countScatter g nGroups = runST $ do-    cnt <- VUM.replicate nGroups (0 :: Int)-    let n = VU.length g-        go !i-            | i >= n = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                c <- VUM.unsafeRead cnt k-                VUM.unsafeWrite cnt k (c + 1)-                go (i + 1)-    go 0-    VU.unsafeFreeze cnt--sumScatter ::-    (VU.Unbox a, Num a) => VU.Vector Int -> Int -> VU.Vector a -> VU.Vector a-sumScatter g nGroups v = runST $ do-    s <- VUM.replicate nGroups 0-    let n = VU.length v-        go !i-            | i >= n = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                cur <- VUM.unsafeRead s k-                VUM.unsafeWrite s k (cur + VU.unsafeIndex v i)-                go (i + 1)-    go 0-    VU.unsafeFreeze s-{-# INLINE sumScatter #-}--extremaScatter ::-    (VU.Unbox a) =>-    (a -> a -> a) -> a -> VU.Vector Int -> Int -> VU.Vector a -> VU.Vector a-extremaScatter combine seed g nGroups v = runST $ do-    m <- VUM.replicate nGroups seed-    let n = VU.length v-        go !i-            | i >= n = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                cur <- VUM.unsafeRead m k-                VUM.unsafeWrite m k (combine cur (VU.unsafeIndex v i))-                go (i + 1)-    go 0-    VU.unsafeFreeze m-{-# INLINE extremaScatter #-}--meanScatter ::-    (VU.Unbox a, Real a) => VU.Vector Int -> Int -> VU.Vector a -> VU.Vector Double-meanScatter g nGroups v = runST $ do-    s <- VUM.replicate nGroups (0 :: Double)-    cnt <- VUM.replicate nGroups (0 :: Int)-    scatterSumCount g v s cnt-    finalizeMean nGroups s cnt-{-# INLINE meanScatter #-}--scatterSumCount ::-    (VU.Unbox a, Real a) =>-    VU.Vector Int ->-    VU.Vector a ->-    VUM.MVector s Double ->-    VUM.MVector s Int ->-    ST s ()-scatterSumCount g v s cnt = go 0-  where-    n = VU.length v-    go !i-        | i >= n = pure ()-        | otherwise = do-            let !k = VU.unsafeIndex g i-                !x = realToFrac (VU.unsafeIndex v i)-            curS <- VUM.unsafeRead s k-            VUM.unsafeWrite s k (curS + x)-            curC <- VUM.unsafeRead cnt k-            VUM.unsafeWrite cnt k (curC + 1)-            go (i + 1)-{-# INLINE scatterSumCount #-}--finalizeMean ::-    Int -> VUM.MVector s Double -> VUM.MVector s Int -> ST s (VU.Vector Double)-finalizeMean nGroups s cnt = do-    out <- VUM.new nGroups-    let go !k-            | k >= nGroups = pure ()-            | otherwise = do-                sv <- VUM.unsafeRead s k-                c <- VUM.unsafeRead cnt k-                VUM.unsafeWrite out k (if c == 0 then 0 / 0 else sv / fromIntegral c)-                go (k + 1)-    go 0-    VU.unsafeFreeze out--varScatter ::-    (VU.Unbox a, Real a) =>-    Bool -> VU.Vector Int -> Int -> VU.Vector a -> VU.Vector Double-varScatter takeSqrt g nGroups v = runST $ do-    cnt <- VUM.replicate nGroups (0 :: Int)-    meanV <- VUM.replicate nGroups (0 :: Double)-    m2 <- VUM.replicate nGroups (0 :: Double)-    let n = VU.length v-        go !i-            | i >= n = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                    !x = realToFrac (VU.unsafeIndex v i)-                c <- VUM.unsafeRead cnt k-                mu <- VUM.unsafeRead meanV k-                mm <- VUM.unsafeRead m2 k-                let !c' = c + 1-                    !delta = x - mu-                    !mu' = mu + delta / fromIntegral c'-                    !mm' = mm + delta * (x - mu')-                VUM.unsafeWrite cnt k c'-                VUM.unsafeWrite meanV k mu'-                VUM.unsafeWrite m2 k mm'-                go (i + 1)-    go 0-    out <- VUM.new nGroups-    let fin !k-            | k >= nGroups = pure ()-            | otherwise = do-                c <- VUM.unsafeRead cnt k-                mm <- VUM.unsafeRead m2 k-                let var = if c < 2 then 0 else mm / fromIntegral (c - 1)-                VUM.unsafeWrite out k (if takeSqrt then sqrt var else var)-                fin (k + 1)-    fin 0-    VU.unsafeFreeze out-{-# INLINE varScatter #-}--top2Scatter ::-    (VU.Unbox a, Real a) => VU.Vector Int -> Int -> VU.Vector a -> VU.Vector Double-top2Scatter g nGroups v = runST $ do-    let ninf = negate (1 / 0) :: Double-    m1 <- VUM.replicate nGroups ninf-    m2 <- VUM.replicate nGroups ninf-    let n = VU.length v-        go !i-            | i >= n = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                    !x = realToFrac (VU.unsafeIndex v i)-                a1 <- VUM.unsafeRead m1 k-                if x > a1-                    then do-                        VUM.unsafeWrite m1 k x-                        VUM.unsafeWrite m2 k a1-                    else do-                        a2 <- VUM.unsafeRead m2 k-                        when (x > a2) (VUM.unsafeWrite m2 k x)-                go (i + 1)-    go 0-    out <- VUM.new nGroups-    let fin !k-            | k >= nGroups = pure ()-            | otherwise = do-                a1 <- VUM.unsafeRead m1 k-                a2 <- VUM.unsafeRead m2 k-                let s = (if isInfinite a1 then 0 else a1) + (if isInfinite a2 then 0 else a2)-                VUM.unsafeWrite out k s-                fin (k + 1)-    fin 0-    VU.unsafeFreeze out-{-# INLINE top2Scatter #-}

@@ -1,338 +0,0 @@-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE ExplicitNamespaces #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeApplications #-}--module DataFrame.Internal.AggKernelDirect (-    directThreshold,-    directReduce,-) where--import Control.Concurrent (forkIO, getNumCapabilities)-import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)-import Control.Exception (SomeException, throwIO, try)-import Data.Type.Equality (TestEquality (..), type (:~:) (Refl))-import qualified Data.Vector.Unboxed as VU-import qualified Data.Vector.Unboxed.Mutable as VUM-import System.IO.Unsafe (unsafePerformIO)-import Type.Reflection (typeRep)--import DataFrame.Internal.AggKernel (Reduction (..))-import DataFrame.Internal.Column (-    Column (..),-    fromUnboxedVector,-    materializePacked,- )--{- | Group-domain size at or below which the direct-indexed accumulator path is-taken; wider domains keep the group-range kernel. The admitted reductions are-order-independent, so the per-worker accumulator merge is exact.--}-directThreshold :: Int-directThreshold = 262144--capabilities :: Int-capabilities = unsafePerformIO getNumCapabilities-{-# NOINLINE capabilities #-}--{- | Below this many rows the parallel fan-out is not worth it; a single-sequential direct pass runs instead (tiny accumulator, one tight loop). Matches-the grouping/scatter parallel threshold.--}-parThreshold :: Int-parThreshold = 200000--{- | Run a recognised reduction through the direct-indexed path. 'Nothing' (so-the caller falls back to the order-preserving kernel) unless the reduction is-order-independent at this element type AND the column is a clean unboxed Int/Double.--}-directReduce :: Reduction -> VU.Vector Int -> Int -> Column -> Maybe Column-directReduce red g nGroups col = case col of-    UnboxedColumn Nothing (v :: VU.Vector a) ->-        case testEquality (typeRep @a) (typeRep @Int) of-            Just Refl -> directInt red g nGroups v-            Nothing -> case testEquality (typeRep @a) (typeRep @Double) of-                Just Refl -> directDouble red g nGroups v-                Nothing -> Nothing-    p@(PackedText _ _) -> directReduce red g nGroups (materializePacked p)-    _ -> Nothing-{-# INLINEABLE directReduce #-}---- | The order-independent reductions over an Int column.-directInt :: Reduction -> VU.Vector Int -> Int -> VU.Vector Int -> Maybe Column-directInt red g nGroups v = case red of-    RCount -> Just (fromUnboxedVector (countDirect g nGroups (VU.length v)))-    RSum -> Just (fromUnboxedVector (sumIntDirect g nGroups v))-    RMin -> Just (fromUnboxedVector (extremaIntDirect True g nGroups v))-    RMax -> Just (fromUnboxedVector (extremaIntDirect False g nGroups v))-    RMean -> Just (fromUnboxedVector (meanIntDirect g nGroups v))-    _ -> Nothing--{- | Over a Double column only @count@ is order-independent; the float-sum/mean/variance reductions must keep the order-preserving kernel.--}-directDouble ::-    Reduction -> VU.Vector Int -> Int -> VU.Vector Double -> Maybe Column-directDouble red g nGroups v = case red of-    RCount -> Just (fromUnboxedVector (countDirect g nGroups (VU.length v)))-    _ -> Nothing---- | Whether to fan out at this row count.-shouldPar :: Int -> Bool-shouldPar n = n >= parThreshold && capabilities > 1--{- | Fork @caps@ workers over disjoint contiguous row ranges of @[0, n)@, each-producing its own private accumulator (no shared array, no sync). Returns the-partials in worker order for the caller's merge; rethrows the first failure.--}-runPartialsOver ::-    Int -> Int -> (Int -> Int -> IO (VUM.IOVector Int)) -> IO [VUM.IOVector Int]-runPartialsOver n caps fill = do-    let !per = (n + caps - 1) `div` caps-        spawn w = do-            var <- newEmptyMVar-            let !lo = min n (w * per)-                !hi = min n (lo + per)-            _ <- forkIO (try (fill lo hi) >>= putMVar var)-            pure var-    vars <- mapM spawn [0 .. caps - 1]-    results <- mapM takeMVar vars-    mapM (either (throwIO @SomeException) pure) results--{- | As 'runPartialsOver' but each worker produces a PAIR of accumulators (e.g.-sum and count for the fused integer mean).--}-runPartialsPairOver ::-    Int ->-    Int ->-    (Int -> Int -> IO (VUM.IOVector Int, VUM.IOVector Int)) ->-    IO [(VUM.IOVector Int, VUM.IOVector Int)]-runPartialsPairOver n caps fill = do-    let !per = (n + caps - 1) `div` caps-        spawn w = do-            var <- newEmptyMVar-            let !lo = min n (w * per)-                !hi = min n (lo + per)-            _ <- forkIO (try (fill lo hi) >>= putMVar var)-            pure var-    vars <- mapM spawn [0 .. caps - 1]-    results <- mapM takeMVar vars-    mapM (either (throwIO @SomeException) pure) results------------------------------------------------------------------------------------ Count (order-independent: per-group row count)----------------------------------------------------------------------------------countDirect :: VU.Vector Int -> Int -> Int -> VU.Vector Int-countDirect g nGroups n-    | not (shouldPar n) =-        unsafePerformIO (countChunk g nGroups 0 n >>= VU.unsafeFreeze)-    | otherwise = unsafePerformIO $ do-        parts <- runPartialsOver n capabilities (countChunk g nGroups)-        mergeIntSum nGroups parts-{-# NOINLINE countDirect #-}--countChunk :: VU.Vector Int -> Int -> Int -> Int -> IO (VUM.IOVector Int)-countChunk g nGroups lo hi = do-    acc <- VUM.replicate nGroups (0 :: Int)-    let go !i-            | i >= hi = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                c <- VUM.unsafeRead acc k-                VUM.unsafeWrite acc k (c + 1)-                go (i + 1)-    go lo-    pure acc------------------------------------------------------------------------------------ Integer sum (exact: merge order irrelevant)----------------------------------------------------------------------------------sumIntDirect :: VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Int-sumIntDirect g nGroups v-    | not (shouldPar n) =-        unsafePerformIO (sumIntChunk g v nGroups 0 n >>= VU.unsafeFreeze)-    | otherwise = unsafePerformIO $ do-        parts <- runPartialsOver n capabilities (sumIntChunk g v nGroups)-        mergeIntSum nGroups parts-  where-    !n = VU.length v-{-# NOINLINE sumIntDirect #-}--sumIntChunk ::-    VU.Vector Int -> VU.Vector Int -> Int -> Int -> Int -> IO (VUM.IOVector Int)-sumIntChunk g v nGroups lo hi = do-    acc <- VUM.replicate nGroups (0 :: Int)-    let go !i-            | i >= hi = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                c <- VUM.unsafeRead acc k-                VUM.unsafeWrite acc k (c + VU.unsafeIndex v i)-                go (i + 1)-    go lo-    pure acc------------------------------------------------------------------------------------ Integer min / max (order-independent)----------------------------------------------------------------------------------extremaIntDirect ::-    Bool -> VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Int-extremaIntDirect isMin g nGroups v-    | not (shouldPar n) =-        unsafePerformIO (extremaIntChunk isMin g v nGroups 0 n >>= VU.unsafeFreeze)-    | otherwise = unsafePerformIO $ do-        parts <- runPartialsOver n capabilities (extremaIntChunk isMin g v nGroups)-        mergeExtremaInt isMin nGroups parts-  where-    !n = VU.length v-{-# NOINLINE extremaIntDirect #-}--extremaIntChunk ::-    Bool ->-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    Int ->-    Int ->-    IO (VUM.IOVector Int)-extremaIntChunk isMin g v nGroups lo hi = do-    let !seed = if isMin then maxBound else minBound-        combine a b = if isMin then min a b else max a b-    acc <- VUM.replicate nGroups seed-    let go !i-            | i >= hi = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                c <- VUM.unsafeRead acc k-                VUM.unsafeWrite acc k (combine c (VU.unsafeIndex v i))-                go (i + 1)-    go lo-    pure acc------------------------------------------------------------------------------------ Integer mean (exact integer sum + count, divided once -> order-independent)----------------------------------------------------------------------------------{- | Integer mean in ONE fused pass: a running integer sum and count per group,-divided once at finalize. The integer sum is exact, so the parallel partial-merge is byte-identical to the sequential single pass at any @-N@.--}-meanIntDirect :: VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Double-meanIntDirect g nGroups v-    | not (shouldPar n) = unsafePerformIO $ do-        (s, c) <- meanIntChunk g v nGroups 0 n-        finalizeMeanInt nGroups s c-    | otherwise = unsafePerformIO $ do-        parts <- runPartialsPairOver n capabilities (meanIntChunk g v nGroups)-        (s, c) <- mergePair nGroups parts-        finalizeMeanInt nGroups s c-  where-    !n = VU.length v-{-# NOINLINE meanIntDirect #-}--meanIntChunk ::-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    Int ->-    Int ->-    IO (VUM.IOVector Int, VUM.IOVector Int)-meanIntChunk g v nGroups lo hi = do-    s <- VUM.replicate nGroups (0 :: Int)-    c <- VUM.replicate nGroups (0 :: Int)-    let go !i-            | i >= hi = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                sv <- VUM.unsafeRead s k-                VUM.unsafeWrite s k (sv + VU.unsafeIndex v i)-                cv <- VUM.unsafeRead c k-                VUM.unsafeWrite c k (cv + 1)-                go (i + 1)-    go lo-    pure (s, c)--finalizeMeanInt ::-    Int -> VUM.IOVector Int -> VUM.IOVector Int -> IO (VU.Vector Double)-finalizeMeanInt nGroups s c = do-    out <- VUM.new nGroups-    let go !k-            | k >= nGroups = pure ()-            | otherwise = do-                sv <- VUM.unsafeRead s k-                cv <- VUM.unsafeRead c k-                VUM.unsafeWrite-                    out-                    k-                    (if cv == 0 then 0 / 0 else fromIntegral sv / fromIntegral cv)-                go (k + 1)-    go 0-    VU.unsafeFreeze out------------------------------------------------------------------------------------ Partial accumulation + merge----------------------------------------------------------------------------------mergeIntSum :: Int -> [VUM.IOVector Int] -> IO (VU.Vector Int)-mergeIntSum nGroups parts = case parts of-    [] -> VU.unsafeFreeze =<< VUM.replicate nGroups 0-    (p0 : rest) -> do-        let add !p = do-                let go !k-                        | k >= nGroups = pure ()-                        | otherwise = do-                            a <- VUM.unsafeRead p0 k-                            b <- VUM.unsafeRead p k-                            VUM.unsafeWrite p0 k (a + b)-                            go (k + 1)-                go 0-        mapM_ add rest-        VU.unsafeFreeze p0--{- | Merge per-worker (sum, count) partials into the first worker's pair by-exact integer addition; returns the accumulated pair for finalize.--}-mergePair ::-    Int ->-    [(VUM.IOVector Int, VUM.IOVector Int)] ->-    IO (VUM.IOVector Int, VUM.IOVector Int)-mergePair nGroups parts = case parts of-    [] -> (,) <$> VUM.replicate nGroups 0 <*> VUM.replicate nGroups 0-    ((s0, c0) : rest) -> do-        let add (s, c) = do-                let go !k-                        | k >= nGroups = pure ()-                        | otherwise = do-                            sa <- VUM.unsafeRead s0 k-                            sb <- VUM.unsafeRead s k-                            VUM.unsafeWrite s0 k (sa + sb)-                            ca <- VUM.unsafeRead c0 k-                            cb <- VUM.unsafeRead c k-                            VUM.unsafeWrite c0 k (ca + cb)-                            go (k + 1)-                go 0-        mapM_ add rest-        pure (s0, c0)--mergeExtremaInt :: Bool -> Int -> [VUM.IOVector Int] -> IO (VU.Vector Int)-mergeExtremaInt isMin nGroups parts = case parts of-    [] ->-        VU.unsafeFreeze =<< VUM.replicate nGroups (if isMin then maxBound else minBound)-    (p0 : rest) -> do-        let combine a b = if isMin then min a b else max a b-            add !p = do-                let go !k-                        | k >= nGroups = pure ()-                        | otherwise = do-                            a <- VUM.unsafeRead p0 k-                            b <- VUM.unsafeRead p k-                            VUM.unsafeWrite p0 k (combine a b)-                            go (k + 1)-                go 0-        mapM_ add rest-        VU.unsafeFreeze p0

@@ -1,391 +0,0 @@-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE ExplicitNamespaces #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeApplications #-}---- | Parallel scatter-accumulate aggregation kernel.-module DataFrame.Internal.AggKernelPar (-    scatterReducePar,-    momentScatterPar,-) where--import Control.Concurrent (forkIO, getNumCapabilities)-import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)-import Control.Exception (SomeException, throwIO, try)-import Control.Monad (when)-import Data.Type.Equality (TestEquality (..), type (:~:) (Refl))-import qualified Data.Vector.Unboxed as VU-import qualified Data.Vector.Unboxed.Mutable as VUM-import System.IO.Unsafe (unsafePerformIO)-import Type.Reflection (typeRep)--import DataFrame.Internal.AggKernel (-    Reduction (..),-    scatterColumnToDouble,-    scatterReduce,- )-import DataFrame.Internal.AggPlan (Moments (..), momentScatter)-import DataFrame.Internal.Column (-    Column (..),-    Columnable,-    fromUnboxedVector,-    materializePacked,- )--parThreshold :: Int-parThreshold = 200000--capabilities :: Int-capabilities = unsafePerformIO getNumCapabilities-{-# NOINLINE capabilities #-}---- | Whether to take the parallel path at this row count.-shouldPar :: Int -> Bool-shouldPar n = n >= parThreshold && capabilities > 1--groupRangeBounds :: VU.Vector Int -> Int -> Int -> VU.Vector Int-groupRangeBounds offs nGroups caps = VU.create $ do-    b <- VUM.new (caps + 1)-    let !nRows = VU.unsafeIndex offs nGroups-        !per = max 1 ((nRows + caps - 1) `div` caps)-        adv !target !gg-            | gg >= nGroups = nGroups-            | VU.unsafeIndex offs gg >= target = gg-            | otherwise = adv target (gg + 1)-        go !w !prev-            | w >= caps = VUM.unsafeWrite b caps nGroups-            | otherwise = do-                let !target = min nRows (w * per)-                    !g = adv target prev-                VUM.unsafeWrite b w g-                go (w + 1) g-    VUM.unsafeWrite b 0 0-    go 1 0-    pure b--forEachRange :: VU.Vector Int -> Int -> (Int -> Int -> IO ()) -> IO ()-forEachRange bounds caps act-    | caps <= 1 = act (VU.unsafeIndex bounds 0) (VU.unsafeIndex bounds caps)-    | otherwise = do-        vars <- mapM spawn [0 .. caps - 1]-        results <- mapM takeMVar vars-        mapM_ (either (throwIO :: SomeException -> IO ()) pure) results-  where-    spawn w = do-        var <- newEmptyMVar-        let !s = VU.unsafeIndex bounds w-            !e = VU.unsafeIndex bounds (w + 1)-        _ <- forkIO (try (act s e) >>= putMVar var)-        pure var--scatterReducePar ::-    Reduction -> VU.Vector Int -> VU.Vector Int -> Int -> Column -> Maybe Column-scatterReducePar red vis offs nGroups col-    | not (shouldPar (VU.length vis)) || nGroups <= 1 =-        scatterReduce red (rtgFromVis vis offs nGroups) nGroups col-    | otherwise = case col of-        UnboxedColumn Nothing (v :: VU.Vector a) ->-            case testEquality (typeRep @a) (typeRep @Int) of-                Just Refl -> Just (reduceParTyped red vis offs nGroups v intIdent)-                Nothing -> case testEquality (typeRep @a) (typeRep @Double) of-                    Just Refl -> Just (reduceParTyped red vis offs nGroups v dblIdent)-                    Nothing -> Nothing-        p@(PackedText _ _) -> scatterReducePar red vis offs nGroups (materializePacked p)-        _ -> Nothing-{-# NOINLINE scatterReducePar #-}--rtgFromVis :: VU.Vector Int -> VU.Vector Int -> Int -> VU.Vector Int-rtgFromVis vis offs nGroups = VU.create $ do-    let n = VU.length vis-    rtg <- VUM.new (max 1 n)-    let go !g-            | g >= nGroups = pure ()-            | otherwise = do-                let !e = VU.unsafeIndex offs (g + 1)-                    inner !pos-                        | pos >= e = pure ()-                        | otherwise = do-                            VUM.unsafeWrite rtg (VU.unsafeIndex vis pos) g-                            inner (pos + 1)-                inner (VU.unsafeIndex offs g)-                go (g + 1)-    go 0-    pure rtg--data Idents a = Idents {minSeed :: !a, maxSeed :: !a}--intIdent :: Idents Int-intIdent = Idents maxBound minBound--dblIdent :: Idents Double-dblIdent = Idents (1 / 0) (negate (1 / 0))--reduceParTyped ::-    forall a.-    (Columnable a, VU.Unbox a, Num a, Ord a, Real a) =>-    Reduction ->-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    VU.Vector a ->-    Idents a ->-    Column-reduceParTyped red vis offs nGroups v idents =-    let !caps = capabilities-        !bounds = groupRangeBounds offs nGroups caps-     in case red of-            RCount -> fromUnboxedVector (unsafePerformIO (countPar vis offs nGroups caps bounds))-            RSum -> fromUnboxedVector (unsafePerformIO (sumPar vis offs nGroups v caps bounds))-            RMin ->-                fromUnboxedVector-                    (unsafePerformIO (extremaPar min (minSeed idents) vis offs nGroups v caps bounds))-            RMax ->-                fromUnboxedVector-                    (unsafePerformIO (extremaPar max (maxSeed idents) vis offs nGroups v caps bounds))-            RMean -> fromUnboxedVector (unsafePerformIO (meanPar vis offs nGroups v caps bounds))-            RVar ->-                fromUnboxedVector-                    (unsafePerformIO (varPar False vis offs nGroups v caps bounds))-            RStd ->-                fromUnboxedVector (unsafePerformIO (varPar True vis offs nGroups v caps bounds))-            RTop2Sum -> fromUnboxedVector (unsafePerformIO (top2Par vis offs nGroups v caps bounds))-{-# INLINE reduceParTyped #-}---- | Iterate the rows of groups @[gs, ge)@ in @valueIndices@/group order.-overGroups ::-    VU.Vector Int -> VU.Vector Int -> Int -> Int -> (Int -> Int -> IO ()) -> IO ()-overGroups vis offs gs ge step = grp gs-  where-    grp !g-        | g >= ge = pure ()-        | otherwise = do-            let !e = VU.unsafeIndex offs (g + 1)-                inner !pos-                    | pos >= e = pure ()-                    | otherwise = step g (VU.unsafeIndex vis pos) >> inner (pos + 1)-            inner (VU.unsafeIndex offs g)-            grp (g + 1)-{-# INLINE overGroups #-}--countPar ::-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    Int ->-    VU.Vector Int ->-    IO (VU.Vector Int)-countPar _vis offs nGroups caps bounds = do-    out <- VUM.replicate nGroups (0 :: Int)-    forEachRange bounds caps $ \gs ge ->-        let grp !g-                | g >= ge = pure ()-                | otherwise = do-                    let !c = VU.unsafeIndex offs (g + 1) - VU.unsafeIndex offs g-                    VUM.unsafeWrite out g c-                    grp (g + 1)-         in grp gs-    VU.unsafeFreeze out--sumPar ::-    (VU.Unbox a, Num a) =>-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    VU.Vector a ->-    Int ->-    VU.Vector Int ->-    IO (VU.Vector a)-sumPar vis offs nGroups v caps bounds = do-    out <- VUM.replicate nGroups 0-    forEachRange bounds caps $ \gs ge ->-        overGroups vis offs gs ge $ \g row -> do-            cur <- VUM.unsafeRead out g-            VUM.unsafeWrite out g (cur + VU.unsafeIndex v row)-    VU.unsafeFreeze out-{-# INLINE sumPar #-}--extremaPar ::-    (VU.Unbox a) =>-    (a -> a -> a) ->-    a ->-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    VU.Vector a ->-    Int ->-    VU.Vector Int ->-    IO (VU.Vector a)-extremaPar combine seed vis offs nGroups v caps bounds = do-    out <- VUM.replicate nGroups seed-    forEachRange bounds caps $ \gs ge ->-        overGroups vis offs gs ge $ \g row -> do-            cur <- VUM.unsafeRead out g-            VUM.unsafeWrite out g (combine cur (VU.unsafeIndex v row))-    VU.unsafeFreeze out-{-# INLINE extremaPar #-}--meanPar ::-    (VU.Unbox a, Real a) =>-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    VU.Vector a ->-    Int ->-    VU.Vector Int ->-    IO (VU.Vector Double)-meanPar vis offs nGroups v caps bounds = do-    s <- VUM.replicate nGroups (0 :: Double)-    cnt <- VUM.replicate nGroups (0 :: Int)-    forEachRange bounds caps $ \gs ge ->-        overGroups vis offs gs ge $ \g row -> do-            let !x = realToFrac (VU.unsafeIndex v row)-            cs <- VUM.unsafeRead s g-            VUM.unsafeWrite s g (cs + x)-            cc <- VUM.unsafeRead cnt g-            VUM.unsafeWrite cnt g (cc + 1)-    out <- VUM.new nGroups-    let fin !k-            | k >= nGroups = pure ()-            | otherwise = do-                sv <- VUM.unsafeRead s k-                c <- VUM.unsafeRead cnt k-                VUM.unsafeWrite out k (if c == 0 then 0 / 0 else sv / fromIntegral c)-                fin (k + 1)-    fin 0-    VU.unsafeFreeze out-{-# INLINE meanPar #-}--varPar ::-    (VU.Unbox a, Real a) =>-    Bool ->-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    VU.Vector a ->-    Int ->-    VU.Vector Int ->-    IO (VU.Vector Double)-varPar takeSqrt vis offs nGroups v caps bounds = do-    cnt <- VUM.replicate nGroups (0 :: Int)-    meanV <- VUM.replicate nGroups (0 :: Double)-    m2 <- VUM.replicate nGroups (0 :: Double)-    forEachRange bounds caps $ \gs ge ->-        overGroups vis offs gs ge $ \g row -> do-            let !x = realToFrac (VU.unsafeIndex v row)-            c <- VUM.unsafeRead cnt g-            mu <- VUM.unsafeRead meanV g-            mm <- VUM.unsafeRead m2 g-            let !c' = c + 1-                !delta = x - mu-                !mu' = mu + delta / fromIntegral c'-                !mm' = mm + delta * (x - mu')-            VUM.unsafeWrite cnt g c'-            VUM.unsafeWrite meanV g mu'-            VUM.unsafeWrite m2 g mm'-    out <- VUM.new nGroups-    let fin !k-            | k >= nGroups = pure ()-            | otherwise = do-                c <- VUM.unsafeRead cnt k-                mm <- VUM.unsafeRead m2 k-                let var = if c < 2 then 0 else mm / fromIntegral (c - 1)-                VUM.unsafeWrite out k (if takeSqrt then sqrt var else var)-                fin (k + 1)-    fin 0-    VU.unsafeFreeze out-{-# INLINE varPar #-}--top2Par ::-    (VU.Unbox a, Real a) =>-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    VU.Vector a ->-    Int ->-    VU.Vector Int ->-    IO (VU.Vector Double)-top2Par vis offs nGroups v caps bounds = do-    let ninf = negate (1 / 0) :: Double-    m1 <- VUM.replicate nGroups ninf-    m2 <- VUM.replicate nGroups ninf-    forEachRange bounds caps $ \gs ge ->-        overGroups vis offs gs ge $ \g row -> do-            let !x = realToFrac (VU.unsafeIndex v row)-            a1 <- VUM.unsafeRead m1 g-            if x > a1-                then do-                    VUM.unsafeWrite m1 g x-                    VUM.unsafeWrite m2 g a1-                else do-                    a2 <- VUM.unsafeRead m2 g-                    when (x > a2) (VUM.unsafeWrite m2 g x)-    out <- VUM.new nGroups-    let fin !k-            | k >= nGroups = pure ()-            | otherwise = do-                a1 <- VUM.unsafeRead m1 k-                a2 <- VUM.unsafeRead m2 k-                let sm = (if isInfinite a1 then 0 else a1) + (if isInfinite a2 then 0 else a2)-                VUM.unsafeWrite out k sm-                fin (k + 1)-    fin 0-    VU.unsafeFreeze out-{-# INLINE top2Par #-}------------------------------------------------------------------------------------ Parallel fused two-column moments (Q9)----------------------------------------------------------------------------------{- | Parallel counterpart of 'momentScatter': one fused pass over both columns,-each group's six sums accumulated within one worker's range. Byte-identical to-'momentScatter'. 'Nothing' unless both columns are non-null unboxed Int/Double.--}-momentScatterPar ::-    VU.Vector Int -> VU.Vector Int -> Int -> Column -> Column -> Maybe Moments-momentScatterPar vis offs nGroups colX colY-    | not (shouldPar (VU.length vis)) || nGroups <= 1 =-        momentScatter (rtgFromVis vis offs nGroups) nGroups colX colY-    | otherwise = do-        xs <- scatterColumnToDouble colX-        ys <- scatterColumnToDouble colY-        let !caps = capabilities-            !bounds = groupRangeBounds offs nGroups caps-        pure (unsafePerformIO (momentPar vis offs nGroups xs ys caps bounds))-{-# NOINLINE momentScatterPar #-}--momentPar ::-    VU.Vector Int ->-    VU.Vector Int ->-    Int ->-    VU.Vector Double ->-    VU.Vector Double ->-    Int ->-    VU.Vector Int ->-    IO Moments-momentPar vis offs nGroups xs ys caps bounds = do-    cnt <- VUM.replicate nGroups (0 :: Int)-    sx <- VUM.replicate nGroups (0 :: Double)-    sy <- VUM.replicate nGroups (0 :: Double)-    sxx <- VUM.replicate nGroups (0 :: Double)-    syy <- VUM.replicate nGroups (0 :: Double)-    sxy <- VUM.replicate nGroups (0 :: Double)-    let bump arr g d = VUM.unsafeRead arr g >>= \c -> VUM.unsafeWrite arr g (c + d)-    forEachRange bounds caps $ \gs ge ->-        overGroups vis offs gs ge $ \g row -> do-            let !x = VU.unsafeIndex xs row-                !y = VU.unsafeIndex ys row-            VUM.unsafeRead cnt g >>= \c -> VUM.unsafeWrite cnt g (c + 1)-            bump sx g x-            bump sy g y-            bump sxx g (x * x)-            bump syy g (y * y)-            bump sxy g (x * y)-    Moments . fromUnboxedVector-        <$> VU.unsafeFreeze cnt-        <*> (fromUnboxedVector <$> VU.unsafeFreeze sx)-        <*> (fromUnboxedVector <$> VU.unsafeFreeze sy)-        <*> (fromUnboxedVector <$> VU.unsafeFreeze sxx)-        <*> (fromUnboxedVector <$> VU.unsafeFreeze syy)-        <*> (fromUnboxedVector <$> VU.unsafeFreeze sxy)

@@ -1,299 +0,0 @@-{-# LANGUAGE AllowAmbiguousTypes #-}-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE ExplicitNamespaces #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeApplications #-}--{- | The aggregation fast-path planner and the two-column moment scatter.-'planAgg' recognises a supported aggregate shape over a clean unboxed Int/Double-column and returns an 'AggPlan'; 'momentScatter' fuses the six regression sums.--}-module DataFrame.Internal.AggPlan (-    AggPlan (..),-    planAgg,-    Moments (..),-    momentScatter,-    MomentPlan (..),-    planMoments,-) where--import qualified Data.Map.Strict as M-import qualified Data.Text as T-import Data.Type.Equality (TestEquality (..), type (:~:) (Refl))-import qualified Data.Vector.Unboxed as VU-import qualified Data.Vector.Unboxed.Mutable as VUM--import Control.Monad.ST (runST)-import DataFrame.Internal.AggKernel (Reduction (..), scatterColumnToDouble)-import DataFrame.Internal.Column (Column (..), fromUnboxedVector)-import DataFrame.Internal.DataFrame (-    DataFrame (derivingExpressions),-    GroupedDataFrame (..),-    getColumn,- )-import DataFrame.Internal.Expression (-    AggStrategy (..),-    BinaryOp (binaryCommutative, binaryName),-    Expr (..),-    UExpr (..),- )-import Type.Reflection (Typeable, typeRep)--{- | The plan 'planAgg' produces for a recognised output expression. The median-plan carries only the column name (the holistic grouped sort lives in the-operations layer, where @vector-algorithms@ is available).--}-data AggPlan-    = -- | A single scatter reduction over one named column.-      PlanScatter Reduction T.Text-    | -- | @max a - min b@ (Q7): two scatters then a vectorized combine.-      PlanMaxMinusMin T.Text T.Text-    | -- | Holistic median over one named column.-      PlanMedian T.Text--{- | Inspect a named output expression; return @Just plan@ on a recognised shape-over a present clean column, else 'Nothing'. Nullable or non-Int/Double columns-are rejected here so the scatter only sees a clean unboxed vector.--}-planAgg :: GroupedDataFrame -> UExpr -> Maybe AggPlan-planAgg gdf (UExpr (expr :: Expr a)) = case expr of-    Agg (FoldAgg tag _ _) (Col name) -> foldPlan tag name-    Agg (MergeAgg tag _ _ _ _) (Col name) -> mergePlan tag name-    Agg (CollectAgg tag _) (Col name) -> collectPlan tag name-    Binary-        op-        (Agg (FoldAgg lt Nothing _) (Col a))-        (Agg (FoldAgg rt Nothing _) (Col b)) ->-            if binaryName op == "sub" && lt == "maximum" && rt == "minimum"-                then requireBoth a b (PlanMaxMinusMin a b)-                else Nothing-    _ -> Nothing-  where-    foldPlan tag name = case tag of-        "sum" -> require name (PlanScatter RSum name)-        "minimum" -> require name (PlanScatter RMin name)-        "maximum" -> require name (PlanScatter RMax name)-        _ -> Nothing-    mergePlan tag name = case tag of-        "mean" -> outputType @Double >> require name (PlanScatter RMean name)-        "count" -> outputType @Int >> require name (PlanScatter RCount name)-        _ -> Nothing-    outputType :: forall t. (Typeable t) => Maybe ()-    outputType = case testEquality (typeRep @a) (typeRep @t) of-        Just Refl -> Just ()-        Nothing -> Nothing-    collectPlan tag name = case tag of-        "stddev" -> require name (PlanScatter RStd name)-        "variance" -> require name (PlanScatter RVar name)-        "top2Sum" -> require name (PlanScatter RTop2Sum name)-        "median" -> require name (PlanMedian name)-        _ -> Nothing-    require name plan = colUnboxedNumeric name >> Just plan-    requireBoth a b plan = colUnboxedNumeric a >> colUnboxedNumeric b >> Just plan-    colUnboxedNumeric name = case getColumn name (fullDataframe gdf) of-        Just c | isUnboxedNumeric c -> Just ()-        _ -> Nothing---- | The matcher only fires on non-null unboxed Int/Double columns.-isUnboxedNumeric :: Column -> Bool-isUnboxedNumeric = \case-    UnboxedColumn Nothing (_ :: VU.Vector a) ->-        case testEquality (typeRep @a) (typeRep @Int) of-            Just Refl -> True-            Nothing -> case testEquality (typeRep @a) (typeRep @Double) of-                Just Refl -> True-                Nothing -> False-    _ -> False--{- | A recognised moment (Q9 regression) aggregate group: six output columns that-form the sufficient statistics of two base columns @x@ and @y@. The caller runs-'momentScatter' once and binds each output name to a field of the result.--}-data MomentPlan = MomentPlan-    { mpColX :: T.Text-    , mpColY :: T.Text-    , mpNName :: T.Text-    , mpSxName :: T.Text-    , mpSyName :: T.Text-    , mpSxxName :: T.Text-    , mpSyyName :: T.Text-    , mpSxyName :: T.Text-    }--{- | The shape of a sum's argument once unary coercions are peeled and derived-columns are resolved through @derivingExpressions@: either linear in one base-column or the product of two base columns (sorted).--}-data Term-    = Lin T.Text-    | Prod T.Text T.Text-    deriving (Eq, Ord, Show)--{- | Recognise the moment shape across a whole @aggregate@ list: exactly-@count@, @sum(x)@, @sum(y)@, @sum(x*x)@, @sum(y*y)@, @sum(x*y)@ over two distinct-clean unboxed base columns. 'Nothing' on any other set.--}-planMoments :: GroupedDataFrame -> [(T.Text, UExpr)] -> Maybe MomentPlan-planMoments gdf aggs-    | length aggs /= 6 = Nothing-    | otherwise = do-        let exprs = derivingExpressions (fullDataframe gdf)-        roles <- traverse (classify exprs) aggs-        let names = M.fromList [(r, nm) | (nm, r) <- roles]-        nName <- M.lookup RoleN names-        (x, y) <- pickBaseColumns roles-        sxName <- M.lookup (RoleLin x) names-        syName <- M.lookup (RoleLin y) names-        sxxName <- M.lookup (RoleProd x x) names-        syyName <- M.lookup (RoleProd y y) names-        sxyName <- M.lookup (RoleProd x y) names-        _ <- if x /= y then Just () else Nothing-        _ <- colUnboxedNumeric x-        _ <- colUnboxedNumeric y-        pure-            MomentPlan-                { mpColX = x-                , mpColY = y-                , mpNName = nName-                , mpSxName = sxName-                , mpSyName = syName-                , mpSxxName = sxxName-                , mpSyyName = syyName-                , mpSxyName = sxyName-                }-  where-    colUnboxedNumeric name = case getColumn name (fullDataframe gdf) of-        Just c | isUnboxedNumeric c -> Just ()-        _ -> Nothing---- | The output role each named aggregation plays in the moment shape.-data Role-    = RoleN-    | RoleLin T.Text-    | RoleProd T.Text T.Text-    deriving (Eq, Ord, Show)---- | Tag a single named aggregation with its moment role, or reject the group.-classify :: M.Map T.Text UExpr -> (T.Text, UExpr) -> Maybe (T.Text, Role)-classify exprs (name, UExpr expr) = case expr of-    Agg (MergeAgg "count" _ _ _ _) _ -> Just (name, RoleN)-    Agg (FoldAgg "sum" _ _) arg -> (\t -> (name, termRole t)) <$> resolveTerm exprs (UExpr arg)-    _ -> Nothing--termRole :: Term -> Role-termRole (Lin a) = RoleLin a-termRole (Prod a b) = RoleProd a b--{- | Resolve a (sum-argument) expression to its 'Term'. Peels @toDouble@-style-unary coercions, follows a derived column to its stored expression, and-recognises a commutative product of two linear terms.--}-resolveTerm :: M.Map T.Text UExpr -> UExpr -> Maybe Term-resolveTerm exprs = go (8 :: Int)-  where-    go 0 _ = Nothing-    go fuel (UExpr e) = case e of-        Col nm -> case M.lookup nm exprs of-            Just ue -> go (fuel - 1) ue-            Nothing -> Just (Lin nm)-        Unary _ inner -> go (fuel - 1) (UExpr inner)-        Binary op l r-            | binaryName op == "mult" && binaryCommutative op -> do-                Lin a <- go (fuel - 1) (UExpr l)-                Lin b <- go (fuel - 1) (UExpr r)-                Just (sortProd a b)-        _ -> Nothing---- | Products are unordered: store the pair sorted so @x*y@ and @y*x@ unify.-sortProd :: T.Text -> T.Text -> Term-sortProd a b-    | a <= b = Prod a b-    | otherwise = Prod b a--{- | From the classified roles, find the unordered pair of base columns that the-linear sums name. There must be exactly two distinct linear-sum columns.--}-pickBaseColumns :: [(T.Text, Role)] -> Maybe (T.Text, T.Text)-pickBaseColumns roles =-    case lins of-        [a, b] | a /= b -> Just (a, b)-        _ -> Nothing-  where-    lins = M.keys (M.fromList [(c, ()) | (_, RoleLin c) <- roles])--{- | The additive moment sums of two columns, each an @nGroups@-length column:-@(n, Sx, Sy, Sxx, Syy, Sxy)@.--}-data Moments = Moments-    { mN :: Column-    , mSx :: Column-    , mSy :: Column-    , mSxx :: Column-    , mSyy :: Column-    , mSxy :: Column-    }--{- | One pass over two Double-coercible columns @x@ and @y@ filling the count and-five sums, collapsing the Q9 regression family's six folds into a single pass.-'Nothing' unless both columns are non-null unboxed Int/Double.--}-momentScatter :: VU.Vector Int -> Int -> Column -> Column -> Maybe Moments-momentScatter g nGroups colX colY = do-    xs <- scatterColumnToDouble colX-    ys <- scatterColumnToDouble colY-    let (cnt, sx, sy, sxx, syy, sxy) = momentPass g nGroups xs ys-    pure-        Moments-            { mN = fromUnboxedVector cnt-            , mSx = fromUnboxedVector sx-            , mSy = fromUnboxedVector sy-            , mSxx = fromUnboxedVector sxx-            , mSyy = fromUnboxedVector syy-            , mSxy = fromUnboxedVector sxy-            }--momentPass ::-    VU.Vector Int ->-    Int ->-    VU.Vector Double ->-    VU.Vector Double ->-    ( VU.Vector Int-    , VU.Vector Double-    , VU.Vector Double-    , VU.Vector Double-    , VU.Vector Double-    , VU.Vector Double-    )-momentPass g nGroups xs ys = runST $ do-    cnt <- VUM.replicate nGroups (0 :: Int)-    sx <- VUM.replicate nGroups (0 :: Double)-    sy <- VUM.replicate nGroups (0 :: Double)-    sxx <- VUM.replicate nGroups (0 :: Double)-    syy <- VUM.replicate nGroups (0 :: Double)-    sxy <- VUM.replicate nGroups (0 :: Double)-    let n = VU.length xs-        bump arr k d = VUM.unsafeRead arr k >>= \c -> VUM.unsafeWrite arr k (c + d)-        go !i-            | i >= n = pure ()-            | otherwise = do-                let !k = VU.unsafeIndex g i-                    !x = VU.unsafeIndex xs i-                    !y = VU.unsafeIndex ys i-                VUM.unsafeRead cnt k >>= \c -> VUM.unsafeWrite cnt k (c + 1)-                bump sx k x-                bump sy k y-                bump sxx k (x * x)-                bump syy k (y * y)-                bump sxy k (x * y)-                go (i + 1)-    go 0-    (,,,,,)-        <$> VU.unsafeFreeze cnt-        <*> VU.unsafeFreeze sx-        <*> VU.unsafeFreeze sy-        <*> VU.unsafeFreeze sxx-        <*> VU.unsafeFreeze syy-        <*> VU.unsafeFreeze sxy

@@ -0,0 +1,832 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ExplicitNamespaces #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++{- | The low-cardinality DENSE reduction kernel: when the group domain is small,+the grouping layer's @rowToGroup@ already maps row -> group, so the reduction+scatters straight off it with no @valueIndices@ gather.++Parallel by ROW range with a private per-worker accumulator of @nGroups@ slots,+merged afterwards — which is why it needs a small domain, and why it admits only+order-independent reductions: the merge must be exact for the result to stay+byte-identical to @-N1@. Anything it rejects falls back to+"DataFrame.Internal.Aggregation.Kernel.Scatter".++The caller decides whether the domain is small enough; see @denseThreshold@ in+the operations layer.+-}+module DataFrame.Internal.Aggregation.Kernel.Dense (+    denseReduce,+    denseMaxMinusMin,+) where++import Control.Monad (when)+import Data.Type.Equality (TestEquality (..), type (:~:) (Refl))+import qualified Data.Vector.Unboxed as VU+import qualified Data.Vector.Unboxed.Mutable as VUM+import System.IO.Unsafe (unsafePerformIO)+import Type.Reflection (typeRep)++import DataFrame.Internal.Aggregation.Reduction (Reduction (..))+import DataFrame.Internal.Column (+    Column (..),+    fromUnboxedVector,+    materializePacked,+ )+import DataFrame.Internal.Control.Concurrent (+    capabilities,+    parThreshold,+    parallelChunks,+    shouldParallelize,+ )++{- | Group count at or below which the float (Double sum/mean) direct+reductions run as ONE sequential row-order pass: the accumulators fit in L2, the+pass is memory-bandwidth bound, and adding each value to its group in ascending+row order is exactly the order the group-range gather kernel uses (the grouping+layer's @valueIndices@ is a stable counting sort), so the result is+byte-identical to it. Above this the parallel chunked variant runs instead (see+'sumDblDense'). Var/std are never taken directly: any chunked merge of+variance state changes the float recurrence, and the group-range gather kernel+already runs them in parallel while replaying the interpreter's per-group+update order bit-for-bit.+-}+seqFloatGroups :: Int+seqFloatGroups = 65536++{- | Run a recognised reduction through the direct-indexed path. 'Nothing' (so+the caller falls back to the order-preserving kernel) unless the reduction is+admitted at this element type AND the column is a clean unboxed Int/Double.+-}+denseReduce :: Reduction -> VU.Vector Int -> Int -> Column -> Maybe Column+denseReduce red g nGroups col = case col of+    UnboxedColumn Nothing (v :: VU.Vector a) ->+        case testEquality (typeRep @a) (typeRep @Int) of+            Just Refl -> denseInt red g nGroups v+            Nothing -> case testEquality (typeRep @a) (typeRep @Double) of+                Just Refl -> denseDouble red g nGroups v+                Nothing -> Nothing+    p@(PackedText _ _) -> denseReduce red g nGroups (materializePacked p)+    _ -> Nothing+{-# INLINEABLE denseReduce #-}++{- | The reductions admitted over an Int column. Sum/min/max/mean/count are+exact in the Int domain (any merge order gives the same bits); top2sum selects+the two largest values (order-independent as a multiset selection) and only+adds them once at finalize. Var/std stay with the group-range gather kernel:+its per-group Welford recurrence replays the interpreter's update order+bit-for-bit, which no chunk-merged direct pass can.+-}+denseInt :: Reduction -> VU.Vector Int -> Int -> VU.Vector Int -> Maybe Column+denseInt red g nGroups v = case red of+    RCount -> Just (fromUnboxedVector (countDense g nGroups (VU.length v)))+    RSum -> Just (fromUnboxedVector (sumIntDense g nGroups v))+    RMin -> Just (fromUnboxedVector (extremaIntDense True g nGroups v))+    RMax -> Just (fromUnboxedVector (extremaIntDense False g nGroups v))+    RMean -> Just (fromUnboxedVector (meanIntDense g nGroups v))+    RTop2Sum -> Just (fromUnboxedVector (top2Dense g nGroups v))+    RTop2Snd -> Just (fromUnboxedVector (top2SndDense g nGroups v))+    _ -> Nothing++{- | The reductions admitted over a Double column. Count/min/max/top2sum are+order-independent (exact per-worker merge, byte-identical at any @-N@). The+float sum/mean run sequentially in row order below 'seqFloatGroups' (matching+the gather kernel's per-group addition order exactly) and as deterministic+chunked partials above it. Var/std keep the gather kernel (see 'denseInt').+-}+denseDouble ::+    Reduction -> VU.Vector Int -> Int -> VU.Vector Double -> Maybe Column+denseDouble red g nGroups v = case red of+    RCount -> Just (fromUnboxedVector (countDense g nGroups (VU.length v)))+    RSum -> Just (fromUnboxedVector (sumDblDense g nGroups v))+    RMean -> Just (fromUnboxedVector (meanDblDense g nGroups v))+    RMin -> Just (fromUnboxedVector (extremaDblDense True g nGroups v))+    RMax -> Just (fromUnboxedVector (extremaDblDense False g nGroups v))+    RTop2Sum -> Just (fromUnboxedVector (top2Dense g nGroups v))+    RTop2Snd -> Just (fromUnboxedVector (top2SndDense g nGroups v))+    _ -> Nothing++{- | The fused @max a - min b@ direct pass: BOTH extrema accumulate in one+streaming loop over the rows (min/max are order-independent, so the per-worker+merge is exact and the result byte-identical to the two gather passes it+replaces). 'Nothing' unless both columns are clean unboxed and same-typed+(Int/Int keeps the Int result of the interpreter; Double/Double the Double one);+mixed pairs keep the gather fallback.+-}+denseMaxMinusMin :: VU.Vector Int -> Int -> Column -> Column -> Maybe Column+denseMaxMinusMin g nGroups ca cb = case (ca, cb) of+    ( UnboxedColumn Nothing (va :: VU.Vector x)+        , UnboxedColumn Nothing (vb :: VU.Vector y)+        )+            | Just Refl <- testEquality (typeRep @x) (typeRep @Int)+            , Just Refl <- testEquality (typeRep @y) (typeRep @Int) ->+                Just (fromUnboxedVector (maxMinusMinDenseInt g nGroups va vb))+            | Just Refl <- testEquality (typeRep @x) (typeRep @Double)+            , Just Refl <- testEquality (typeRep @y) (typeRep @Double) ->+                Just (fromUnboxedVector (maxMinusMinDenseDbl g nGroups va vb))+    _ -> Nothing+{-# INLINEABLE denseMaxMinusMin #-}++{- | Monomorphic entry points: the 'testEquality' dispatch above only yields an+unsafe coercion, so a direct call to the polymorphic 'maxMinusMinDense' there+would stay at the abstract element type and never meet its SPECIALIZE rules+(measured ~3x on the whole pass); calling through these fixed-type wrappers+(the coercion lands on the argument) does.+-}+maxMinusMinDenseInt ::+    VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Int -> VU.Vector Int+maxMinusMinDenseInt = maxMinusMinDense minBound maxBound+{-# NOINLINE maxMinusMinDenseInt #-}++maxMinusMinDenseDbl ::+    VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Double -> VU.Vector Double+maxMinusMinDenseDbl = maxMinusMinDense (negate (1 / 0)) (1 / 0)+{-# NOINLINE maxMinusMinDenseDbl #-}++-- | Whether to fan out at this row count.+shouldPar :: Int -> Bool+shouldPar = shouldParallelize parThreshold++{- | Fork @caps@ workers over disjoint contiguous row ranges of @[0, n)@, each+producing its own private accumulator (no shared array, no sync). Returns the+partials in worker order for the caller's merge; rethrows the first failure.+The chunking is a fixed function of @n@ and @caps@, so any merge over the+partials is deterministic at a given @-N@.+-}+runPartialsOver ::+    Int -> Int -> (Int -> Int -> IO acc) -> IO [acc]+runPartialsOver n _caps = parallelChunks parThreshold n++-------------------------------------------------------------------------------+-- Count (order-independent: per-group row count)+-------------------------------------------------------------------------------++countDense :: VU.Vector Int -> Int -> Int -> VU.Vector Int+countDense g nGroups n = unsafePerformIO $ do+    parts <- runPartialsOver n capabilities (countChunk g nGroups)+    mergeIntSum nGroups parts+{-# NOINLINE countDense #-}++countChunk :: VU.Vector Int -> Int -> Int -> Int -> IO (VUM.IOVector Int)+countChunk g nGroups lo hi = do+    acc <- VUM.replicate nGroups (0 :: Int)+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                c <- VUM.unsafeRead acc k+                VUM.unsafeWrite acc k (c + 1)+                go (i + 1)+    go lo+    pure acc++-------------------------------------------------------------------------------+-- Integer sum (exact: merge order irrelevant)+-------------------------------------------------------------------------------++sumIntDense :: VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Int+sumIntDense g nGroups v = unsafePerformIO $ do+    parts <- runPartialsOver (VU.length v) capabilities (sumIntChunk g v nGroups)+    mergeIntSum nGroups parts+{-# NOINLINE sumIntDense #-}++sumIntChunk ::+    VU.Vector Int -> VU.Vector Int -> Int -> Int -> Int -> IO (VUM.IOVector Int)+sumIntChunk g v nGroups lo hi = do+    acc <- VUM.replicate nGroups (0 :: Int)+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                c <- VUM.unsafeRead acc k+                VUM.unsafeWrite acc k (c + VU.unsafeIndex v i)+                go (i + 1)+    go lo+    pure acc++-------------------------------------------------------------------------------+-- Double sum / mean (streaming scatter; chunked partials above seqFloatGroups)+-------------------------------------------------------------------------------++{- | Double group sums. At or below 'seqFloatGroups' a single sequential pass in+ascending row order — each group's additions happen in exactly the order the+group-range gather kernel performs them, so the result is byte-identical to it.+Above that, per-worker chunk partials merged in worker order: still+deterministic at a fixed @-N@, but the float summation order differs from the+sequential pass.+-}+sumDblDense :: VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Double+sumDblDense g nGroups v+    | nGroups <= seqFloatGroups || not (shouldPar n) =+        unsafePerformIO (sumDblChunk g v nGroups 0 n >>= VU.unsafeFreeze)+    | otherwise = unsafePerformIO $ do+        parts <- runPartialsOver n capabilities (sumDblChunk g v nGroups)+        mergeDblSum nGroups parts+  where+    !n = VU.length v+{-# NOINLINE sumDblDense #-}++sumDblChunk ::+    VU.Vector Int ->+    VU.Vector Double ->+    Int ->+    Int ->+    Int ->+    IO (VUM.IOVector Double)+sumDblChunk g v nGroups lo hi = do+    acc <- VUM.replicate nGroups (0 :: Double)+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                c <- VUM.unsafeRead acc k+                VUM.unsafeWrite acc k (c + VU.unsafeIndex v i)+                go (i + 1)+    go lo+    pure acc++{- | Double mean: fused (Double sum, count) per group, divided once at finalize.+Same order policy as 'sumDblDense' (sequential row order is byte-identical to+the gather kernel; the chunked variant changes the float summation order).+-}+meanDblDense :: VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Double+meanDblDense g nGroups v+    | nGroups <= seqFloatGroups || not (shouldPar n) = unsafePerformIO $ do+        (s, c) <- meanDblChunk g v nGroups 0 n+        finalizeMeanDbl nGroups s c+    | otherwise = unsafePerformIO $ do+        parts <- runPartialsOver n capabilities (meanDblChunk g v nGroups)+        (s, c) <- mergeMeanDbl nGroups parts+        finalizeMeanDbl nGroups s c+  where+    !n = VU.length v+{-# NOINLINE meanDblDense #-}++meanDblChunk ::+    VU.Vector Int ->+    VU.Vector Double ->+    Int ->+    Int ->+    Int ->+    IO (VUM.IOVector Double, VUM.IOVector Int)+meanDblChunk g v nGroups lo hi = do+    s <- VUM.replicate nGroups (0 :: Double)+    c <- VUM.replicate nGroups (0 :: Int)+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                sv <- VUM.unsafeRead s k+                VUM.unsafeWrite s k (sv + VU.unsafeIndex v i)+                cv <- VUM.unsafeRead c k+                VUM.unsafeWrite c k (cv + 1)+                go (i + 1)+    go lo+    pure (s, c)++finalizeMeanDbl ::+    Int -> VUM.IOVector Double -> VUM.IOVector Int -> IO (VU.Vector Double)+finalizeMeanDbl nGroups s c = do+    out <- VUM.new nGroups+    let go !k+            | k >= nGroups = pure ()+            | otherwise = do+                sv <- VUM.unsafeRead s k+                cv <- VUM.unsafeRead c k+                VUM.unsafeWrite+                    out+                    k+                    (if cv == 0 then 0 / 0 else sv / fromIntegral cv)+                go (k + 1)+    go 0+    VU.unsafeFreeze out++-------------------------------------------------------------------------------+-- Integer min / max (order-independent)+-------------------------------------------------------------------------------++extremaIntDense ::+    Bool -> VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Int+extremaIntDense isMin g nGroups v = unsafePerformIO $ do+    parts <-+        runPartialsOver (VU.length v) capabilities (extremaIntChunk isMin g v nGroups)+    mergeExtremaInt isMin nGroups parts+{-# NOINLINE extremaIntDense #-}++extremaIntChunk ::+    Bool ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    Int ->+    Int ->+    IO (VUM.IOVector Int)+extremaIntChunk isMin g v nGroups lo hi = do+    let !seed = if isMin then maxBound else minBound+        combine a b = if isMin then min a b else max a b+    acc <- VUM.replicate nGroups seed+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                c <- VUM.unsafeRead acc k+                VUM.unsafeWrite acc k (combine c (VU.unsafeIndex v i))+                go (i + 1)+    go lo+    pure acc++-------------------------------------------------------------------------------+-- Double min / max (order-independent: exact per-worker merge)+-------------------------------------------------------------------------------++extremaDblDense ::+    Bool -> VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Double+extremaDblDense isMin g nGroups v+    | not (shouldPar n) =+        unsafePerformIO (extremaDblChunk isMin g v nGroups 0 n >>= VU.unsafeFreeze)+    | otherwise = unsafePerformIO $ do+        parts <- runPartialsOver n capabilities (extremaDblChunk isMin g v nGroups)+        mergeExtremaDbl isMin nGroups parts+  where+    !n = VU.length v+{-# NOINLINE extremaDblDense #-}++extremaDblChunk ::+    Bool ->+    VU.Vector Int ->+    VU.Vector Double ->+    Int ->+    Int ->+    Int ->+    IO (VUM.IOVector Double)+extremaDblChunk isMin g v nGroups lo hi = do+    let !seed = if isMin then 1 / 0 else negate (1 / 0)+        combine a b = if isMin then min a b else max a b+    acc <- VUM.replicate nGroups seed+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                c <- VUM.unsafeRead acc k+                VUM.unsafeWrite acc k (combine c (VU.unsafeIndex v i))+                go (i + 1)+    go lo+    pure acc++-------------------------------------------------------------------------------+-- Fused max(a) - min(b) (order-independent: exact per-worker merge)+-------------------------------------------------------------------------------++{- | One streaming pass accumulating @max a@ and @min b@ together. @maxSeed@ is+the identity of @max@ (the type's least value), @minSeed@ of @min@ (its+greatest).+-}+maxMinusMinDense ::+    (VU.Unbox a, Num a, Ord a) =>+    a ->+    a ->+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    VU.Vector a ->+    VU.Vector a+{-# SPECIALIZE maxMinusMinDense ::+    Int ->+    Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    VU.Vector Int+    #-}+{-# SPECIALIZE maxMinusMinDense ::+    Double ->+    Double ->+    VU.Vector Int ->+    Int ->+    VU.Vector Double ->+    VU.Vector Double ->+    VU.Vector Double+    #-}+maxMinusMinDense maxSeed minSeed g nGroups va vb+    | not (shouldPar n) = unsafePerformIO $ do+        (mx, mn) <- maxMinusMinChunk maxSeed minSeed g va vb nGroups 0 n+        finalizeMaxMinusMin nGroups mx mn+    | otherwise = unsafePerformIO $ do+        parts <-+            runPartialsOver+                n+                capabilities+                (maxMinusMinChunk maxSeed minSeed g va vb nGroups)+        (mx, mn) <- mergeMaxMin nGroups parts+        finalizeMaxMinusMin nGroups mx mn+  where+    !n = VU.length va+{- INLINEABLE (not NOINLINE) so the SPECIALIZE pragmas above take effect; the+kernel is a pure function of its arguments, so the usual unsafePerformIO+sharing concern does not apply. -}+{-# INLINEABLE maxMinusMinDense #-}++maxMinusMinChunk ::+    (VU.Unbox a, Ord a) =>+    a ->+    a ->+    VU.Vector Int ->+    VU.Vector a ->+    VU.Vector a ->+    Int ->+    Int ->+    Int ->+    IO (VUM.IOVector a, VUM.IOVector a)+maxMinusMinChunk maxSeed minSeed g va vb nGroups lo hi = do+    mx <- VUM.replicate nGroups maxSeed+    mn <- VUM.replicate nGroups minSeed+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                cx <- VUM.unsafeRead mx k+                VUM.unsafeWrite mx k (max cx (VU.unsafeIndex va i))+                cn <- VUM.unsafeRead mn k+                VUM.unsafeWrite mn k (min cn (VU.unsafeIndex vb i))+                go (i + 1)+    go lo+    pure (mx, mn)++mergeMaxMin ::+    (VU.Unbox a, Ord a) =>+    Int ->+    [(VUM.IOVector a, VUM.IOVector a)] ->+    IO (VUM.IOVector a, VUM.IOVector a)+mergeMaxMin nGroups parts = case parts of+    [] -> error "mergeMaxMin: no partials"+    ((mx0, mn0) : rest) -> do+        let add (mx, mn) = do+                let go !k+                        | k >= nGroups = pure ()+                        | otherwise = do+                            xa <- VUM.unsafeRead mx0 k+                            xb <- VUM.unsafeRead mx k+                            VUM.unsafeWrite mx0 k (max xa xb)+                            na <- VUM.unsafeRead mn0 k+                            nb <- VUM.unsafeRead mn k+                            VUM.unsafeWrite mn0 k (min na nb)+                            go (k + 1)+                go 0+        mapM_ add rest+        pure (mx0, mn0)++finalizeMaxMinusMin ::+    (VU.Unbox a, Num a) =>+    Int ->+    VUM.IOVector a ->+    VUM.IOVector a ->+    IO (VU.Vector a)+finalizeMaxMinusMin nGroups mx mn = do+    out <- VUM.new nGroups+    let go !k+            | k >= nGroups = pure ()+            | otherwise = do+                a <- VUM.unsafeRead mx k+                b <- VUM.unsafeRead mn k+                VUM.unsafeWrite out k (a - b)+                go (k + 1)+    go 0+    VU.unsafeFreeze out++-------------------------------------------------------------------------------+-- Top-2 sum (order-independent multiset selection; one float add at finalize)+-------------------------------------------------------------------------------++{- | Sum of the two largest values per group. Each accumulator holds the+(largest, second-largest) pair seen so far; merging two pairs keeps the top two+of the four candidates. No float ADDITION happens until the single @m1 + m2@ at+finalize, so the result is byte-identical to the gather kernel regardless of+chunking. Mirrors the gather kernel exactly, including the @realToFrac@ per+element and the @-inf -> 0@ guards at finalize.+-}+top2Dense ::+    (VU.Unbox a, Real a) => VU.Vector Int -> Int -> VU.Vector a -> VU.Vector Double+{-# SPECIALIZE top2Dense ::+    VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Double+    #-}+{-# SPECIALIZE top2Dense ::+    VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Double+    #-}+top2Dense g nGroups v+    | not (shouldPar n) = unsafePerformIO $ do+        (m1, m2) <- top2Chunk g v nGroups 0 n+        finalizeTop2 nGroups m1 m2+    | otherwise = unsafePerformIO $ do+        parts <- runPartialsOver n capabilities (top2Chunk g v nGroups)+        (m1, m2) <- mergeTop2 nGroups parts+        finalizeTop2 nGroups m1 m2+  where+    !n = VU.length v+{- INLINEABLE (not NOINLINE) so the SPECIALIZE pragmas above take effect; pure+function of its arguments, so unsafePerformIO sharing is not a concern. -}+{-# INLINEABLE top2Dense #-}++{- | Second-largest value per group: the exact same per-worker+(largest, second-largest) accumulator and merge as 'top2Dense'+('top2Chunk'/'mergeTop2'), finalized to the second max alone. A group of+size < 2 finalizes its @-inf@ seed to NaN — documented behaviour (the+db-benchmark Q8 data has no size-1 @id6@ groups). Order-independent multiset+selection, so byte-identical at any @-N@.+-}+top2SndDense ::+    (VU.Unbox a, Real a) => VU.Vector Int -> Int -> VU.Vector a -> VU.Vector Double+{-# SPECIALIZE top2SndDense ::+    VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Double+    #-}+{-# SPECIALIZE top2SndDense ::+    VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Double+    #-}+top2SndDense g nGroups v+    | not (shouldPar n) = unsafePerformIO $ do+        (m1, m2) <- top2Chunk g v nGroups 0 n+        finalizeTop2Snd nGroups m1 m2+    | otherwise = unsafePerformIO $ do+        parts <- runPartialsOver n capabilities (top2Chunk g v nGroups)+        (m1, m2) <- mergeTop2 nGroups parts+        finalizeTop2Snd nGroups m1 m2+  where+    !n = VU.length v+{- INLINEABLE (not NOINLINE) so the SPECIALIZE pragmas above take effect; pure+function of its arguments, so unsafePerformIO sharing is not a concern. -}+{-# INLINEABLE top2SndDense #-}++top2Chunk ::+    (VU.Unbox a, Real a) =>+    VU.Vector Int ->+    VU.Vector a ->+    Int ->+    Int ->+    Int ->+    IO (VUM.IOVector Double, VUM.IOVector Double)+top2Chunk g v nGroups lo hi = do+    let ninf = negate (1 / 0) :: Double+    m1 <- VUM.replicate nGroups ninf+    m2 <- VUM.replicate nGroups ninf+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                    !x = realToFrac (VU.unsafeIndex v i)+                a1 <- VUM.unsafeRead m1 k+                if x > a1+                    then do+                        VUM.unsafeWrite m1 k x+                        VUM.unsafeWrite m2 k a1+                    else do+                        a2 <- VUM.unsafeRead m2 k+                        when (x > a2) (VUM.unsafeWrite m2 k x)+                go (i + 1)+    go lo+    pure (m1, m2)++{- | Top two of the four candidates @{a1, a2, b1, b2}@ per group (each pair+already ordered @m1 >= m2@, @-inf@ seeds included).+-}+mergeTop2 ::+    Int ->+    [(VUM.IOVector Double, VUM.IOVector Double)] ->+    IO (VUM.IOVector Double, VUM.IOVector Double)+mergeTop2 nGroups parts = case parts of+    [] -> error "mergeTop2: no partials"+    ((m10, m20) : rest) -> do+        let add (m1, m2) = do+                let go !k+                        | k >= nGroups = pure ()+                        | otherwise = do+                            a1 <- VUM.unsafeRead m10 k+                            a2 <- VUM.unsafeRead m20 k+                            b1 <- VUM.unsafeRead m1 k+                            b2 <- VUM.unsafeRead m2 k+                            if b1 > a1+                                then do+                                    VUM.unsafeWrite m10 k b1+                                    VUM.unsafeWrite m20 k (max a1 b2)+                                else VUM.unsafeWrite m20 k (max a2 b1)+                            go (k + 1)+                go 0+        mapM_ add rest+        pure (m10, m20)++finalizeTop2 ::+    Int -> VUM.IOVector Double -> VUM.IOVector Double -> IO (VU.Vector Double)+finalizeTop2 nGroups m1 m2 = do+    out <- VUM.new nGroups+    let go !k+            | k >= nGroups = pure ()+            | otherwise = do+                a1 <- VUM.unsafeRead m1 k+                a2 <- VUM.unsafeRead m2 k+                let s = (if isInfinite a1 then 0 else a1) + (if isInfinite a2 then 0 else a2)+                VUM.unsafeWrite out k s+                go (k + 1)+    go 0+    VU.unsafeFreeze out++finalizeTop2Snd ::+    Int -> VUM.IOVector Double -> VUM.IOVector Double -> IO (VU.Vector Double)+finalizeTop2Snd nGroups _m1 m2 = do+    out <- VUM.new nGroups+    let go !k+            | k >= nGroups = pure ()+            | otherwise = do+                a2 <- VUM.unsafeRead m2 k+                VUM.unsafeWrite out k (if isInfinite a2 then 0 / 0 else a2)+                go (k + 1)+    go 0+    VU.unsafeFreeze out++-------------------------------------------------------------------------------+-- Integer mean (exact integer sum + count, divided once -> order-independent)+-------------------------------------------------------------------------------++{- | Integer mean in ONE fused pass: a running integer sum and count per group,+divided once at finalize. The integer sum is exact, so the parallel partial+merge is byte-identical to the sequential single pass at any @-N@.+-}+meanIntDense :: VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Double+meanIntDense g nGroups v+    | not (shouldPar n) = unsafePerformIO $ do+        (s, c) <- meanIntChunk g v nGroups 0 n+        finalizeMeanInt nGroups s c+    | otherwise = unsafePerformIO $ do+        parts <- runPartialsOver n capabilities (meanIntChunk g v nGroups)+        (s, c) <- mergePair nGroups parts+        finalizeMeanInt nGroups s c+  where+    !n = VU.length v+{-# NOINLINE meanIntDense #-}++meanIntChunk ::+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    Int ->+    Int ->+    IO (VUM.IOVector Int, VUM.IOVector Int)+meanIntChunk g v nGroups lo hi = do+    s <- VUM.replicate nGroups (0 :: Int)+    c <- VUM.replicate nGroups (0 :: Int)+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                sv <- VUM.unsafeRead s k+                VUM.unsafeWrite s k (sv + VU.unsafeIndex v i)+                cv <- VUM.unsafeRead c k+                VUM.unsafeWrite c k (cv + 1)+                go (i + 1)+    go lo+    pure (s, c)++finalizeMeanInt ::+    Int -> VUM.IOVector Int -> VUM.IOVector Int -> IO (VU.Vector Double)+finalizeMeanInt nGroups s c = do+    out <- VUM.new nGroups+    let go !k+            | k >= nGroups = pure ()+            | otherwise = do+                sv <- VUM.unsafeRead s k+                cv <- VUM.unsafeRead c k+                VUM.unsafeWrite+                    out+                    k+                    (if cv == 0 then 0 / 0 else fromIntegral sv / fromIntegral cv)+                go (k + 1)+    go 0+    VU.unsafeFreeze out++-------------------------------------------------------------------------------+-- Partial accumulation + merge+-------------------------------------------------------------------------------++mergeIntSum :: Int -> [VUM.IOVector Int] -> IO (VU.Vector Int)+mergeIntSum nGroups parts = case parts of+    [] -> VU.unsafeFreeze =<< VUM.replicate nGroups 0+    (p0 : rest) -> do+        let add !p = do+                let go !k+                        | k >= nGroups = pure ()+                        | otherwise = do+                            a <- VUM.unsafeRead p0 k+                            b <- VUM.unsafeRead p k+                            VUM.unsafeWrite p0 k (a + b)+                            go (k + 1)+                go 0+        mapM_ add rest+        VU.unsafeFreeze p0++{- | Sum the per-worker Double partials into the first worker's accumulator IN+WORKER ORDER: deterministic at a fixed @-N@, but the float summation order is+chunk-major rather than the sequential row order.+-}+mergeDblSum :: Int -> [VUM.IOVector Double] -> IO (VU.Vector Double)+mergeDblSum nGroups parts = case parts of+    [] -> VU.unsafeFreeze =<< VUM.replicate nGroups 0+    (p0 : rest) -> do+        let add !p = do+                let go !k+                        | k >= nGroups = pure ()+                        | otherwise = do+                            a <- VUM.unsafeRead p0 k+                            b <- VUM.unsafeRead p k+                            VUM.unsafeWrite p0 k (a + b)+                            go (k + 1)+                go 0+        mapM_ add rest+        VU.unsafeFreeze p0++{- | Merge per-worker (sum, count) partials into the first worker's pair by+exact integer addition; returns the accumulated pair for finalize.+-}+mergePair ::+    Int ->+    [(VUM.IOVector Int, VUM.IOVector Int)] ->+    IO (VUM.IOVector Int, VUM.IOVector Int)+mergePair nGroups parts = case parts of+    [] -> (,) <$> VUM.replicate nGroups 0 <*> VUM.replicate nGroups 0+    ((s0, c0) : rest) -> do+        let add (s, c) = do+                let go !k+                        | k >= nGroups = pure ()+                        | otherwise = do+                            sa <- VUM.unsafeRead s0 k+                            sb <- VUM.unsafeRead s k+                            VUM.unsafeWrite s0 k (sa + sb)+                            ca <- VUM.unsafeRead c0 k+                            cb <- VUM.unsafeRead c k+                            VUM.unsafeWrite c0 k (ca + cb)+                            go (k + 1)+                go 0+        mapM_ add rest+        pure (s0, c0)++{- | As 'mergePair' but for the Double (sum, count) partials of the Double mean;+worker-order float sums (see 'mergeDblSum').+-}+mergeMeanDbl ::+    Int ->+    [(VUM.IOVector Double, VUM.IOVector Int)] ->+    IO (VUM.IOVector Double, VUM.IOVector Int)+mergeMeanDbl nGroups parts = case parts of+    [] -> (,) <$> VUM.replicate nGroups 0 <*> VUM.replicate nGroups 0+    ((s0, c0) : rest) -> do+        let add (s, c) = do+                let go !k+                        | k >= nGroups = pure ()+                        | otherwise = do+                            sa <- VUM.unsafeRead s0 k+                            sb <- VUM.unsafeRead s k+                            VUM.unsafeWrite s0 k (sa + sb)+                            ca <- VUM.unsafeRead c0 k+                            cb <- VUM.unsafeRead c k+                            VUM.unsafeWrite c0 k (ca + cb)+                            go (k + 1)+                go 0+        mapM_ add rest+        pure (s0, c0)++mergeExtremaInt :: Bool -> Int -> [VUM.IOVector Int] -> IO (VU.Vector Int)+mergeExtremaInt isMin nGroups parts = case parts of+    [] ->+        VU.unsafeFreeze =<< VUM.replicate nGroups (if isMin then maxBound else minBound)+    (p0 : rest) -> do+        let combine a b = if isMin then min a b else max a b+            add !p = do+                let go !k+                        | k >= nGroups = pure ()+                        | otherwise = do+                            a <- VUM.unsafeRead p0 k+                            b <- VUM.unsafeRead p k+                            VUM.unsafeWrite p0 k (combine a b)+                            go (k + 1)+                go 0+        mapM_ add rest+        VU.unsafeFreeze p0++mergeExtremaDbl :: Bool -> Int -> [VUM.IOVector Double] -> IO (VU.Vector Double)+mergeExtremaDbl isMin nGroups parts = case parts of+    [] ->+        VU.unsafeFreeze+            =<< VUM.replicate nGroups (if isMin then 1 / 0 else negate (1 / 0))+    (p0 : rest) -> do+        let combine a b = if isMin then min a b else max a b+            add !p = do+                let go !k+                        | k >= nGroups = pure ()+                        | otherwise = do+                            a <- VUM.unsafeRead p0 k+                            b <- VUM.unsafeRead p k+                            VUM.unsafeWrite p0 k (combine a b)+                            go (k + 1)+                go 0+        mapM_ add rest+        VU.unsafeFreeze p0

@@ -0,0 +1,757 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ExplicitNamespaces #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++{- | Fused multi-reduction aggregation kernels: several reductions over the+SAME grouping, evaluated in one pass over the rows.++Two shapes, picked by group-domain width:++* 'mkFusedAgg' \/ 'runFusedAggs' — the STREAMING pass, driven by @rowToGroup@.+  No @valueIndices@ and no placement pass; each worker keeps a private+  accumulator per reduction and the partials merge over group slices. Capped at+  'streamGroupCap', above which the per-worker arrays overflow cache.++* 'mkGatherAgg' \/ 'runGatherAggs' — the GATHER pass, driven by the grouped+  @(valueIndices, offsets)@ layout, for group domains too wide to stream.++Both amortize the memory traffic of the grouping across every reduction in the+batch instead of re-reading it once per reduction.+-}+module DataFrame.Internal.Aggregation.Kernel.Fused (+    FusedAgg,+    mkFusedAgg,+    runFusedAggs,+    GatherAgg,+    mkGatherAgg,+    runGatherAggs,+) where++import Control.Exception (evaluate)+import Control.Monad (replicateM, when)+import Data.Type.Equality (TestEquality (..), type (:~:) (Refl))+import qualified Data.Vector.Unboxed as VU+import qualified Data.Vector.Unboxed.Mutable as VUM+import System.IO.Unsafe (unsafePerformIO)+import Type.Reflection (typeRep)++import DataFrame.Internal.Aggregation.Kernel.Scatter (+    groupRangeBounds,+    groupSlices,+    overGroupsAcc,+    streamGroupCap,+ )+import DataFrame.Internal.Aggregation.Reduction (Reduction (..))+import DataFrame.Internal.Column (Column (..), fromUnboxedVector)+import DataFrame.Internal.Control.Concurrent (+    capabilities,+    forkJoin,+    parThreshold,+    parallelBounds_,+    shouldParallelize,+ )++-- | Whether to fan out at this row count.+shouldPar :: Int -> Bool+shouldPar = shouldParallelize parThreshold++-------------------------------------------------------------------------------+-- Fused multi-reduction streaming pass+-------------------------------------------------------------------------------++{- | One fused reduction of a multi-reduction streaming pass: how to allocate a+per-worker accumulator, fold a row range into it, merge another worker's+accumulator into it over a group range (callers merge in worker order), and+finalize the fully merged accumulator into the output column.+-}+data FusedAgg+    = forall s.+        FusedAgg+        (IO s)+        -- \^ allocate one worker's accumulator+        (s -> Int -> Int -> IO ())+        -- \^ accumulate rows [lo, hi) in ascending order+        (s -> s -> Int -> Int -> IO ())+        -- \^ merge the second accumulator into the first over groups [lo, hi)+        (s -> IO Column)++-- \^ finalize the merged accumulator++{- | Below this many groups a Double sum/mean does NOT join the fused streaming+pass: the per-expression kernel it would replace+('DataFrame.Internal.AggKernelDirect.sumDblDirect' under its @seqFloatGroups@+policy, value mirrored here) runs those as ONE sequential row-order pass that+is byte-identical to the gather kernels and the interpreter, and the vectorized+parity gate asserts exactly that. Above it the per-expression kernel already+merges chunk partials in worker order, so fusing changes nothing semantically+new. Int reductions and count/min/max are exact under any merge and always+fuse.+-}+fusedSeqFloatGroups :: Int+fusedSeqFloatGroups = 65536++{- | Build the fused-pass reduction for one @(reduction, column)@ pair, or+'Nothing' when the pair cannot stream (nullable/boxed columns; the+order-sensitive var/std/top2 reductions, which keep their per-expression+kernels; or a Double sum/mean below 'fusedSeqFloatGroups', which keeps its+byte-identical sequential pass). Sum/min/max/count/mean over Int are exact+under any chunk merge; the admitted Double sum and mean merge their per-worker+partials in worker order (deterministic at a fixed @-N@, chunk-major float+summation order).+-}+mkFusedAgg :: Int -> VU.Vector Int -> Reduction -> Column -> Maybe FusedAgg+mkFusedAgg nGroups rtg red col+    | nGroups <= 0 || nGroups > streamGroupCap = Nothing+    | otherwise = case col of+        UnboxedColumn Nothing (v :: VU.Vector a)+            | Just Refl <- testEquality (typeRep @a) (typeRep @Int) ->+                case red of+                    RCount -> Just (countFusedAgg nGroups rtg)+                    RSum -> Just (sumIntFusedAgg nGroups rtg v)+                    RMean -> Just (meanIntFusedAgg nGroups rtg v)+                    RMin -> Just (extremaIntFusedAgg True nGroups rtg v)+                    RMax -> Just (extremaIntFusedAgg False nGroups rtg v)+                    _ -> Nothing+            | Just Refl <- testEquality (typeRep @a) (typeRep @Double) ->+                case red of+                    RCount -> Just (countFusedAgg nGroups rtg)+                    RSum+                        | nGroups > fusedSeqFloatGroups ->+                            Just (sumDblFusedAgg nGroups rtg v)+                    RMean+                        | nGroups > fusedSeqFloatGroups ->+                            Just (meanDblFusedAgg nGroups rtg v)+                    RMin -> Just (extremaDblFusedAgg True nGroups rtg v)+                    RMax -> Just (extremaDblFusedAgg False nGroups rtg v)+                    -- Top-2 selection is an exact multiset selection (no float+                    -- arithmetic before finalize), so its per-worker merge is+                    -- byte-identical to the per-expression kernels at any -N.+                    RTop2Snd -> Just (top2SndDblFusedAgg nGroups rtg v)+                    _ -> Nothing+        _ -> Nothing++{- | Execute all fused reductions in ONE pass over @rowToGroup@ and the value+columns: the rows are split into one contiguous chunk per capability, each+worker walks its chunk in 4096-row blocks running every reduction's+accumulate-step on the block (the block's @rowToGroup@ slice stays in L1+across the k steps), and each reduction then merges its per-worker partials in+fixed worker order (parallel over group slices) and finalizes. Per-group update+order within each worker is ascending original row order, and chunk boundaries+are a fixed function of the row and capability counts, so the result is+deterministic at a fixed @-N@; on a single capability it is bit-identical to+running each unfused sequential scatter kernel separately.++Pure w.r.t. its immutable inputs (deterministic fan-out and merge order), so+the 'unsafePerformIO' is safe.+-}+runFusedAggs :: Int -> Int -> [FusedAgg] -> [Column]+runFusedAggs n nGroups aggs = unsafePerformIO $ do+    let !caps' = if shouldPar n then capabilities else 1+        !per = (max 1 n + caps' - 1) `div` caps'+    opened <- mapM (openFusedAgg caps' nGroups) aggs++    let stepsFor w = map (\(steps, _) -> steps !! w) opened+    _ <-+        forkJoin+            [ blockRun (stepsFor w) lo hi+            | w <- [0 .. caps' - 1]+            , let lo = min n (w * per)+            , let hi = min n (lo + per)+            ]+    mapM snd opened+{-# NOINLINE runFusedAggs #-}++{- | Open one fused reduction for @caps'@ workers: its per-worker step+functions (worker order) and the merge+finalize action.+-}+openFusedAgg :: Int -> Int -> FusedAgg -> IO ([Int -> Int -> IO ()], IO Column)+openFusedAgg caps' nGroups (FusedAgg new step mergeR fin) = do+    ss <- replicateM caps' new+    let finish = case ss of+            [] -> error "runFusedAggs: no workers"+            (s0 : rest) -> do+                _ <-+                    forkJoin+                        [ mapM_ (\s -> mergeR s0 s lo hi) rest+                        | (lo, hi) <- groupSlices nGroups+                        ]+                fin s0+    pure (map step ss, finish)++-- | Rows per fused block: the block's rowToGroup slice (32KB) stays in L1.+fusedBlock :: Int+fusedBlock = 4096++blockRun :: [Int -> Int -> IO ()] -> Int -> Int -> IO ()+blockRun steps lo0 hi = go lo0+  where+    go !lo+        | lo >= hi = pure ()+        | otherwise = do+            let !e = min hi (lo + fusedBlock)+            mapM_ (\s -> s lo e) steps+            go e++-- Individual fused reductions. Each step loop is monomorphic.++countFusedAgg :: Int -> VU.Vector Int -> FusedAgg+countFusedAgg nGroups rtg =+    FusedAgg+        (VUM.replicate nGroups (0 :: Int))+        (countStepK rtg)+        addIntRange+        (fmap fromUnboxedVector . VU.unsafeFreeze)++countStepK :: VU.Vector Int -> VUM.IOVector Int -> Int -> Int -> IO ()+countStepK rtg acc lo hi = go lo+  where+    go !i+        | i >= hi = pure ()+        | otherwise = do+            let !k = VU.unsafeIndex rtg i+            c <- VUM.unsafeRead acc k+            VUM.unsafeWrite acc k (c + 1)+            go (i + 1)++sumIntFusedAgg :: Int -> VU.Vector Int -> VU.Vector Int -> FusedAgg+sumIntFusedAgg nGroups rtg v =+    FusedAgg+        (VUM.replicate nGroups (0 :: Int))+        (sumStepInt rtg v)+        addIntRange+        (fmap fromUnboxedVector . VU.unsafeFreeze)++sumStepInt ::+    VU.Vector Int -> VU.Vector Int -> VUM.IOVector Int -> Int -> Int -> IO ()+sumStepInt rtg v acc lo hi = go lo+  where+    go !i+        | i >= hi = pure ()+        | otherwise = do+            let !k = VU.unsafeIndex rtg i+            c <- VUM.unsafeRead acc k+            VUM.unsafeWrite acc k (c + VU.unsafeIndex v i)+            go (i + 1)++sumDblFusedAgg :: Int -> VU.Vector Int -> VU.Vector Double -> FusedAgg+sumDblFusedAgg nGroups rtg v =+    FusedAgg+        (VUM.replicate nGroups (0 :: Double))+        (sumStepDbl rtg v)+        addDblRange+        (fmap fromUnboxedVector . VU.unsafeFreeze)++sumStepDbl ::+    VU.Vector Int -> VU.Vector Double -> VUM.IOVector Double -> Int -> Int -> IO ()+sumStepDbl rtg v acc lo hi = go lo+  where+    go !i+        | i >= hi = pure ()+        | otherwise = do+            let !k = VU.unsafeIndex rtg i+            c <- VUM.unsafeRead acc k+            VUM.unsafeWrite acc k (c + VU.unsafeIndex v i)+            go (i + 1)++{- | The mean aggs hold their (sum, count) pair INTERLEAVED in one array —+slots @2g@/@2g+1@ share a cache line (pairs are 16-byte aligned, so they never+straddle one), halving the accumulator misses of the random per-row update+against two separate arrays (measured ~13% off a fused sum+mean pass at 1e6+groups / 1e8 rows on -N16). The count is exact in both layouts (an integer, or+integer-valued Double additions well below 2^53), so sums, merges and the+finalize divide are bit-identical to the two-array layout.+-}+meanIntFusedAgg :: Int -> VU.Vector Int -> VU.Vector Int -> FusedAgg+meanIntFusedAgg nGroups rtg v =+    FusedAgg+        (VUM.replicate (2 * nGroups) (0 :: Int))+        (meanStepInt rtg v)+        (\a b lo hi -> addIntRange a b (2 * lo) (2 * hi))+        ( \s -> do+            sv <- VU.unsafeFreeze s+            pure+                ( fromUnboxedVector+                    ( VU.generate+                        nGroups+                        ( \g ->+                            let !sx = VU.unsafeIndex sv (2 * g)+                                !cx = VU.unsafeIndex sv (2 * g + 1)+                             in if cx == 0+                                    then 0 / 0+                                    else fromIntegral sx / fromIntegral cx :: Double+                        )+                    )+                )+        )++meanStepInt ::+    VU.Vector Int ->+    VU.Vector Int ->+    VUM.IOVector Int ->+    Int ->+    Int ->+    IO ()+meanStepInt rtg v s lo hi = go lo+  where+    go !i+        | i >= hi = pure ()+        | otherwise = do+            let !k2 = 2 * VU.unsafeIndex rtg i+            sv <- VUM.unsafeRead s k2+            VUM.unsafeWrite s k2 (sv + VU.unsafeIndex v i)+            cv <- VUM.unsafeRead s (k2 + 1)+            VUM.unsafeWrite s (k2 + 1) (cv + 1)+            go (i + 1)++-- | See 'meanIntFusedAgg' for the interleaved accumulator layout.+meanDblFusedAgg :: Int -> VU.Vector Int -> VU.Vector Double -> FusedAgg+meanDblFusedAgg nGroups rtg v =+    FusedAgg+        (VUM.replicate (2 * nGroups) (0 :: Double))+        (meanStepDbl rtg v)+        (\a b lo hi -> addDblRange a b (2 * lo) (2 * hi))+        ( \s -> do+            sv <- VU.unsafeFreeze s+            pure+                ( fromUnboxedVector+                    ( VU.generate+                        nGroups+                        ( \g ->+                            let !sx = VU.unsafeIndex sv (2 * g)+                                !cx = VU.unsafeIndex sv (2 * g + 1)+                             in if cx == 0 then 0 / 0 else sx / cx+                        )+                    )+                )+        )++meanStepDbl ::+    VU.Vector Int ->+    VU.Vector Double ->+    VUM.IOVector Double ->+    Int ->+    Int ->+    IO ()+meanStepDbl rtg v s lo hi = go lo+  where+    go !i+        | i >= hi = pure ()+        | otherwise = do+            let !k2 = 2 * VU.unsafeIndex rtg i+            sv <- VUM.unsafeRead s k2+            VUM.unsafeWrite s k2 (sv + VU.unsafeIndex v i)+            cv <- VUM.unsafeRead s (k2 + 1)+            VUM.unsafeWrite s (k2 + 1) (cv + 1)+            go (i + 1)++extremaIntFusedAgg :: Bool -> Int -> VU.Vector Int -> VU.Vector Int -> FusedAgg+extremaIntFusedAgg isMin nGroups rtg v =+    FusedAgg+        (VUM.replicate nGroups (if isMin then maxBound else minBound :: Int))+        (extremaStepInt isMin rtg v)+        (combineIntRange isMin)+        (fmap fromUnboxedVector . VU.unsafeFreeze)++extremaStepInt ::+    Bool ->+    VU.Vector Int ->+    VU.Vector Int ->+    VUM.IOVector Int ->+    Int ->+    Int ->+    IO ()+extremaStepInt isMin rtg v acc lo hi = go lo+  where+    go !i+        | i >= hi = pure ()+        | otherwise = do+            let !k = VU.unsafeIndex rtg i+                !x = VU.unsafeIndex v i+            c <- VUM.unsafeRead acc k+            VUM.unsafeWrite acc k (if isMin then min c x else max c x)+            go (i + 1)++extremaDblFusedAgg ::+    Bool -> Int -> VU.Vector Int -> VU.Vector Double -> FusedAgg+extremaDblFusedAgg isMin nGroups rtg v =+    FusedAgg+        (VUM.replicate nGroups (if isMin then 1 / 0 else negate (1 / 0) :: Double))+        (extremaStepDbl isMin rtg v)+        (combineDblRange isMin)+        (fmap fromUnboxedVector . VU.unsafeFreeze)++extremaStepDbl ::+    Bool ->+    VU.Vector Int ->+    VU.Vector Double ->+    VUM.IOVector Double ->+    Int ->+    Int ->+    IO ()+extremaStepDbl isMin rtg v acc lo hi = go lo+  where+    go !i+        | i >= hi = pure ()+        | otherwise = do+            let !k = VU.unsafeIndex rtg i+                !x = VU.unsafeIndex v i+            c <- VUM.unsafeRead acc k+            VUM.unsafeWrite acc k (if isMin then min c x else max c x)+            go (i + 1)++{- | Fused second-largest over a Double column. The per-group+(largest, second-largest) pair is INTERLEAVED at slots @2g@/@2g+1@ (one cache+line per group, as 'meanDblFusedAgg'); the update is the same top-2 selection+as every other top2 kernel, the merge keeps the top two of the four candidates+per group (exact — no float arithmetic), and the finalize returns the second+max, NaN for a group of size < 2 (the @-inf@ seed; see+'DataFrame.Internal.AggKernel.top2SndScatter').+-}+top2SndDblFusedAgg :: Int -> VU.Vector Int -> VU.Vector Double -> FusedAgg+top2SndDblFusedAgg nGroups rtg v =+    FusedAgg+        (VUM.replicate (2 * nGroups) (negate (1 / 0) :: Double))+        (top2SndStepDbl rtg v)+        mergeTop2Range+        ( \s -> do+            sv <- VU.unsafeFreeze s+            pure+                ( fromUnboxedVector+                    ( VU.generate+                        nGroups+                        ( \g ->+                            let !a2 = VU.unsafeIndex sv (2 * g + 1)+                             in if isInfinite a2 then 0 / 0 else a2+                        )+                    )+                )+        )++top2SndStepDbl ::+    VU.Vector Int ->+    VU.Vector Double ->+    VUM.IOVector Double ->+    Int ->+    Int ->+    IO ()+top2SndStepDbl rtg v s lo hi = go lo+  where+    go !i+        | i >= hi = pure ()+        | otherwise = do+            let !k2 = 2 * VU.unsafeIndex rtg i+                !x = VU.unsafeIndex v i+            a1 <- VUM.unsafeRead s k2+            if x > a1+                then do+                    VUM.unsafeWrite s k2 x+                    VUM.unsafeWrite s (k2 + 1) a1+                else do+                    a2 <- VUM.unsafeRead s (k2 + 1)+                    when (x > a2) $ VUM.unsafeWrite s (k2 + 1) x+            go (i + 1)++-- | Top two of the four candidates per group (pairs already ordered m1 >= m2).+mergeTop2Range ::+    VUM.IOVector Double -> VUM.IOVector Double -> Int -> Int -> IO ()+mergeTop2Range a b lo hi = go lo+  where+    go !g+        | g >= hi = pure ()+        | otherwise = do+            let !g2 = 2 * g+            a1 <- VUM.unsafeRead a g2+            a2 <- VUM.unsafeRead a (g2 + 1)+            b1 <- VUM.unsafeRead b g2+            b2 <- VUM.unsafeRead b (g2 + 1)+            if b1 > a1+                then do+                    VUM.unsafeWrite a g2 b1+                    VUM.unsafeWrite a (g2 + 1) (max a1 b2)+                else VUM.unsafeWrite a (g2 + 1) (max a2 b1)+            go (g + 1)++addIntRange :: VUM.IOVector Int -> VUM.IOVector Int -> Int -> Int -> IO ()+addIntRange a b lo hi = go lo+  where+    go !g+        | g >= hi = pure ()+        | otherwise = do+            x <- VUM.unsafeRead a g+            y <- VUM.unsafeRead b g+            VUM.unsafeWrite a g (x + y)+            go (g + 1)++addDblRange :: VUM.IOVector Double -> VUM.IOVector Double -> Int -> Int -> IO ()+addDblRange a b lo hi = go lo+  where+    go !g+        | g >= hi = pure ()+        | otherwise = do+            x <- VUM.unsafeRead a g+            y <- VUM.unsafeRead b g+            VUM.unsafeWrite a g (x + y)+            go (g + 1)++combineIntRange ::+    Bool -> VUM.IOVector Int -> VUM.IOVector Int -> Int -> Int -> IO ()+combineIntRange isMin a b lo hi = go lo+  where+    go !g+        | g >= hi = pure ()+        | otherwise = do+            x <- VUM.unsafeRead a g+            y <- VUM.unsafeRead b g+            VUM.unsafeWrite a g (if isMin then min x y else max x y)+            go (g + 1)++combineDblRange ::+    Bool -> VUM.IOVector Double -> VUM.IOVector Double -> Int -> Int -> IO ()+combineDblRange isMin a b lo hi = go lo+  where+    go !g+        | g >= hi = pure ()+        | otherwise = do+            x <- VUM.unsafeRead a g+            y <- VUM.unsafeRead b g+            VUM.unsafeWrite a g (if isMin then min x y else max x y)+            go (g + 1)++-------------------------------------------------------------------------------+-- Fused multi-reduction gather pass (nGroups above 'streamGroupCap')+-------------------------------------------------------------------------------++{- | One fused reduction of a multi-reduction GATHER pass: allocate the output+array, fold a contiguous group range (each group's rows via the shared+@valueIndices@ slice, accumulator in registers, one write per group), finalize.+Group ranges are disjoint across workers, so there is no merge and every+reduction reproduces the exact per-group fold order and formula of its unfused+gather kernel ('reduceParTyped') — results are bit-identical to running the+kernels separately, at any @-N@.+-}+data GatherAgg+    = forall s.+        GatherAgg+        (IO s)+        (s -> Int -> Int -> IO ())+        (s -> IO Column)++{- | Build the fused gather reduction for one @(reduction, column)@ pair, or+'Nothing' when the pair cannot fuse (nullable/boxed columns, or var/std/top2,+which keep their per-expression gather kernels). @vis@/@offs@ are captured+lazily: nothing is forced until the pass actually runs.+-}+mkGatherAgg ::+    Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    Reduction ->+    Column ->+    Maybe GatherAgg+mkGatherAgg nGroups vis offs red col = case col of+    UnboxedColumn Nothing (v :: VU.Vector a)+        | Just Refl <- testEquality (typeRep @a) (typeRep @Int) ->+            case red of+                RCount -> Just countGather+                RSum -> Just (outGather (gatherSumInt vis offs v))+                RMean -> Just (outGatherD (gatherMeanInt vis offs v))+                RMin -> Just (outGather (gatherExtremaInt True vis offs v))+                RMax -> Just (outGather (gatherExtremaInt False vis offs v))+                _ -> Nothing+        | Just Refl <- testEquality (typeRep @a) (typeRep @Double) ->+            case red of+                RCount -> Just countGather+                RSum -> Just (outGatherD (gatherSumDbl vis offs v))+                RMean -> Just (outGatherD (gatherMeanDbl vis offs v))+                RMin -> Just (outGatherD (gatherExtremaDbl True vis offs v))+                RMax -> Just (outGatherD (gatherExtremaDbl False vis offs v))+                _ -> Nothing+    _ -> Nothing+  where+    countGather =+        GatherAgg+            (VUM.new nGroups)+            (gatherCount offs)+            (fmap fromUnboxedVector . VU.unsafeFreeze)+    outGather step =+        GatherAgg+            (VUM.new nGroups :: IO (VUM.IOVector Int))+            step+            (fmap fromUnboxedVector . VU.unsafeFreeze)+    outGatherD step =+        GatherAgg+            (VUM.new nGroups :: IO (VUM.IOVector Double))+            step+            (fmap fromUnboxedVector . VU.unsafeFreeze)++{- | Number of groups each fused-gather block hands to every reduction before+moving on: the block's @valueIndices@ slice stays hot in cache across the k+per-reduction loops.+-}+gatherBlock :: Int+gatherBlock = 32++{- | Execute all fused gather reductions in one traversal: workers own disjoint+contiguous group ranges (row-balanced, same policy as every gather kernel), and+walk them in 'gatherBlock'-group blocks running each reduction's fold on the+block. Deterministic and bit-identical to the unfused kernels (see+'GatherAgg'). Forces @valueIndices@ once, before the fan-out.++Pure w.r.t. its immutable inputs, so the 'unsafePerformIO' is safe.+-}+runGatherAggs ::+    VU.Vector Int -> VU.Vector Int -> Int -> [GatherAgg] -> [Column]+runGatherAggs vis offs nGroups aggs = unsafePerformIO $ do+    _ <- evaluate (VU.length vis)+    let !caps = capabilities+        !bounds = groupRangeBounds offs nGroups caps+    opened <-+        mapM (\(GatherAgg new step fin) -> do s <- new; pure (step s, fin s)) aggs+    parallelBounds_ caps bounds $ \gs ge ->+        let go !g+                | g >= ge = pure ()+                | otherwise = do+                    let !e = min ge (g + gatherBlock)+                    mapM_ (\(st, _) -> st g e) opened+                    go e+         in go gs+    mapM snd opened+{-# NOINLINE runGatherAggs #-}++-- Monomorphic per-reduction gather folds; each replicates the exact per-group+-- recurrence of its unfused kernel above ('countPar'/'sumPar'/'extremaPar'/+-- 'meanPar'), so fused results are bit-identical.++gatherCount :: VU.Vector Int -> VUM.IOVector Int -> Int -> Int -> IO ()+gatherCount offs out gs ge = go gs+  where+    go !g+        | g >= ge = pure ()+        | otherwise = do+            let !c = VU.unsafeIndex offs (g + 1) - VU.unsafeIndex offs g+            VUM.unsafeWrite out g c+            go (g + 1)++gatherSumInt ::+    VU.Vector Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    VUM.IOVector Int ->+    Int ->+    Int ->+    IO ()+gatherSumInt vis offs v out gs ge =+    overGroupsAcc vis offs gs ge 0 (\acc row -> acc + VU.unsafeIndex v row) $+        VUM.unsafeWrite out++gatherSumDbl ::+    VU.Vector Int ->+    VU.Vector Int ->+    VU.Vector Double ->+    VUM.IOVector Double ->+    Int ->+    Int ->+    IO ()+gatherSumDbl vis offs v out gs ge =+    overGroupsAcc vis offs gs ge 0 (\acc row -> acc + VU.unsafeIndex v row) $+        VUM.unsafeWrite out++gatherExtremaInt ::+    Bool ->+    VU.Vector Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    VUM.IOVector Int ->+    Int ->+    Int ->+    IO ()+gatherExtremaInt isMin vis offs v out gs ge =+    overGroupsAcc+        vis+        offs+        gs+        ge+        (if isMin then maxBound else minBound)+        ( \acc row ->+            let !x = VU.unsafeIndex v row+             in if isMin then min acc x else max acc x+        )+        (VUM.unsafeWrite out)++gatherExtremaDbl ::+    Bool ->+    VU.Vector Int ->+    VU.Vector Int ->+    VU.Vector Double ->+    VUM.IOVector Double ->+    Int ->+    Int ->+    IO ()+gatherExtremaDbl isMin vis offs v out gs ge =+    overGroupsAcc+        vis+        offs+        gs+        ge+        (if isMin then 1 / 0 else negate (1 / 0))+        ( \acc row ->+            let !x = VU.unsafeIndex v row+             in if isMin then min acc x else max acc x+        )+        (VUM.unsafeWrite out)++-- | Exact replica of 'meanPar''s per-group loop (Int element type).+gatherMeanInt ::+    VU.Vector Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    VUM.IOVector Double ->+    Int ->+    Int ->+    IO ()+gatherMeanInt vis offs v out gs ge = grp gs+  where+    grp !g+        | g >= ge = pure ()+        | otherwise = do+            let !e = VU.unsafeIndex offs (g + 1)+                inner !pos !acc+                    | pos >= e = acc+                    | otherwise =+                        inner+                            (pos + 1)+                            (acc + fromIntegral (VU.unsafeIndex v (VU.unsafeIndex vis pos)))+                !s0 = VU.unsafeIndex offs g+                !total = inner s0 0+                !c = e - s0+            VUM.unsafeWrite out g (if c == 0 then 0 / 0 else total / fromIntegral c)+            grp (g + 1)++-- | Exact replica of 'meanPar''s per-group loop (Double element type).+gatherMeanDbl ::+    VU.Vector Int ->+    VU.Vector Int ->+    VU.Vector Double ->+    VUM.IOVector Double ->+    Int ->+    Int ->+    IO ()+gatherMeanDbl vis offs v out gs ge = grp gs+  where+    grp !g+        | g >= ge = pure ()+        | otherwise = do+            let !e = VU.unsafeIndex offs (g + 1)+                inner !pos !acc+                    | pos >= e = acc+                    | otherwise =+                        inner+                            (pos + 1)+                            (acc + VU.unsafeIndex v (VU.unsafeIndex vis pos))+                !s0 = VU.unsafeIndex offs g+                !total = inner s0 0+                !c = e - s0+            VUM.unsafeWrite out g (if c == 0 then 0 / 0 else total / fromIntegral c)+            grp (g + 1)

@@ -0,0 +1,389 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ExplicitNamespaces #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++{- | The fused two-column moment kernel: one pass over @x@ and @y@ producing the+six sufficient statistics @(n, Sx, Sy, Sxx, Syy, Sxy)@ per group, from which the+whole correlation\/regression family (mean, variance, covariance, correlation,+OLS slope) is algebra requiring no further look at the rows.++The sequential and parallel passes live together because they must agree on+floating-point accumulation order. Moments are additive, so a row-range split+with a merge would be correct in exact arithmetic — but float addition is not+associative, so 'momentScatterPar' partitions by GROUP range instead: every+group accumulates start-to-finish inside a single worker, in the same order as+'momentScatter'. That is what makes the two byte-identical at any @-N@.+-}+module DataFrame.Internal.Aggregation.Kernel.Moments (+    Moments (..),+    momentScatter,+    momentScatterPar,+    momentStreamPar,+) where++import Control.Monad.ST (runST)+import Data.Type.Equality (TestEquality (..), type (:~:) (Refl))+import qualified Data.Vector.Unboxed as VU+import qualified Data.Vector.Unboxed.Mutable as VUM+import System.IO.Unsafe (unsafePerformIO)+import Type.Reflection (typeRep)++import DataFrame.Internal.Aggregation.Kernel.Scatter (+    groupRangeBounds,+    groupSlices,+    rtgFromVis,+    streamGroupCap,+ )+import DataFrame.Internal.Aggregation.Reduction (cleanDoubleVector)+import DataFrame.Internal.Column (Column (..), fromUnboxedVector)+import DataFrame.Internal.Control.Concurrent (+    capabilities,+    forkJoin,+    parThreshold,+    parallelBounds_,+    shouldParallelize,+ )++-- | Whether to fan out at this row count.+shouldPar :: Int -> Bool+shouldPar = shouldParallelize parThreshold++{- | The additive moment sums of two columns, each an @nGroups@-length column:+@(n, Sx, Sy, Sxx, Syy, Sxy)@.+-}+data Moments = Moments+    { mN :: Column+    , mSx :: Column+    , mSy :: Column+    , mSxx :: Column+    , mSyy :: Column+    , mSxy :: Column+    }++{- | One pass over two Double-coercible columns @x@ and @y@ filling the count and+five sums, collapsing the Q9 regression family's six folds into a single pass.+'Nothing' unless both columns are non-null unboxed Int/Double.+-}+momentScatter :: VU.Vector Int -> Int -> Column -> Column -> Maybe Moments+momentScatter g nGroups colX colY = do+    xs <- cleanDoubleVector colX+    ys <- cleanDoubleVector colY+    let (cnt, sx, sy, sxx, syy, sxy) = momentPass g nGroups xs ys+    pure+        Moments+            { mN = fromUnboxedVector cnt+            , mSx = fromUnboxedVector sx+            , mSy = fromUnboxedVector sy+            , mSxx = fromUnboxedVector sxx+            , mSyy = fromUnboxedVector syy+            , mSxy = fromUnboxedVector sxy+            }++momentPass ::+    VU.Vector Int ->+    Int ->+    VU.Vector Double ->+    VU.Vector Double ->+    ( VU.Vector Int+    , VU.Vector Double+    , VU.Vector Double+    , VU.Vector Double+    , VU.Vector Double+    , VU.Vector Double+    )+momentPass g nGroups xs ys = runST $ do+    cnt <- VUM.replicate nGroups (0 :: Int)+    sx <- VUM.replicate nGroups (0 :: Double)+    sy <- VUM.replicate nGroups (0 :: Double)+    sxx <- VUM.replicate nGroups (0 :: Double)+    syy <- VUM.replicate nGroups (0 :: Double)+    sxy <- VUM.replicate nGroups (0 :: Double)+    let n = VU.length xs+        bump arr k d = VUM.unsafeRead arr k >>= \c -> VUM.unsafeWrite arr k (c + d)+        go !i+            | i >= n = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                    !x = VU.unsafeIndex xs i+                    !y = VU.unsafeIndex ys i+                VUM.unsafeRead cnt k >>= \c -> VUM.unsafeWrite cnt k (c + 1)+                bump sx k x+                bump sy k y+                bump sxx k (x * x)+                bump syy k (y * y)+                bump sxy k (x * y)+                go (i + 1)+    go 0+    (,,,,,)+        <$> VU.unsafeFreeze cnt+        <*> VU.unsafeFreeze sx+        <*> VU.unsafeFreeze sy+        <*> VU.unsafeFreeze sxx+        <*> VU.unsafeFreeze syy+        <*> VU.unsafeFreeze sxy++{- | Parallel counterpart of 'momentScatter': one fused pass over both columns,+each group's six sums accumulated within one worker's range. Byte-identical to+'momentScatter'. 'Nothing' unless both columns are non-null unboxed Int/Double.+-}+momentScatterPar ::+    VU.Vector Int -> VU.Vector Int -> Int -> Column -> Column -> Maybe Moments+momentScatterPar vis offs nGroups colX colY+    | not (shouldPar (VU.length vis)) || nGroups <= 1 =+        momentScatter (rtgFromVis vis offs nGroups) nGroups colX colY+    | otherwise = do+        xs <- cleanDoubleVector colX+        ys <- cleanDoubleVector colY+        let !caps = capabilities+            !bounds = groupRangeBounds offs nGroups caps+        pure (unsafePerformIO (momentPar vis offs nGroups xs ys caps bounds))+{-# NOINLINE momentScatterPar #-}++-------------------------------------------------------------------------------++momentPar ::+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Double ->+    VU.Vector Double ->+    Int ->+    VU.Vector Int ->+    IO Moments+momentPar vis offs nGroups xs ys caps bounds = do+    cnt <- VUM.replicate nGroups (0 :: Int)+    sx <- VUM.replicate nGroups (0 :: Double)+    sy <- VUM.replicate nGroups (0 :: Double)+    sxx <- VUM.replicate nGroups (0 :: Double)+    syy <- VUM.replicate nGroups (0 :: Double)+    sxy <- VUM.replicate nGroups (0 :: Double)+    parallelBounds_ caps bounds $ \gs ge ->+        -- The six running sums carried in registers per group, written once.+        let grp !g+                | g >= ge = pure ()+                | otherwise = do+                    let !e = VU.unsafeIndex offs (g + 1)+                        inner !pos !ax !ay !axx !ayy !axy+                            | pos >= e = do+                                VUM.unsafeWrite sx g ax+                                VUM.unsafeWrite sy g ay+                                VUM.unsafeWrite sxx g axx+                                VUM.unsafeWrite syy g ayy+                                VUM.unsafeWrite sxy g axy+                            | otherwise =+                                let !row = VU.unsafeIndex vis pos+                                    !x = VU.unsafeIndex xs row+                                    !y = VU.unsafeIndex ys row+                                 in inner+                                        (pos + 1)+                                        (ax + x)+                                        (ay + y)+                                        (axx + x * x)+                                        (ayy + y * y)+                                        (axy + x * y)+                        !s0 = VU.unsafeIndex offs g+                    VUM.unsafeWrite cnt g (e - s0)+                    inner s0 0 0 0 0 0+                    grp (g + 1)+         in grp gs+    Moments . fromUnboxedVector+        <$> VU.unsafeFreeze cnt+        <*> (fromUnboxedVector <$> VU.unsafeFreeze sx)+        <*> (fromUnboxedVector <$> VU.unsafeFreeze sy)+        <*> (fromUnboxedVector <$> VU.unsafeFreeze sxx)+        <*> (fromUnboxedVector <$> VU.unsafeFreeze syy)+        <*> (fromUnboxedVector <$> VU.unsafeFreeze sxy)++-------------------------------------------------------------------------------+-- Streaming fused two-column moments (Q9)+-------------------------------------------------------------------------------++{- | Streaming counterpart of 'momentScatterPar': one fused pass over+@rowToGroup@ and the two TYPED value columns (Int values convert to Double+in-register — bit-identical to the @VU.map fromIntegral@ materialization it+replaces, with no 800MB intermediate column and no sequential conversion pass).+Each worker accumulates the six per-group sums over its contiguous row chunk in+original row order; partials merge in fixed worker order (counts exactly, the+five Double sums in chunk-major float order — deterministic at a fixed @-N@,+but a different summation order than the per-group gather kernel).+'Nothing' above 'streamGroupCap' or unless both columns are clean unboxed+Int/Double; the caller then keeps the gather path.+-}+momentStreamPar :: VU.Vector Int -> Int -> Column -> Column -> Maybe Moments+momentStreamPar rtg nGroups colX colY+    | nGroups <= 0 || nGroups > streamGroupCap = Nothing+    | otherwise = case (colX, colY) of+        ( UnboxedColumn Nothing (vx :: VU.Vector x)+            , UnboxedColumn Nothing (vy :: VU.Vector y)+            )+                | Just Refl <- testEquality (typeRep @x) (typeRep @Int)+                , Just Refl <- testEquality (typeRep @y) (typeRep @Int) ->+                    Just (momentStreamII rtg nGroups vx vy)+                | Just Refl <- testEquality (typeRep @x) (typeRep @Int)+                , Just Refl <- testEquality (typeRep @y) (typeRep @Double) ->+                    Just (momentStreamID rtg nGroups vx vy)+                | Just Refl <- testEquality (typeRep @x) (typeRep @Double)+                , Just Refl <- testEquality (typeRep @y) (typeRep @Int) ->+                    Just (momentStreamDI rtg nGroups vx vy)+                | Just Refl <- testEquality (typeRep @x) (typeRep @Double)+                , Just Refl <- testEquality (typeRep @y) (typeRep @Double) ->+                    Just (momentStreamDD rtg nGroups vx vy)+        _ -> Nothing+{-# NOINLINE momentStreamPar #-}++{- | Monomorphic entry points (see 'reduceParInt' for why the 'testEquality'+dispatch needs them).+-}+momentStreamII ::+    VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Int -> Moments+momentStreamII = momentStreamTyped+{-# NOINLINE momentStreamII #-}++momentStreamID ::+    VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Double -> Moments+momentStreamID = momentStreamTyped+{-# NOINLINE momentStreamID #-}++momentStreamDI ::+    VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Int -> Moments+momentStreamDI = momentStreamTyped+{-# NOINLINE momentStreamDI #-}++momentStreamDD ::+    VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Double -> Moments+momentStreamDD = momentStreamTyped+{-# NOINLINE momentStreamDD #-}++-- | The six per-group running sums of one worker chunk.+data MomentAcc = MomentAcc+    { maCnt :: !(VUM.IOVector Int)+    , maSx :: !(VUM.IOVector Double)+    , maSy :: !(VUM.IOVector Double)+    , maSxx :: !(VUM.IOVector Double)+    , maSyy :: !(VUM.IOVector Double)+    , maSxy :: !(VUM.IOVector Double)+    }++newMomentAcc :: Int -> IO MomentAcc+newMomentAcc nGroups =+    MomentAcc+        <$> VUM.replicate nGroups 0+        <*> VUM.replicate nGroups 0+        <*> VUM.replicate nGroups 0+        <*> VUM.replicate nGroups 0+        <*> VUM.replicate nGroups 0+        <*> VUM.replicate nGroups 0++momentStreamTyped ::+    forall a b.+    (VU.Unbox a, VU.Unbox b, Real a, Real b) =>+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    VU.Vector b ->+    Moments+{- The SPECIALIZE pragmas matter for the same reason as 'reduceParTyped': the+per-element @realToFrac@ must rewrite to @int2Double@/@id@ at a concrete type+or it goes through 'Rational' at runtime. -}+{-# SPECIALIZE momentStreamTyped ::+    VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Int -> Moments+    #-}+{-# SPECIALIZE momentStreamTyped ::+    VU.Vector Int -> Int -> VU.Vector Int -> VU.Vector Double -> Moments+    #-}+{-# SPECIALIZE momentStreamTyped ::+    VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Int -> Moments+    #-}+{-# SPECIALIZE momentStreamTyped ::+    VU.Vector Int -> Int -> VU.Vector Double -> VU.Vector Double -> Moments+    #-}+momentStreamTyped rtg nGroups vx vy = unsafePerformIO $ do+    let !n = VU.length rtg+        !caps' = if shouldPar n then capabilities else 1+        !per = (max 1 n + caps' - 1) `div` caps'+    parts <-+        forkJoin+            [ momentStreamChunk rtg nGroups vx vy lo hi+            | w <- [0 .. caps' - 1]+            , let lo = min n (w * per)+            , let hi = min n (lo + per)+            ]+    case parts of+        [] -> error "momentStreamTyped: no partials"+        (p0 : rest) -> do+            _ <-+                forkJoin+                    [ mapM_ (\p -> mergeMomentRange p0 p lo hi) rest+                    | (lo, hi) <- groupSlices nGroups+                    ]+            freezeMoments p0+{-# INLINEABLE momentStreamTyped #-}++momentStreamChunk ::+    (VU.Unbox a, VU.Unbox b, Real a, Real b) =>+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    VU.Vector b ->+    Int ->+    Int ->+    IO MomentAcc+momentStreamChunk rtg nGroups vx vy lo hi = do+    acc@(MomentAcc cnt sx sy sxx syy sxy) <- newMomentAcc nGroups+    let go !i+            | i >= hi = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex rtg i+                    !x = realToFrac (VU.unsafeIndex vx i) :: Double+                    !y = realToFrac (VU.unsafeIndex vy i) :: Double+                c <- VUM.unsafeRead cnt k+                VUM.unsafeWrite cnt k (c + 1)+                ax <- VUM.unsafeRead sx k+                VUM.unsafeWrite sx k (ax + x)+                ay <- VUM.unsafeRead sy k+                VUM.unsafeWrite sy k (ay + y)+                axx <- VUM.unsafeRead sxx k+                VUM.unsafeWrite sxx k (axx + x * x)+                ayy <- VUM.unsafeRead syy k+                VUM.unsafeWrite syy k (ayy + y * y)+                axy <- VUM.unsafeRead sxy k+                VUM.unsafeWrite sxy k (axy + x * y)+                go (i + 1)+    go lo+    pure acc+{-# INLINE momentStreamChunk #-}++mergeMomentRange :: MomentAcc -> MomentAcc -> Int -> Int -> IO ()+mergeMomentRange a b lo hi = go lo+  where+    go !g+        | g >= hi = pure ()+        | otherwise = do+            addI (maCnt a) (maCnt b) g+            addD (maSx a) (maSx b) g+            addD (maSy a) (maSy b) g+            addD (maSxx a) (maSxx b) g+            addD (maSyy a) (maSyy b) g+            addD (maSxy a) (maSxy b) g+            go (g + 1)+    addI p q g = do+        x <- VUM.unsafeRead p g+        y <- VUM.unsafeRead q g+        VUM.unsafeWrite p g (x + y)+    addD p q g = do+        x <- VUM.unsafeRead p g+        y <- VUM.unsafeRead q g+        VUM.unsafeWrite p g (x + y)++freezeMoments :: MomentAcc -> IO Moments+freezeMoments (MomentAcc cnt sx sy sxx syy sxy) =+    Moments . fromUnboxedVector+        <$> VU.unsafeFreeze cnt+        <*> (fromUnboxedVector <$> VU.unsafeFreeze sx)+        <*> (fromUnboxedVector <$> VU.unsafeFreeze sy)+        <*> (fromUnboxedVector <$> VU.unsafeFreeze sxx)+        <*> (fromUnboxedVector <$> VU.unsafeFreeze syy)+        <*> (fromUnboxedVector <$> VU.unsafeFreeze sxy)

@@ -0,0 +1,818 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ExplicitNamespaces #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++{- | The group-range scatter-accumulate reduction kernel: reduces a value column+over the grouped layout @(valueIndices, offsets)@.++Sequential and parallel are the same algorithm at different row counts, so they+live together. 'scatterReducePar' cuts the GROUP axis into @caps@ ranges of+roughly equal row count and lets workers write disjoint slots of one shared+output — no per-worker accumulator, no merge — which keeps each group's+accumulation order identical to 'scatterReduce' and the results byte-identical+at any @-N@. Below 'parThreshold' it delegates to 'scatterReduce' directly.++Contrast "DataFrame.Internal.Aggregation.Kernel.Dense", which scatters off+@rowToGroup@ with no gather but needs a small dense group domain.+-}+module DataFrame.Internal.Aggregation.Kernel.Scatter (+    scatterReduce,+    scatterReducePar,+    maxMinusMinScatterPar,+    top2SndScatter,++    -- * Group-range helpers+    -- $shared+    groupRangeBounds,+    rtgFromVis,+    overGroupsAcc,+    groupSlices,+    streamGroupCap,+) where++import Control.Monad (when)+import Control.Monad.ST (ST, runST)+import Data.Type.Equality (TestEquality (..), type (:~:) (Refl))+import qualified Data.Vector.Unboxed as VU+import qualified Data.Vector.Unboxed.Mutable as VUM+import System.IO.Unsafe (unsafePerformIO)+import Type.Reflection (typeRep)++import DataFrame.Internal.Aggregation.Reduction (Reduction (..))+import DataFrame.Internal.Column (+    Column (..),+    Columnable,+    fromUnboxedVector,+    materializePacked,+ )+import DataFrame.Internal.Control.Concurrent (+    capabilities,+    chunksFor,+    parThreshold,+    parallelBounds_,+    shouldParallelize,+ )++{- $shared+Also used by "DataFrame.Internal.Aggregation.Kernel.Moments", which partitions+the group axis the same way.+-}++-- | Whether to fan out at this row count.+shouldPar :: Int -> Bool+shouldPar = shouldParallelize parThreshold++scatterReduce ::+    Reduction -> VU.Vector Int -> Int -> Column -> Maybe Column+scatterReduce red g nGroups col = case col of+    UnboxedColumn Nothing (v :: VU.Vector a) ->+        case testEquality (typeRep @a) (typeRep @Int) of+            Just Refl -> Just (reduceTyped red g nGroups v intIdent)+            Nothing -> case testEquality (typeRep @a) (typeRep @Double) of+                Just Refl -> Just (reduceTyped red g nGroups v dblIdent)+                Nothing -> Nothing+    p@(PackedText _ _) -> scatterReduce red g nGroups (materializePacked p)+    _ -> Nothing+{-# INLINEABLE scatterReduce #-}++-- | Per-type seed identities for the order-preserving reductions.+data Idents a = Idents {minSeed :: !a, maxSeed :: !a}++intIdent :: Idents Int+intIdent = Idents maxBound minBound++dblIdent :: Idents Double+dblIdent = Idents (1 / 0) (negate (1 / 0))++reduceTyped ::+    forall a.+    (Columnable a, VU.Unbox a, Num a, Ord a, Real a) =>+    Reduction -> VU.Vector Int -> Int -> VU.Vector a -> Idents a -> Column+reduceTyped red g nGroups v idents = case red of+    RCount -> fromUnboxedVector (countScatter g nGroups)+    RSum -> fromUnboxedVector (sumScatter g nGroups v)+    RMin -> fromUnboxedVector (extremaScatter min (minSeed idents) g nGroups v)+    RMax -> fromUnboxedVector (extremaScatter max (maxSeed idents) g nGroups v)+    RMean -> fromUnboxedVector (meanScatter g nGroups v)+    RVar -> fromUnboxedVector (varScatter False g nGroups v)+    RStd -> fromUnboxedVector (varScatter True g nGroups v)+    RTop2Sum -> fromUnboxedVector (top2Scatter g nGroups v)+    RTop2Snd -> fromUnboxedVector (top2SndScatter g nGroups v)+{-# INLINE reduceTyped #-}++countScatter :: VU.Vector Int -> Int -> VU.Vector Int+countScatter g nGroups = runST $ do+    cnt <- VUM.replicate nGroups (0 :: Int)+    let n = VU.length g+        go !i+            | i >= n = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                c <- VUM.unsafeRead cnt k+                VUM.unsafeWrite cnt k (c + 1)+                go (i + 1)+    go 0+    VU.unsafeFreeze cnt++sumScatter ::+    (VU.Unbox a, Num a) => VU.Vector Int -> Int -> VU.Vector a -> VU.Vector a+sumScatter g nGroups v = runST $ do+    s <- VUM.replicate nGroups 0+    let n = VU.length v+        go !i+            | i >= n = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                cur <- VUM.unsafeRead s k+                VUM.unsafeWrite s k (cur + VU.unsafeIndex v i)+                go (i + 1)+    go 0+    VU.unsafeFreeze s+{-# INLINE sumScatter #-}++extremaScatter ::+    (VU.Unbox a) =>+    (a -> a -> a) -> a -> VU.Vector Int -> Int -> VU.Vector a -> VU.Vector a+extremaScatter combine seed g nGroups v = runST $ do+    m <- VUM.replicate nGroups seed+    let n = VU.length v+        go !i+            | i >= n = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                cur <- VUM.unsafeRead m k+                VUM.unsafeWrite m k (combine cur (VU.unsafeIndex v i))+                go (i + 1)+    go 0+    VU.unsafeFreeze m+{-# INLINE extremaScatter #-}++meanScatter ::+    (VU.Unbox a, Real a) => VU.Vector Int -> Int -> VU.Vector a -> VU.Vector Double+meanScatter g nGroups v = runST $ do+    s <- VUM.replicate nGroups (0 :: Double)+    cnt <- VUM.replicate nGroups (0 :: Int)+    scatterSumCount g v s cnt+    finalizeMean nGroups s cnt+{-# INLINE meanScatter #-}++scatterSumCount ::+    (VU.Unbox a, Real a) =>+    VU.Vector Int ->+    VU.Vector a ->+    VUM.MVector s Double ->+    VUM.MVector s Int ->+    ST s ()+scatterSumCount g v s cnt = go 0+  where+    n = VU.length v+    go !i+        | i >= n = pure ()+        | otherwise = do+            let !k = VU.unsafeIndex g i+                !x = realToFrac (VU.unsafeIndex v i)+            curS <- VUM.unsafeRead s k+            VUM.unsafeWrite s k (curS + x)+            curC <- VUM.unsafeRead cnt k+            VUM.unsafeWrite cnt k (curC + 1)+            go (i + 1)+{-# INLINE scatterSumCount #-}++finalizeMean ::+    Int -> VUM.MVector s Double -> VUM.MVector s Int -> ST s (VU.Vector Double)+finalizeMean nGroups s cnt = do+    out <- VUM.new nGroups+    let go !k+            | k >= nGroups = pure ()+            | otherwise = do+                sv <- VUM.unsafeRead s k+                c <- VUM.unsafeRead cnt k+                VUM.unsafeWrite out k (if c == 0 then 0 / 0 else sv / fromIntegral c)+                go (k + 1)+    go 0+    VU.unsafeFreeze out++varScatter ::+    (VU.Unbox a, Real a) =>+    Bool -> VU.Vector Int -> Int -> VU.Vector a -> VU.Vector Double+varScatter takeSqrt g nGroups v = runST $ do+    cnt <- VUM.replicate nGroups (0 :: Int)+    meanV <- VUM.replicate nGroups (0 :: Double)+    m2 <- VUM.replicate nGroups (0 :: Double)+    let n = VU.length v+        go !i+            | i >= n = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                    !x = realToFrac (VU.unsafeIndex v i)+                c <- VUM.unsafeRead cnt k+                mu <- VUM.unsafeRead meanV k+                mm <- VUM.unsafeRead m2 k+                let !c' = c + 1+                    !delta = x - mu+                    !mu' = mu + delta / fromIntegral c'+                    !mm' = mm + delta * (x - mu')+                VUM.unsafeWrite cnt k c'+                VUM.unsafeWrite meanV k mu'+                VUM.unsafeWrite m2 k mm'+                go (i + 1)+    go 0+    out <- VUM.new nGroups+    let fin !k+            | k >= nGroups = pure ()+            | otherwise = do+                c <- VUM.unsafeRead cnt k+                mm <- VUM.unsafeRead m2 k+                let var = if c < 2 then 0 else mm / fromIntegral (c - 1)+                VUM.unsafeWrite out k (if takeSqrt then sqrt var else var)+                fin (k + 1)+    fin 0+    VU.unsafeFreeze out+{-# INLINE varScatter #-}++top2Scatter ::+    (VU.Unbox a, Real a) => VU.Vector Int -> Int -> VU.Vector a -> VU.Vector Double+top2Scatter g nGroups v = runST $ do+    let ninf = negate (1 / 0) :: Double+    m1 <- VUM.replicate nGroups ninf+    m2 <- VUM.replicate nGroups ninf+    let n = VU.length v+        go !i+            | i >= n = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                    !x = realToFrac (VU.unsafeIndex v i)+                a1 <- VUM.unsafeRead m1 k+                if x > a1+                    then do+                        VUM.unsafeWrite m1 k x+                        VUM.unsafeWrite m2 k a1+                    else do+                        a2 <- VUM.unsafeRead m2 k+                        when (x > a2) (VUM.unsafeWrite m2 k x)+                go (i + 1)+    go 0+    out <- VUM.new nGroups+    let fin !k+            | k >= nGroups = pure ()+            | otherwise = do+                a1 <- VUM.unsafeRead m1 k+                a2 <- VUM.unsafeRead m2 k+                let s = (if isInfinite a1 then 0 else a1) + (if isInfinite a2 then 0 else a2)+                VUM.unsafeWrite out k s+                fin (k + 1)+    fin 0+    VU.unsafeFreeze out+{-# INLINE top2Scatter #-}++groupRangeBounds :: VU.Vector Int -> Int -> Int -> VU.Vector Int+groupRangeBounds offs nGroups caps = VU.create $ do+    b <- VUM.new (caps + 1)+    let !nRows = VU.unsafeIndex offs nGroups+        !per = max 1 ((nRows + caps - 1) `div` caps)+        adv !target !gg+            | gg >= nGroups = nGroups+            | VU.unsafeIndex offs gg >= target = gg+            | otherwise = adv target (gg + 1)+        go !w !prev+            | w >= caps = VUM.unsafeWrite b caps nGroups+            | otherwise = do+                let !target = min nRows (w * per)+                    !g = adv target prev+                VUM.unsafeWrite b w g+                go (w + 1) g+    VUM.unsafeWrite b 0 0+    go 1 0+    pure b++scatterReducePar ::+    Reduction -> VU.Vector Int -> VU.Vector Int -> Int -> Column -> Maybe Column+scatterReducePar red vis offs nGroups col+    | not (shouldParallelize parThreshold (VU.length vis)) || nGroups <= 1 =+        scatterReduce red (rtgFromVis vis offs nGroups) nGroups col+    | otherwise = case col of+        UnboxedColumn Nothing (v :: VU.Vector a) ->+            case testEquality (typeRep @a) (typeRep @Int) of+                Just Refl -> Just (reduceParInt red vis offs nGroups v)+                Nothing -> case testEquality (typeRep @a) (typeRep @Double) of+                    Just Refl -> Just (reduceParDouble red vis offs nGroups v)+                    Nothing -> Nothing+        p@(PackedText _ _) -> scatterReducePar red vis offs nGroups (materializePacked p)+        _ -> Nothing+{-# NOINLINE scatterReducePar #-}++{- | Monomorphic entry points: the 'testEquality' dispatch above only yields an+unsafe coercion, so a direct call to the polymorphic 'reduceParTyped' there+would stay at the abstract element type and never meet its SPECIALIZE rules;+calling through these fixed-type wrappers (the coercion lands on the argument)+does.+-}+reduceParInt ::+    Reduction -> VU.Vector Int -> VU.Vector Int -> Int -> VU.Vector Int -> Column+reduceParInt red vis offs nGroups v = reduceParTyped red vis offs nGroups v intIdent++reduceParDouble ::+    Reduction -> VU.Vector Int -> VU.Vector Int -> Int -> VU.Vector Double -> Column+reduceParDouble red vis offs nGroups v = reduceParTyped red vis offs nGroups v dblIdent++rtgFromVis :: VU.Vector Int -> VU.Vector Int -> Int -> VU.Vector Int+rtgFromVis vis offs nGroups = VU.create $ do+    let n = VU.length vis+    rtg <- VUM.new (max 1 n)+    let go !g+            | g >= nGroups = pure ()+            | otherwise = do+                let !e = VU.unsafeIndex offs (g + 1)+                    inner !pos+                        | pos >= e = pure ()+                        | otherwise = do+                            VUM.unsafeWrite rtg (VU.unsafeIndex vis pos) g+                            inner (pos + 1)+                inner (VU.unsafeIndex offs g)+                go (g + 1)+    go 0+    pure rtg++reduceParTyped ::+    forall a.+    (Columnable a, VU.Unbox a, Num a, Ord a, Real a) =>+    Reduction ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    Idents a ->+    Column+{- The SPECIALIZE pragmas matter: without them the @realToFrac@ in the+mean/var/top2 kernels survives to runtime as a dictionary call through+'Rational' (the Double->Double/Int->Double rewrite rules only fire once the+type is concrete), costing ~4x on the whole pass. -}+{-# SPECIALIZE reduceParTyped ::+    Reduction ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Int ->+    Idents Int ->+    Column+    #-}+{-# SPECIALIZE reduceParTyped ::+    Reduction ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Double ->+    Idents Double ->+    Column+    #-}+reduceParTyped red vis offs nGroups v idents =+    let !caps = capabilities+        !bounds = groupRangeBounds offs nGroups caps+     in case red of+            RCount -> fromUnboxedVector (unsafePerformIO (countPar vis offs nGroups caps bounds))+            RSum -> fromUnboxedVector (unsafePerformIO (sumPar vis offs nGroups v caps bounds))+            RMin ->+                fromUnboxedVector+                    (unsafePerformIO (extremaPar min (minSeed idents) vis offs nGroups v caps bounds))+            RMax ->+                fromUnboxedVector+                    (unsafePerformIO (extremaPar max (maxSeed idents) vis offs nGroups v caps bounds))+            RMean -> fromUnboxedVector (unsafePerformIO (meanPar vis offs nGroups v caps bounds))+            RVar ->+                fromUnboxedVector+                    (unsafePerformIO (varPar False vis offs nGroups v caps bounds))+            RStd ->+                fromUnboxedVector (unsafePerformIO (varPar True vis offs nGroups v caps bounds))+            RTop2Sum -> fromUnboxedVector (unsafePerformIO (top2Par vis offs nGroups v caps bounds))+            RTop2Snd ->+                fromUnboxedVector (unsafePerformIO (top2SndPar vis offs nGroups v caps bounds))+{-# INLINEABLE reduceParTyped #-}++{- | For each group in @[gs, ge)@, fold the group's rows (in @valueIndices@+order, i.e. ascending original-row order) into an accumulator held in+registers, then hand the final accumulator to @done@ exactly once. Keeping the+running state out of memory leaves one write per group instead of a+read-modify-write per row; the per-group fold order is unchanged, so results+stay byte-identical to the row-wise variant.+-}+overGroupsAcc ::+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    Int ->+    acc ->+    (acc -> Int -> acc) ->+    (Int -> acc -> IO ()) ->+    IO ()+overGroupsAcc vis offs gs ge seed step done = grp gs+  where+    grp !g+        | g >= ge = pure ()+        | otherwise = do+            let !e = VU.unsafeIndex offs (g + 1)+                inner !pos !acc+                    | pos >= e = pure acc+                    | otherwise = inner (pos + 1) (step acc (VU.unsafeIndex vis pos))+            acc <- inner (VU.unsafeIndex offs g) seed+            done g acc+            grp (g + 1)+{-# INLINE overGroupsAcc #-}++countPar ::+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector Int)+countPar _vis offs nGroups caps bounds = do+    out <- VUM.replicate nGroups (0 :: Int)+    parallelBounds_ caps bounds $ \gs ge ->+        let grp !g+                | g >= ge = pure ()+                | otherwise = do+                    let !c = VU.unsafeIndex offs (g + 1) - VU.unsafeIndex offs g+                    VUM.unsafeWrite out g c+                    grp (g + 1)+         in grp gs+    VU.unsafeFreeze out++sumPar ::+    (VU.Unbox a, Num a) =>+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector a)+sumPar vis offs nGroups v caps bounds = do+    out <- VUM.replicate nGroups 0+    parallelBounds_ caps bounds $ \gs ge ->+        overGroupsAcc vis offs gs ge 0 (\acc row -> acc + VU.unsafeIndex v row) $+            VUM.unsafeWrite out+    VU.unsafeFreeze out+{-# INLINE sumPar #-}++extremaPar ::+    (VU.Unbox a) =>+    (a -> a -> a) ->+    a ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector a)+extremaPar combine seed vis offs nGroups v caps bounds = do+    out <- VUM.replicate nGroups seed+    parallelBounds_ caps bounds $ \gs ge ->+        overGroupsAcc+            vis+            offs+            gs+            ge+            seed+            (\acc row -> combine acc (VU.unsafeIndex v row))+            $ VUM.unsafeWrite out+    VU.unsafeFreeze out+{-# INLINE extremaPar #-}++meanPar ::+    (VU.Unbox a, Real a) =>+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector Double)+meanPar vis offs nGroups v caps bounds = do+    out <- VUM.replicate nGroups (0 :: Double)+    parallelBounds_ caps bounds $ \gs ge ->+        let grp !g+                | g >= ge = pure ()+                | otherwise = do+                    let !e = VU.unsafeIndex offs (g + 1)+                        inner !pos !acc+                            | pos >= e = acc+                            | otherwise =+                                inner+                                    (pos + 1)+                                    (acc + realToFrac (VU.unsafeIndex v (VU.unsafeIndex vis pos)))+                        !s0 = VU.unsafeIndex offs g+                        !total = inner s0 0+                        !c = e - s0+                    VUM.unsafeWrite out g (if c == 0 then 0 / 0 else total / fromIntegral c)+                    grp (g + 1)+         in grp gs+    VU.unsafeFreeze out+{-# INLINE meanPar #-}++varPar ::+    (VU.Unbox a, Real a) =>+    Bool ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector Double)+varPar takeSqrt vis offs nGroups v caps bounds = do+    out <- VUM.replicate nGroups (0 :: Double)+    parallelBounds_ caps bounds $ \gs ge ->+        -- Per-group Welford state (count, mean, M2) carried in registers; the+        -- update order per group is the same ascending row order as before.+        let grp !g+                | g >= ge = pure ()+                | otherwise = do+                    let !e = VU.unsafeIndex offs (g + 1)+                        inner !pos !c !mu !mm+                            | pos >= e =+                                let var = if c < 2 then 0 else mm / fromIntegral (c - 1)+                                 in if takeSqrt then sqrt var else var+                            | otherwise =+                                let !x = realToFrac (VU.unsafeIndex v (VU.unsafeIndex vis pos))+                                    !c' = c + 1+                                    !delta = x - mu+                                    !mu' = mu + delta / fromIntegral c'+                                    !mm' = mm + delta * (x - mu')+                                 in inner (pos + 1) c' mu' mm'+                        !res = inner (VU.unsafeIndex offs g) (0 :: Int) 0 0+                    VUM.unsafeWrite out g res+                    grp (g + 1)+         in grp gs+    VU.unsafeFreeze out+{-# INLINE varPar #-}++top2Par ::+    (VU.Unbox a, Real a) =>+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector Double)+top2Par vis offs nGroups v caps bounds = do+    let ninf = negate (1 / 0) :: Double+    out <- VUM.replicate nGroups (0 :: Double)+    parallelBounds_ caps bounds $ \gs ge ->+        -- The (largest, second-largest) pair carried in registers per group.+        let grp !g+                | g >= ge = pure ()+                | otherwise = do+                    let !e = VU.unsafeIndex offs (g + 1)+                        inner !pos !a1 !a2+                            | pos >= e =+                                (if isInfinite a1 then 0 else a1)+                                    + (if isInfinite a2 then 0 else a2)+                            | otherwise =+                                let !x = realToFrac (VU.unsafeIndex v (VU.unsafeIndex vis pos))+                                 in if x > a1+                                        then inner (pos + 1) x a1+                                        else inner (pos + 1) a1 (max a2 x)+                        !res = inner (VU.unsafeIndex offs g) ninf ninf+                    VUM.unsafeWrite out g res+                    grp (g + 1)+         in grp gs+    VU.unsafeFreeze out+{-# INLINE top2Par #-}++{- | Second-largest value per group: the same (largest, second-largest)+register pair as 'top2Par', finalized to the second max alone. Size-1 groups+finalize the @-inf@ seed to NaN (documented; see+'DataFrame.Internal.AggKernel.top2SndScatter').+-}+top2SndPar ::+    (VU.Unbox a, Real a) =>+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector Double)+top2SndPar vis offs nGroups v caps bounds = do+    let ninf = negate (1 / 0) :: Double+    out <- VUM.replicate nGroups (0 :: Double)+    parallelBounds_ caps bounds $ \gs ge ->+        -- The (largest, second-largest) pair carried in registers per group.+        let grp !g+                | g >= ge = pure ()+                | otherwise = do+                    let !e = VU.unsafeIndex offs (g + 1)+                        inner !pos !a1 !a2+                            | pos >= e = if isInfinite a2 then 0 / 0 else a2+                            | otherwise =+                                let !x = realToFrac (VU.unsafeIndex v (VU.unsafeIndex vis pos))+                                 in if x > a1+                                        then inner (pos + 1) x a1+                                        else inner (pos + 1) a1 (max a2 x)+                        !res = inner (VU.unsafeIndex offs g) ninf ninf+                    VUM.unsafeWrite out g res+                    grp (g + 1)+         in grp gs+    VU.unsafeFreeze out+{-# INLINE top2SndPar #-}++-------------------------------------------------------------------------------+-- Parallel fused max(a) - min(b) (Q7 at wide group domains)+-------------------------------------------------------------------------------++{- | Fused @max a - min b@ over the group-range layout: ONE traversal of+@valueIndices@ accumulating both extrema, parallel by disjoint group range with+no cross-worker merge. min/max are order-independent, so the result is+byte-identical to running the two gather extrema passes separately; the fusion+halves the index traffic. 'Nothing' below the parallel threshold or unless both+columns are clean unboxed and same-typed (Int/Int keeps the Int result of the+interpreter; Double/Double the Double one) — the caller then keeps its two-pass+fallback.+-}+maxMinusMinScatterPar ::+    VU.Vector Int -> VU.Vector Int -> Int -> Column -> Column -> Maybe Column+maxMinusMinScatterPar vis offs nGroups ca cb+    | not (shouldPar (VU.length vis)) || nGroups <= 1 = Nothing+    | otherwise = case (ca, cb) of+        ( UnboxedColumn Nothing (va :: VU.Vector x)+            , UnboxedColumn Nothing (vb :: VU.Vector y)+            )+                | Just Refl <- testEquality (typeRep @x) (typeRep @Int)+                , Just Refl <- testEquality (typeRep @y) (typeRep @Int) ->+                    Just (maxMinusMinParInt vis offs nGroups va vb caps bounds)+                | Just Refl <- testEquality (typeRep @x) (typeRep @Double)+                , Just Refl <- testEquality (typeRep @y) (typeRep @Double) ->+                    Just (maxMinusMinParDbl vis offs nGroups va vb caps bounds)+        _ -> Nothing+  where+    !caps = capabilities+    !bounds = groupRangeBounds offs nGroups caps+{-# NOINLINE maxMinusMinScatterPar #-}++{- | Monomorphic entry points (see 'reduceParInt' for why the 'testEquality'+dispatch needs them).+-}+maxMinusMinParInt ::+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Int ->+    Column+maxMinusMinParInt vis offs nGroups va vb caps bounds =+    fromUnboxedVector+        ( unsafePerformIO+            (maxMinusMinPar minBound maxBound vis offs nGroups va vb caps bounds)+        )+{-# NOINLINE maxMinusMinParInt #-}++maxMinusMinParDbl ::+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Double ->+    VU.Vector Double ->+    Int ->+    VU.Vector Int ->+    Column+maxMinusMinParDbl vis offs nGroups va vb caps bounds =+    fromUnboxedVector+        ( unsafePerformIO+            (maxMinusMinPar (negate (1 / 0)) (1 / 0) vis offs nGroups va vb caps bounds)+        )+{-# NOINLINE maxMinusMinParDbl #-}++maxMinusMinPar ::+    (VU.Unbox a, Num a, Ord a) =>+    a ->+    a ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector a ->+    VU.Vector a ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector a)+{-# SPECIALIZE maxMinusMinPar ::+    Int ->+    Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector Int)+    #-}+{-# SPECIALIZE maxMinusMinPar ::+    Double ->+    Double ->+    VU.Vector Int ->+    VU.Vector Int ->+    Int ->+    VU.Vector Double ->+    VU.Vector Double ->+    Int ->+    VU.Vector Int ->+    IO (VU.Vector Double)+    #-}+maxMinusMinPar maxSeed minSeed vis offs nGroups va vb caps bounds = do+    out <- VUM.new nGroups+    parallelBounds_ caps bounds $ \gs ge ->+        -- Both extrema carried in registers per group; one traversal of the+        -- shared index slice reads both value columns.+        let grp !g+                | g >= ge = pure ()+                | otherwise = do+                    let !e = VU.unsafeIndex offs (g + 1)+                        inner !pos !mx !mn+                            | pos >= e = mx - mn+                            | otherwise =+                                let !row = VU.unsafeIndex vis pos+                                 in inner+                                        (pos + 1)+                                        (max mx (VU.unsafeIndex va row))+                                        (min mn (VU.unsafeIndex vb row))+                        !res = inner (VU.unsafeIndex offs g) maxSeed minSeed+                    VUM.unsafeWrite out g res+                    grp (g + 1)+         in grp gs+    VU.unsafeFreeze out++-------------------------------------------------------------------------------+-- Streaming (rowToGroup-scatter) kernels: no valueIndices, no placement pass+-------------------------------------------------------------------------------++{- | Group-count cap for the FUSED streaming rtg-scatter kernels+('DataFrame.Internal.Aggregation.Kernel.Moments.momentStreamPar',+'DataFrame.Internal.Aggregation.Kernel.Fused.runFusedAggs'). Above+'DataFrame.Internal.Grouping.Direct.directThreshold' the per-worker accumulator+arrays overflow cache, so a SINGLE streaming reduction loses to a gather pass —+the per-expression dispatch keeps that threshold. A fused multi-reduction pass+amortizes those misses across all its reductions AND avoids the deferred+@valueIndices@ placement entirely, which flips the comparison (measured at 1e6+groups / 1e8 rows on -N16: rowToGroup 0.4s + fused 3-sum stream 1.4s, against+placement 1.1s + fused gather 1.05s), so the fused cap extends to+'directGroupThreshold' — every direct-grouped frame can stream. Wider+groupings are necessarily hash-path (eager @valueIndices@) and use the fused+GATHER kernel ('DataFrame.Internal.Aggregation.Kernel.Fused.runGatherAggs')+instead. Memory: @capabilities * nGroups@ words per accumulator array, at most+~128MB transient at -N16.+-}+streamGroupCap :: Int+streamGroupCap = 1048576++{- | Near-equal contiguous slices of the group domain for parallel merges.+Below 4096 groups (or single-capability) the merge stays on one thread.+-}+groupSlices :: Int -> [(Int, Int)]+groupSlices = chunksFor 4096++{- | Second-largest value per group: the same (largest, second-largest)+accumulator pair as 'top2Scatter', but the finalize returns the second max+alone. A group of size 1 (or 0) leaves the @-inf@ seed in the second slot, so+its output is NaN — documented behaviour (the db-benchmark Q8 data has no+size-1 @id6@ groups). Like 'top2Scatter''s @-inf -> 0@ guard, an actual+infinite data value in the second slot is indistinguishable from the seed.+-}+top2SndScatter ::+    (VU.Unbox a, Real a) => VU.Vector Int -> Int -> VU.Vector a -> VU.Vector Double+top2SndScatter g nGroups v = runST $ do+    let ninf = negate (1 / 0) :: Double+    m1 <- VUM.replicate nGroups ninf+    m2 <- VUM.replicate nGroups ninf+    let n = VU.length v+        go !i+            | i >= n = pure ()+            | otherwise = do+                let !k = VU.unsafeIndex g i+                    !x = realToFrac (VU.unsafeIndex v i)+                a1 <- VUM.unsafeRead m1 k+                if x > a1+                    then do+                        VUM.unsafeWrite m1 k x+                        VUM.unsafeWrite m2 k a1+                    else do+                        a2 <- VUM.unsafeRead m2 k+                        when (x > a2) (VUM.unsafeWrite m2 k x)+                go (i + 1)+    go 0+    out <- VUM.new nGroups+    let fin !k+            | k >= nGroups = pure ()+            | otherwise = do+                a2 <- VUM.unsafeRead m2 k+                VUM.unsafeWrite out k (if isInfinite a2 then 0 / 0 else a2)+                fin (k + 1)+    fin 0+    VU.unsafeFreeze out+{-# INLINE top2SndScatter #-}

@@ -0,0 +1,222 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE ExplicitNamespaces #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++{- | The aggregation fast-path planner. 'planAgg' recognises a supported+aggregate shape over a clean unboxed Int/Double column and returns an 'AggPlan';+'planMoments' recognises the six-fold regression shape and returns a+'MomentPlan'. Planning only — the kernels live under "Kernel".+-}+module DataFrame.Internal.Aggregation.Plan (+    AggPlan (..),+    planAgg,+    MomentPlan (..),+    planMoments,+) where++import qualified Data.Map.Strict as M+import qualified Data.Text as T+import Data.Type.Equality (TestEquality (..), type (:~:) (Refl))+import qualified Data.Vector.Unboxed as VU++import DataFrame.Internal.Aggregation.Reduction (Reduction (..))+import DataFrame.Internal.Column (Column (..))+import DataFrame.Internal.DataFrame (+    DataFrame (derivingExpressions),+    GroupedDataFrame (..),+    getColumn,+ )+import DataFrame.Internal.Expression (+    AggStrategy (..),+    BinaryOp (binaryCommutative, binaryName),+    Expr (..),+    UExpr (..),+ )+import Type.Reflection (Typeable, typeRep)++{- | The plan 'planAgg' produces for a recognised output expression. The median+plan carries only the column name (the holistic grouped sort lives in the+operations layer, where @vector-algorithms@ is available).+-}+data AggPlan+    = -- | A single scatter reduction over one named column.+      PlanScatter Reduction T.Text+    | -- | @max a - min b@ (Q7): two scatters then a vectorized combine.+      PlanMaxMinusMin T.Text T.Text+    | -- | Holistic median over one named column.+      PlanMedian T.Text++{- | Inspect a named output expression; return @Just plan@ on a recognised shape+over a present clean column, else 'Nothing'. Nullable or non-Int/Double columns+are rejected here so the scatter only sees a clean unboxed vector.+-}+planAgg :: GroupedDataFrame -> UExpr -> Maybe AggPlan+planAgg gdf (UExpr (expr :: Expr a)) = case expr of+    Agg (FoldAgg tag _ _) (Col name) -> foldPlan tag name+    Agg (MergeAgg tag _ _ _ _) (Col name) -> mergePlan tag name+    Agg (CollectAgg tag _) (Col name) -> collectPlan tag name+    Binary+        op+        (Agg (FoldAgg lt Nothing _) (Col a))+        (Agg (FoldAgg rt Nothing _) (Col b)) ->+            if binaryName op == "sub" && lt == "maximum" && rt == "minimum"+                then requireBoth a b (PlanMaxMinusMin a b)+                else Nothing+    _ -> Nothing+  where+    foldPlan tag name = case tag of+        "sum" -> require name (PlanScatter RSum name)+        "minimum" -> require name (PlanScatter RMin name)+        "maximum" -> require name (PlanScatter RMax name)+        _ -> Nothing+    mergePlan tag name = case tag of+        "mean" -> outputType @Double >> require name (PlanScatter RMean name)+        "count" -> outputType @Int >> require name (PlanScatter RCount name)+        _ -> Nothing+    outputType :: forall t. (Typeable t) => Maybe ()+    outputType = case testEquality (typeRep @a) (typeRep @t) of+        Just Refl -> Just ()+        Nothing -> Nothing+    collectPlan tag name = case tag of+        "stddev" -> require name (PlanScatter RStd name)+        "variance" -> require name (PlanScatter RVar name)+        "top2Sum" -> require name (PlanScatter RTop2Sum name)+        "top2Snd" -> require name (PlanScatter RTop2Snd name)+        "median" -> require name (PlanMedian name)+        _ -> Nothing+    require name plan = colUnboxedNumeric name >> Just plan+    requireBoth a b plan = colUnboxedNumeric a >> colUnboxedNumeric b >> Just plan+    colUnboxedNumeric name = case getColumn name (fullDataframe gdf) of+        Just c | isUnboxedNumeric c -> Just ()+        _ -> Nothing++-- | The matcher only fires on non-null unboxed Int/Double columns.+isUnboxedNumeric :: Column -> Bool+isUnboxedNumeric = \case+    UnboxedColumn Nothing (_ :: VU.Vector a) ->+        case testEquality (typeRep @a) (typeRep @Int) of+            Just Refl -> True+            Nothing -> case testEquality (typeRep @a) (typeRep @Double) of+                Just Refl -> True+                Nothing -> False+    _ -> False++{- | A recognised moment (Q9 regression) aggregate group: six output columns that+form the sufficient statistics of two base columns @x@ and @y@. The caller runs+'momentScatter' once and binds each output name to a field of the result.+-}+data MomentPlan = MomentPlan+    { mpColX :: T.Text+    , mpColY :: T.Text+    , mpNName :: T.Text+    , mpSxName :: T.Text+    , mpSyName :: T.Text+    , mpSxxName :: T.Text+    , mpSyyName :: T.Text+    , mpSxyName :: T.Text+    }++{- | The shape of a sum's argument once unary coercions are peeled and derived+columns are resolved through @derivingExpressions@: either linear in one base+column or the product of two base columns (sorted).+-}+data Term+    = Lin T.Text+    | Prod T.Text T.Text+    deriving (Eq, Ord, Show)++{- | Recognise the moment shape across a whole @aggregate@ list: exactly+@count@, @sum(x)@, @sum(y)@, @sum(x*x)@, @sum(y*y)@, @sum(x*y)@ over two distinct+clean unboxed base columns. 'Nothing' on any other set.+-}+planMoments :: GroupedDataFrame -> [(T.Text, UExpr)] -> Maybe MomentPlan+planMoments gdf aggs+    | length aggs /= 6 = Nothing+    | otherwise = do+        let exprs = derivingExpressions (fullDataframe gdf)+        roles <- traverse (classify exprs) aggs+        let names = M.fromList [(r, nm) | (nm, r) <- roles]+        nName <- M.lookup RoleN names+        (x, y) <- pickBaseColumns roles+        sxName <- M.lookup (RoleLin x) names+        syName <- M.lookup (RoleLin y) names+        sxxName <- M.lookup (RoleProd x x) names+        syyName <- M.lookup (RoleProd y y) names+        sxyName <- M.lookup (RoleProd x y) names+        _ <- if x /= y then Just () else Nothing+        _ <- colUnboxedNumeric x+        _ <- colUnboxedNumeric y+        pure+            MomentPlan+                { mpColX = x+                , mpColY = y+                , mpNName = nName+                , mpSxName = sxName+                , mpSyName = syName+                , mpSxxName = sxxName+                , mpSyyName = syyName+                , mpSxyName = sxyName+                }+  where+    colUnboxedNumeric name = case getColumn name (fullDataframe gdf) of+        Just c | isUnboxedNumeric c -> Just ()+        _ -> Nothing++-- | The output role each named aggregation plays in the moment shape.+data Role+    = RoleN+    | RoleLin T.Text+    | RoleProd T.Text T.Text+    deriving (Eq, Ord, Show)++-- | Tag a single named aggregation with its moment role, or reject the group.+classify :: M.Map T.Text UExpr -> (T.Text, UExpr) -> Maybe (T.Text, Role)+classify exprs (name, UExpr expr) = case expr of+    Agg (MergeAgg "count" _ _ _ _) _ -> Just (name, RoleN)+    Agg (FoldAgg "sum" _ _) arg -> (\t -> (name, termRole t)) <$> resolveTerm exprs (UExpr arg)+    _ -> Nothing++termRole :: Term -> Role+termRole (Lin a) = RoleLin a+termRole (Prod a b) = RoleProd a b++{- | Resolve a (sum-argument) expression to its 'Term'. Peels @toDouble@-style+unary coercions, follows a derived column to its stored expression, and+recognises a commutative product of two linear terms.+-}+resolveTerm :: M.Map T.Text UExpr -> UExpr -> Maybe Term+resolveTerm exprs = go (8 :: Int)+  where+    go 0 _ = Nothing+    go fuel (UExpr e) = case e of+        Col nm -> case M.lookup nm exprs of+            Just ue -> go (fuel - 1) ue+            Nothing -> Just (Lin nm)+        Unary _ inner -> go (fuel - 1) (UExpr inner)+        Binary op l r+            | binaryName op == "mult" && binaryCommutative op -> do+                Lin a <- go (fuel - 1) (UExpr l)+                Lin b <- go (fuel - 1) (UExpr r)+                Just (sortProd a b)+        _ -> Nothing++-- | Products are unordered: store the pair sorted so @x*y@ and @y*x@ unify.+sortProd :: T.Text -> T.Text -> Term+sortProd a b+    | a <= b = Prod a b+    | otherwise = Prod b a++{- | From the classified roles, find the unordered pair of base columns that the+linear sums name. There must be exactly two distinct linear-sum columns.+-}+pickBaseColumns :: [(T.Text, Role)] -> Maybe (T.Text, T.Text)+pickBaseColumns roles =+    case lins of+        [a, b] | a /= b -> Just (a, b)+        _ -> Nothing+  where+    lins = M.keys (M.fromList [(c, ()) | (_, RoleLin c) <- roles])

@@ -0,0 +1,44 @@+{-# LANGUAGE ScopedTypeVariables #-}++{- | The vocabulary shared by the aggregation planner and the aggregation+kernels: the set of recognised reductions, and the admission gate that decides+whether a column may enter a Double-valued fast path.++This module holds no kernel and no policy, so the planner can depend on it+without depending on an implementation.+-}+module DataFrame.Internal.Aggregation.Reduction (+    Reduction (..),+    cleanDoubleVector,+) where++import qualified Data.Vector.Unboxed as VU++import DataFrame.Internal.Column (+    Column (..),+    materializePacked,+ )+import DataFrame.Internal.Column.Conversion (toDoubleVector)++{- | A recognised fast-path reduction over a single value column. The element+type (Int vs Double) is resolved at scatter time; sum/min/max preserve the+column's element type, everything else produces a Double column.+-}+data Reduction+    = RSum+    | RCount+    | RMin+    | RMax+    | RMean+    | RStd+    | RVar+    | RTop2Sum+    | RTop2Snd+    deriving (Eq, Show)++cleanDoubleVector :: Column -> Maybe (VU.Vector Double)+cleanDoubleVector col = case col of+    UnboxedColumn Nothing _ -> either (const Nothing) Just (toDoubleVector col)+    BoxedColumn Nothing _ -> either (const Nothing) Just (toDoubleVector col)+    p@(PackedText _ _) -> cleanDoubleVector (materializePacked p)+    _ -> Nothing

@@ -0,0 +1,113 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE MagicHash #-}++{- | A poor-man's hash used by 'DataFrame.Internal.Grouping' to bucket rows+without depending on @hashable@. Each value is folded into an 'Int' with an+FxHash-style step (rotate, xor, multiply); small and not cryptographic.+-}+module DataFrame.Internal.Algorithms.Hash (+    fnvOffset,+    nullSalt,+    mixInt,+    mixDouble,+    mixBool,+    mixChar,+    mixText,+    mixBytes,+    mixShow,+) where++import Data.Bits (rotateL, unsafeShiftL, unsafeShiftR, xor)+import Data.Char (ord)+import qualified Data.Text as T+import qualified Data.Text.Array as A+#if MIN_VERSION_text(2,1,0)+import Data.Array.Byte (ByteArray (ByteArray))+#else+import Data.Text.Array (Array (ByteArray))+#endif+import Data.Text.Internal (Text (Text))+import GHC.Exts (Int (I#), indexWord8Array#, indexWord8ArrayAsWord64#)+import GHC.Word (Word64 (W64#), Word8 (W8#))++{- | FNV-1a 64-bit offset basis (used as the initial accumulator).+The literal is unsigned and exceeds 'Int' range, so we round-trip through+'Word64' to get the well-defined two's-complement bit pattern.+-}+fnvOffset :: Int+fnvOffset = fromIntegral (0xcbf29ce484222325 :: Word64)++-- | FNV-1a 64-bit prime.+fnvPrime :: Int+fnvPrime = 0x00000100000001b3++{- | Sentinel mixed in for a /null/ slot, so @Nothing@ does not hash the same as+a present value with equal bits (e.g. @Just 0@). A fixed distinctive constant+keeps null hashing deterministic; a real value equal to it collides only rarely.+-}+nullSalt :: Int+nullSalt = fromIntegral (0x9E3779B97F4A7C15 :: Word64)++{- | Mix an 'Int' into the accumulator with an FxHash-style step. The rotate+diffuses each value's bits before the next is folded in, avoiding the structured+collisions a plain xor-then-multiply produces on small/adjacent group keys.+-}+mixInt :: Int -> Int -> Int+mixInt acc x = (rotateL acc 13 `xor` x) * fnvPrime+{-# INLINE mixInt #-}++{- | Mix a 'Double' into the accumulator. Loses sub-millisecond precision+but matches the bucketing the old hashable-based code used.+-}+mixDouble :: Int -> Double -> Int+mixDouble acc d = mixInt acc (floor (d * 1000))+{-# INLINE mixDouble #-}++mixBool :: Int -> Bool -> Int+mixBool acc b = mixInt acc (if b then 1 else 0)+{-# INLINE mixBool #-}++mixChar :: Int -> Char -> Int+mixChar acc = mixInt acc . ord+{-# INLINE mixChar #-}++{- | Mix a 'T.Text' value into the accumulator over its raw UTF-8 bytes, eight at+a time. Reading a whole 'Word64' per step cuts the multiply count ~8x on long+keys while staying collision-equivalent (UTF-8 is injective).+-}+mixText :: Int -> T.Text -> Int+mixText !acc (Text arr off len) = mixBytes acc arr off len+{-# INLINE mixText #-}++{- | Mix a raw UTF-8 byte slice @[off, off+len)@ of a 'Data.Text.Array.Array'+into the accumulator, eight bytes at a time. The shared kernel behind+'mixText' and the packed-text hash path, so the two never drift.+-}+mixBytes :: Int -> A.Array -> Int -> Int -> Int+mixBytes !acc arr off len = goBytes (goWords acc off) wordsEnd+  where+    !(ByteArray ba) = arr+    !nWords = len `unsafeShiftR` 3+    !wordsEnd = off + (nWords `unsafeShiftL` 3)+    !end = off + len+    goWords !h !i+        | i >= wordsEnd = h+        | otherwise =+            let !(I# i#) = i+                !w = fromIntegral (W64# (indexWord8ArrayAsWord64# ba i#)) :: Int+             in goWords (mixInt h w) (i + 8)+    goBytes !h !i+        | i >= end = h+        | otherwise =+            let !(I# i#) = i+                !b = fromIntegral (W8# (indexWord8Array# ba i#)) :: Int+             in goBytes (mixInt h b) (i + 1)+{-# INLINE mixBytes #-}++{- | Fallback for arbitrary 'Show'-able values. Slower but covers types+without a dedicated combinator (e.g. 'Day', 'UTCTime').+-}+mixShow :: (Show a) => Int -> a -> Int+mixShow acc = mixText acc . T.pack . show+{-# INLINE mixShow #-}

@@ -0,0 +1,101 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ScopedTypeVariables #-}++{- | Stable rank of a set of group representatives by ascending unsigned hash+order. Shared by the sequential and parallel group-by canonical-ordering steps+so they stay bit-for-bit identical. @O(ng)@ stable LSD radix sort.+-}+module DataFrame.Internal.Algorithms.Rank.Radix (+    rankByHash,+    sortKey,+) where++import Control.Monad (when)+import Control.Monad.Primitive (PrimMonad)+import Data.Bits (unsafeShiftR, (.&.))+import qualified Data.Vector.Unboxed as VU+import qualified Data.Vector.Unboxed.Mutable as VUM+import Data.Word (Word64)++{- | Unsigned sort key of a hash: ascending 'Word64' order of @sortKey h@ equals+ascending signed-'Int' order of @h@. Reinterpreted to 'Int' for the byte-wise+radix passes (the byte mask makes the sign extension irrelevant).+-}+sortKey :: Int -> Int+sortKey h = fromIntegral (fromIntegral h + 0x8000000000000000 :: Word64)+{-# INLINE sortKey #-}++-- | See the module header. @readHash@ supplies the hash of local group @gid@.+rankByHash ::+    forall m. (PrimMonad m) => (Int -> m Int) -> Int -> m (VU.Vector Int)+rankByHash readHash ng = do+    rankM <- VUM.new (max 1 ng)+    if ng <= 1+        then when (ng == 1) (VUM.unsafeWrite rankM 0 0)+        else do+            keysA <- VUM.new ng+            orderA <- VUM.new ng+            let seed !i+                    | i >= ng = pure ()+                    | otherwise = do+                        h <- readHash i+                        VUM.unsafeWrite keysA i (sortKey h)+                        VUM.unsafeWrite orderA i i+                        seed (i + 1)+            seed 0+            keysB <- VUM.new ng+            orderB <- VUM.new ng+            counts <- VUM.new 256+            let pass ::+                    Int ->+                    VUM.MVector (VUM.PrimState m) Int ->+                    VUM.MVector (VUM.PrimState m) Int ->+                    VUM.MVector (VUM.PrimState m) Int ->+                    VUM.MVector (VUM.PrimState m) Int ->+                    m ()+                pass !shiftBits !srcK !srcO !dstK !dstO = do+                    VUM.set counts 0+                    let count !i+                            | i >= ng = pure ()+                            | otherwise = do+                                k <- VUM.unsafeRead srcK i+                                let !b = (k `unsafeShiftR` shiftBits) .&. 0xff+                                VUM.unsafeRead counts b >>= VUM.unsafeWrite counts b . (+ 1)+                                count (i + 1)+                    count 0+                    let scan !b !acc+                            | b >= 256 = pure ()+                            | otherwise = do+                                c <- VUM.unsafeRead counts b+                                VUM.unsafeWrite counts b acc+                                scan (b + 1) (acc + c)+                    scan 0 0+                    let place !i+                            | i >= ng = pure ()+                            | otherwise = do+                                k <- VUM.unsafeRead srcK i+                                o <- VUM.unsafeRead srcO i+                                let !b = (k `unsafeShiftR` shiftBits) .&. 0xff+                                pos <- VUM.unsafeRead counts b+                                VUM.unsafeWrite counts b (pos + 1)+                                VUM.unsafeWrite dstK pos k+                                VUM.unsafeWrite dstO pos o+                                place (i + 1)+                    place 0+            pass 0 keysA orderA keysB orderB+            pass 8 keysB orderB keysA orderA+            pass 16 keysA orderA keysB orderB+            pass 24 keysB orderB keysA orderA+            pass 32 keysA orderA keysB orderB+            pass 40 keysB orderB keysA orderA+            pass 48 keysA orderA keysB orderB+            pass 56 keysB orderB keysA orderA+            let inv !r+                    | r >= ng = pure ()+                    | otherwise = do+                        g <- VUM.unsafeRead orderA r+                        VUM.unsafeWrite rankM g r+                        inv (r + 1)+            inv 0+    VU.unsafeFreeze rankM+{-# INLINEABLE rankByHash #-}

@@ -0,0 +1,332 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ScopedTypeVariables #-}++{- | Parallel stable sort of row indices by ascending unsigned order of a per-row+'Int' hash, used by the join build side. A counting sort buckets rows into+key-ordered partitions that workers LSD-radix-sort in parallel, with no merge step.+-}+module DataFrame.Internal.Algorithms.Sort.Radix.Parallel (+    parSortByHash,+    parSortThreshold,+) where++import Control.Concurrent (getNumCapabilities)+import Control.Monad (forM_, when)+import Data.Bits (countLeadingZeros, unsafeShiftR, (.&.))+import qualified Data.Vector.Unboxed as VU+import qualified Data.Vector.Unboxed.Mutable as VUM+import Data.Word (Word64)+import DataFrame.Internal.Algorithms.Rank.Radix (sortKey)+import DataFrame.Internal.Control.Concurrent (+    capabilities,+    forkJoin,+    forkJoin_,+    pooledIndices,+ )+import System.IO.Unsafe (unsafePerformIO)++{- | Below this many rows the partition/fork overhead is not worth it; the+caller's sequential LSD radix path is used instead.+-}+parSortThreshold :: Int+parSortThreshold = 500000++{- | Top-bits partition index of a hash: the high @64 - shift@ bits of its+unsigned 'sortKey'. Ascending partition order equals ascending key order.+-}+partIx :: Int -> Int -> Int+partIx shift h = fromIntegral ((fromIntegral (sortKey h) :: Word64) `unsafeShiftR` shift)+{-# INLINE partIx #-}++-- | Number of partitions: a power of two, at least @4 * caps@, floored at 256.+numPartitionsFor :: Int -> Int+numPartitionsFor caps = go 1+  where+    target = max 256 (4 * caps)+    go p+        | p >= target = p+        | otherwise = go (p * 2)++-- | @floor (log2 x)@ for a power-of-two @x@.+intLog2 :: Int -> Int+intLog2 x = 63 - countLeadingZeros x+{-# INLINE intLog2 #-}++{- | Parallel stable sort of @[0, n)@ by ascending unsigned hash order. See the+module header for the ordering contract.+-}+parSortByHash :: Int -> VU.Vector Int -> (VU.Vector Int, VU.Vector Int)+parSortByHash n hashes+    | n <= 1 =+        (hashes, VU.enumFromN 0 n)+    | n < parSortThreshold || capabilities <= 1 =+        seqSortByHash n hashes+    | otherwise = unsafePerformIO (parSortByHashIO n hashes)+{-# NOINLINE parSortByHash #-}++-------------------------------------------------------------------------------+-- Sequential LSD radix sort (also the per-partition worker kernel)+-------------------------------------------------------------------------------++{- | Stable LSD radix sort of @[0, n)@ by ascending 'sortKey' of their hash, 8+bits per pass over the full 64-bit key. Returns @(sortedHashes, sortedIndices)@.+-}+seqSortByHash :: Int -> VU.Vector Int -> (VU.Vector Int, VU.Vector Int)+seqSortByHash n hashes = unsafePerformIO $ do+    keysA <- VUM.new n+    orderA <- VUM.new n+    let seed !i+            | i >= n = pure ()+            | otherwise = do+                VUM.unsafeWrite keysA i (sortKey (VU.unsafeIndex hashes i))+                VUM.unsafeWrite orderA i i+                seed (i + 1)+    seed 0+    keysB <- VUM.new n+    orderB <- VUM.new n+    radixPasses n keysA orderA keysB orderB+    order <- VU.unsafeFreeze orderA+    pure (VU.unsafeBackpermute hashes order, order)++{- | Run all eight stable 8-bit LSD passes, ping-ponging between the two+key/order buffer pairs so the sorted order lands back in @(keysA, orderA)@.+@keysA[i]@ must already hold @sortKey (hash of orderA[i])@ on entry.+-}+radixPasses ::+    Int ->+    VUM.IOVector Int ->+    VUM.IOVector Int ->+    VUM.IOVector Int ->+    VUM.IOVector Int ->+    IO ()+radixPasses = radixPassesN 8++{- | Run the first @np@ stable 8-bit LSD passes (bits @0 .. 8*np-1@),+ping-ponging between the buffer pairs. For odd @np@ the sorted order lands in+@(keysB, orderB)@, for even in @(keysA, orderA)@. Callers whose rows share+their top bytes (per-partition sorts partitioned on the top byte) can pass+@np = 7@: the eighth pass would be a stable identity copy.+-}+radixPassesN ::+    Int ->+    Int ->+    VUM.IOVector Int ->+    VUM.IOVector Int ->+    VUM.IOVector Int ->+    VUM.IOVector Int ->+    IO ()+radixPassesN np n keysA orderA keysB orderB = do+    counts <- VUM.new 256+    let pass ::+            Int ->+            VUM.IOVector Int ->+            VUM.IOVector Int ->+            VUM.IOVector Int ->+            VUM.IOVector Int ->+            IO ()+        pass !shiftBits !srcK !srcO !dstK !dstO = do+            VUM.set counts 0+            let count !i+                    | i >= n = pure ()+                    | otherwise = do+                        k <- VUM.unsafeRead srcK i+                        let !b = (k `unsafeShiftR` shiftBits) .&. 0xff+                        VUM.unsafeRead counts b >>= VUM.unsafeWrite counts b . (+ 1)+                        count (i + 1)+            count 0+            let scan !b !acc+                    | b >= 256 = pure ()+                    | otherwise = do+                        c <- VUM.unsafeRead counts b+                        VUM.unsafeWrite counts b acc+                        scan (b + 1) (acc + c)+            scan 0 0+            let place !i+                    | i >= n = pure ()+                    | otherwise = do+                        k <- VUM.unsafeRead srcK i+                        o <- VUM.unsafeRead srcO i+                        let !b = (k `unsafeShiftR` shiftBits) .&. 0xff+                        pos <- VUM.unsafeRead counts b+                        VUM.unsafeWrite counts b (pos + 1)+                        VUM.unsafeWrite dstK pos k+                        VUM.unsafeWrite dstO pos o+                        place (i + 1)+            place 0+        run !k+            | k >= np = pure ()+            | even k = pass (8 * k) keysA orderA keysB orderB >> run (k + 1)+            | otherwise = pass (8 * k) keysB orderB keysA orderA >> run (k + 1)+    run 0++-------------------------------------------------------------------------------+-- Parallel path: counting-sort partition, then per-partition sort in parallel+-------------------------------------------------------------------------------++parSortByHashIO :: Int -> VU.Vector Int -> IO (VU.Vector Int, VU.Vector Int)+parSortByHashIO n hashes = do+    caps <- getNumCapabilities+    let !p = numPartitionsFor caps+        !shift = 64 - intLog2 p+    (partStart, partRows, partHashes) <- partitionRows n hashes p shift+    outOrder <- VUM.new n+    outKeys <- VUM.new n+    sortPartitions caps p partStart partRows partHashes outOrder outKeys+    order <- VU.unsafeFreeze outOrder+    sortedHashes <- VU.unsafeFreeze outKeys+    pure (sortedHashes, order)++{- | Bucket every row index into its top-bits partition by a counting sort.+Returns the exclusive prefix sum @partStart@ (length @p+1@, @partStart[p] == n@),+the row indices laid out partition-by-partition in ascending key order, and+each sorted position's hash in the same layout (so downstream passes read+hashes sequentially instead of a random @hashes[row]@ gather per row).++Runs chunked across capabilities: per-chunk partition histograms are prefix+summed (in chunk order) into disjoint per-chunk write cursors, so the scatter+threads never contend and each partition keeps its rows in ascending original+row order — bit-for-bit the sequential counting sort's layout.+-}+partitionRows ::+    Int ->+    VU.Vector Int ->+    Int ->+    Int ->+    IO (VU.Vector Int, VU.Vector Int, VU.Vector Int)+partitionRows n hashes p shift = do+    caps <- getNumCapabilities+    let chunks = rowChunks caps n+    cursors <- forkJoin [histChunk hashes p shift lo hi | (lo, hi) <- chunks]+    -- Exclusive prefix over partitions (outer) and chunks (inner): partStart+    -- from the totals, and each chunk's histogram rewritten into its cursor.+    partStartM <- VUM.new (p + 1)+    let seed !pp !acc+            | pp >= p = VUM.unsafeWrite partStartM p acc+            | otherwise = do+                VUM.unsafeWrite partStartM pp acc+                let inner [] !a = pure a+                    inner (cur : rest) !a = do+                        t <- VUM.unsafeRead cur pp+                        VUM.unsafeWrite cur pp a+                        inner rest (a + t)+                acc' <- inner cursors acc+                seed (pp + 1) acc'+    seed 0 0+    rowsM <- VUM.new (max 1 n)+    rowHashM <- VUM.new (max 1 n)+    forkJoin_+        [ scatterChunk hashes shift cur rowsM rowHashM lo hi+        | ((lo, hi), cur) <- zip chunks cursors+        ]+    partStart <- VU.unsafeFreeze partStartM+    partRows <- VU.unsafeFreeze rowsM+    partHashes <- VU.unsafeFreeze rowHashM+    pure (partStart, partRows, partHashes)++-- | Contiguous near-equal row chunks, one per capability; empties dropped.+rowChunks :: Int -> Int -> [(Int, Int)]+rowChunks caps n =+    [ (lo, hi)+    | w <- [0 .. caps - 1]+    , let lo = min n (w * per)+    , let hi = min n (lo + per)+    , lo < hi+    ]+  where+    !per = (n + max 1 caps - 1) `div` max 1 caps++-- | Per-partition counts of one row chunk.+histChunk :: VU.Vector Int -> Int -> Int -> Int -> Int -> IO (VUM.IOVector Int)+histChunk hashes p shift lo hi = do+    acc <- VUM.replicate p (0 :: Int)+    let go !i+            | i >= hi = pure acc+            | otherwise = do+                let !pp = partIx shift (VU.unsafeIndex hashes i)+                c <- VUM.unsafeRead acc pp+                VUM.unsafeWrite acc pp (c + 1)+                go (i + 1)+    go lo++{- | Scatter one row chunk into the partitioned layout using the chunk's+pre-summed cursor (disjoint write regions per chunk, no contention).+-}+scatterChunk ::+    VU.Vector Int ->+    Int ->+    VUM.IOVector Int ->+    VUM.IOVector Int ->+    VUM.IOVector Int ->+    Int ->+    Int ->+    IO ()+scatterChunk hashes shift cursor rowsM rowHashM lo hi = go lo+  where+    go !i+        | i >= hi = pure ()+        | otherwise = do+            let !h = VU.unsafeIndex hashes i+                !pp = partIx shift h+            pos <- VUM.unsafeRead cursor pp+            VUM.unsafeWrite rowsM pos i+            VUM.unsafeWrite rowHashM pos h+            VUM.unsafeWrite cursor pp (pos + 1)+            go (i + 1)++{- | Stable-sort each partition by full key, writing sorted original indices+into @outOrder@ and their hashes into @outKeys@ at the partition's slot range.+Forks @caps@ workers that pull partition indices off a shared atomic counter.+Within a partition the counting sort already left rows in ascending original+order, so the LSD radix sort's stability reproduces the global @(key, row)@+order. Partitions below two elements are already sorted (counting sort kept+original order) and are copied directly.++@partHashes@ is the partition-layout hash vector from 'partitionRows', so+seeding reads hashes sequentially; only 7 LSD passes run (the top byte is the+partition byte, constant within a partition), and the sorted hash is recovered+from the sort key ('sortKey' is self-inverse) instead of a random gather.+-}+sortPartitions ::+    Int ->+    Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    VU.Vector Int ->+    VUM.IOVector Int ->+    VUM.IOVector Int ->+    IO ()+sortPartitions caps p partStart partRows partHashes outOrder outKeys =+    pooledIndices caps p sortOne+  where+    sortOne !pp = do+        let !s = VU.unsafeIndex partStart pp+            !e = VU.unsafeIndex partStart (pp + 1)+            !sz = e - s+        when (sz > 0) $+            if sz == 1+                then do+                    VUM.unsafeWrite outOrder s (VU.unsafeIndex partRows s)+                    VUM.unsafeWrite outKeys s (VU.unsafeIndex partHashes s)+                else do+                    keysA <- VUM.new sz+                    orderA <- VUM.new sz+                    let seed !i+                            | i >= sz = pure ()+                            | otherwise = do+                                VUM.unsafeWrite keysA i (sortKey (VU.unsafeIndex partHashes (s + i)))+                                VUM.unsafeWrite orderA i (VU.unsafeIndex partRows (s + i))+                                seed (i + 1)+                    seed 0+                    keysB <- VUM.new sz+                    orderB <- VUM.new sz+                    radixPassesN 7 sz keysA orderA keysB orderB+                    let emit !i+                            | i >= sz = pure ()+                            | otherwise = do+                                o <- VUM.unsafeRead orderB i+                                k <- VUM.unsafeRead keysB i+                                VUM.unsafeWrite outOrder (s + i) o+                                -- sortKey is an involution: recover the hash.+                                VUM.unsafeWrite outKeys (s + i) (sortKey k)+                                emit (i + 1)+                    emit 0

@@ -1,1835 +1,26 @@-{-# LANGUAGE AllowAmbiguousTypes #-}-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE InstanceSigs #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE PolyKinds #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeApplications #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE UndecidableInstances #-}--module DataFrame.Internal.Column where--import qualified Data.Text as T-import qualified Data.Vector as VB-import qualified Data.Vector.Generic as VG-import qualified Data.Vector.Mutable as VBM-import qualified Data.Vector.Unboxed as VU-import qualified Data.Vector.Unboxed.Mutable as VUM--import Control.Exception (throw)-import Control.Monad (forM_, when)-import Control.Monad.ST (ST, runST)-import Data.Bits (-    complement,-    popCount,-    setBit,-    shiftL,-    shiftR,-    testBit,-    (.&.),- )-import Data.Kind (Type)-import Data.Maybe-import Data.Type.Equality (TestEquality (..))-import Data.Word (Word8)-import DataFrame.Errors-import DataFrame.Internal.PackedText (-    PackedTextData (..),-    packedGather,-    packedIndexText,-    packedLength,-    packedSlice,-    packedTake,-    sliceEqBytes,- )-import DataFrame.Internal.Types-import System.IO.Unsafe (unsafePerformIO)-import System.Random-import Type.Reflection---- | A bit-packed validity bitmap. Bit @i@ = 1 means row @i@ is valid (not null).-type Bitmap = VU.Vector Word8--{- | Type-erased column GADT. Pattern-matching on the constructor recovers the-representation; nullability is an optional bit-packed 'Bitmap' (@Nothing@ = no-nulls, @Just bm@ = bit @i@ set iff row @i@ is valid).--}-data Column where-    BoxedColumn :: (Columnable a) => Maybe Bitmap -> VB.Vector a -> Column-    UnboxedColumn ::-        (Columnable a, VU.Unbox a) => Maybe Bitmap -> VU.Vector a -> Column-    -- Efficient intermediate formats.-    -- Bit-packed Text: shared UTF-8 byte buffer + row offsets + optional bitmap;-    -- Text is materialized on demand. Only CSV ingest emits this; user-built-    -- Text columns stay 'BoxedColumn'.-    PackedText :: Maybe Bitmap -> {-# UNPACK #-} !PackedTextData -> Column-    -- A join's same-named non-key column pair ('mergeColumns'): both sides-    -- keep their native (packed/dict/unboxed) representation; per-row 'These'-    -- values only materialize on element access ('materializeMerged').-    MergedColumn :: !Column -> !Column -> Column--{- | A mutable companion struct to dataframe columns.--Used mostly as an intermediate structure for I/O.--}-data MutableColumn where-    MBoxedColumn :: (Columnable a) => VBM.IOVector a -> MutableColumn-    MUnboxedColumn :: (Columnable a, VU.Unbox a) => VUM.IOVector a -> MutableColumn---- ------------------------------------------------------------------------------ Bitmap helpers--- ------------------------------------------------------------------------------- | Test whether row @i@ is valid (not null) in a bitmap.-bitmapTestBit :: Bitmap -> Int -> Bool-bitmapTestBit bm i = testBit (VU.unsafeIndex bm (i `shiftR` 3)) (i .&. 7)-{-# INLINE bitmapTestBit #-}---- | Build a fully-valid bitmap for @n@ rows (all bits set).-allValidBitmap :: Int -> Bitmap-allValidBitmap n =-    let bytes = (n + 7) `shiftR` 3-        lastBits = n .&. 7-        full = VU.replicate (bytes - 1) 0xFF-        lastByte = if lastBits == 0 then 0xFF else (1 `shiftL` lastBits) - 1-     in if bytes == 0 then VU.empty else VU.snoc full lastByte-{-# INLINE allValidBitmap #-}--{- | Build a bitmap from a @VU.Vector Word8@ validity vector-(1 = valid, 0 = null), as produced by Arrow / Parquet decoders.--}-buildBitmapFromValid :: VU.Vector Word8 -> Bitmap-buildBitmapFromValid valid =-    let n = VU.length valid-        bytes = (n + 7) `shiftR` 3-     in VU.generate bytes $ \b ->-            let base = b `shiftL` 3-                setBitIf acc bit =-                    let idx = base + bit-                     in if idx < n && VU.unsafeIndex valid idx /= 0-                            then setBit acc bit-                            else acc-             in foldl setBitIf (0 :: Word8) [0 .. 7]--{- | Build a bitmap from a list of null-row indices.-@nullIdxs@ are the positions that are NULL.--}-buildBitmapFromNulls :: Int -> [Int] -> Bitmap-buildBitmapFromNulls n nullIdxs =-    let base = allValidBitmap n-     in VU.modify-            ( \mv ->-                forM_ nullIdxs $ \i -> do-                    let byteIdx = i `shiftR` 3-                        bitIdx = i .&. 7-                    v <- VUM.unsafeRead mv byteIdx-                    VUM.unsafeWrite mv byteIdx (clearBit8 v bitIdx)-            )-            base-  where-    clearBit8 :: Word8 -> Int -> Word8-    clearBit8 b bit = b .&. complement (1 `shiftL` bit)---- | Slice a bitmap for rows @[start .. start+len-1]@.-bitmapSlice :: Int -> Int -> Bitmap -> Bitmap-bitmapSlice start len bm-    | start .&. 7 == 0 =-        let startByte = start `shiftR` 3-            bytes = min ((len + 7) `shiftR` 3) (VU.length bm - startByte)-         in VU.slice startByte bytes bm-    | otherwise =-        let n = min len (VU.length bm `shiftL` 3 - start)-         in buildBitmapFromValid $-                VU.generate n $-                    \i -> if bitmapTestBit bm (start + i) then 1 else 0---- | Concatenate two bitmaps covering @n1@ and @n2@ rows respectively.-bitmapConcat :: Int -> Bitmap -> Int -> Bitmap -> Bitmap-bitmapConcat n1 bm1 n2 bm2 =-    buildBitmapFromValid $-        VU.generate (n1 + n2) $ \i ->-            if i < n1-                then if bitmapTestBit bm1 i then 1 else 0-                else if bitmapTestBit bm2 (i - n1) then 1 else 0---- | Combine two bitmaps with AND (both must be valid for result to be valid).-mergeBitmaps :: Bitmap -> Bitmap -> Bitmap-mergeBitmaps = VU.zipWith (.&.)--{- | Materialize a nullable column from @VB.Vector (Maybe a)@; picks 'UnboxedColumn'-when @a@ is unboxable, else 'BoxedColumn'. Always attaches a bitmap so the column-reads as nullable even with no 'Nothing' values.--}-fromMaybeVec :: forall a. (Columnable a) => VB.Vector (Maybe a) -> Column-fromMaybeVec v = case sUnbox @a of-    STrue -> fromMaybeVecUnboxed v-    SFalse ->-        let n = VB.length v-            nullIdxs = [i | i <- [0 .. n - 1], isNothing (VB.unsafeIndex v i)]-            bm = if null nullIdxs then allValidBitmap n else buildBitmapFromNulls n nullIdxs-            dat = VB.map (fromMaybe (errorWithoutStackTrace "fromMaybeVec: Nothing slot")) v-         in BoxedColumn (Just bm) dat--{- | Materialize a nullable 'UnboxedColumn' to @VB.Vector (Maybe a)@ using runST.-Always attaches a bitmap so the column is recognized as nullable even when-no 'Nothing' values are present (preserves the Maybe type marker).--}-fromMaybeVecUnboxed ::-    forall a. (Columnable a, VU.Unbox a) => VB.Vector (Maybe a) -> Column-fromMaybeVecUnboxed v =-    let n = VB.length v-        nullIdxs = [i | i <- [0 .. n - 1], isNothing (VB.unsafeIndex v i)]-        bm = if null nullIdxs then allValidBitmap n else buildBitmapFromNulls n nullIdxs-        dat = runST $ do-            mv <- VUM.new n-            VG.iforM_ v $ \i mx -> forM_ mx (VUM.unsafeWrite mv i)-            VU.unsafeFreeze mv-     in UnboxedColumn (Just bm) dat---- | Whether row @i@ is null, respecting the bitmap.-columnElemIsNull :: Column -> Int -> Bool-columnElemIsNull (BoxedColumn (Just bm) _) i = not (bitmapTestBit bm i)-columnElemIsNull (UnboxedColumn (Just bm) _) i = not (bitmapTestBit bm i)-columnElemIsNull (PackedText (Just bm) _) i = not (bitmapTestBit bm i)-columnElemIsNull _ _ = False---- | Return the 'Maybe Bitmap' from a column.-columnBitmap :: Column -> Maybe Bitmap-columnBitmap (BoxedColumn bm _) = bm-columnBitmap (UnboxedColumn bm _) = bm-columnBitmap (PackedText bm _) = bm-columnBitmap (MergedColumn _ _) = Nothing--{- | Decode a 'PackedText' into a @BoxedColumn Text@ (bit-identical to-materializing at freeze). Identity on every other column.--}-materializePacked :: Column -> Column-materializePacked (PackedText bm p) =-    BoxedColumn bm (VB.generate (packedLength p) (packedIndexText p))-materializePacked c = c-{-# INLINE materializePacked #-}---- | Whether a column is a 'PackedText'.-isPackedText :: Column -> Bool-isPackedText (PackedText _ _) = True-isPackedText _ = False-{-# INLINE isPackedText #-}---- | Whether a column is a 'MergedColumn'.-isMergedColumn :: Column -> Bool-isMergedColumn (MergedColumn _ _) = True-isMergedColumn _ = False-{-# INLINE isMergedColumn #-}--{- | 'MergedColumn' defers element construction, so forcing must still surface-the one deferred error — a row null on both sides — inside strict IO/executor-boundaries. O(rows) bitmap walk, no allocation; both-null needs a bitmap on-each side, so anything else passes immediately.--}-checkMergedNoBothNull :: Column -> Column -> ()-checkMergedNoBothNull a b = case (columnBitmap a, columnBitmap b) of-    (Just ba, Just bb) ->-        let !n = min (columnLength a) (columnLength b)-            go !i-                | i >= n = ()-                | bitmapTestBit ba i || bitmapTestBit bb i = go (i + 1)-                | otherwise = error "mergeColumns: both null"-         in go 0-    _ -> ()---- ------------------------------------------------------------------------------ End bitmap helpers--- -----------------------------------------------------------------------------{- | A wrapper around the type-erased 'Column' carrying a phantom element type,-used to type-check expressions. The phantom is not guaranteed to match the-underlying vector's type.--}-data TypedColumn a where-    TColumn :: (Columnable a) => Column -> TypedColumn a--instance (Eq a) => Eq (TypedColumn a) where-    (==) :: (Eq a) => TypedColumn a -> TypedColumn a -> Bool-    (==) (TColumn a) (TColumn b) = a == b---- | Gets the underlying value from a TypedColumn.-unwrapTypedColumn :: TypedColumn a -> Column-unwrapTypedColumn (TColumn value) = value---- | Gets the underlying vector from a TypedColumn.-vectorFromTypedColumn :: TypedColumn a -> VB.Vector a-vectorFromTypedColumn (TColumn value) = either throw id (toVector value)---- | Checks if a column contains missing values (has a bitmap).-hasMissing :: Column -> Bool-hasMissing (BoxedColumn (Just _) _) = True-hasMissing (UnboxedColumn (Just _) _) = True-hasMissing (PackedText (Just _) _) = True-hasMissing _ = False---- | Checks if a column contains only missing values.-allMissing :: Column -> Bool-allMissing (BoxedColumn (Just bm) col) = VU.all (== 0) bm && not (VB.null col)-allMissing (UnboxedColumn (Just bm) col) = VU.all (== 0) bm && not (VU.null col)-allMissing (PackedText (Just bm) p) = VU.all (== 0) bm && packedLength p > 0-allMissing _ = False---- | Checks if a column contains numeric values.-isNumeric :: Column -> Bool-isNumeric c@(MergedColumn _ _) = isNumeric (mergedHead c)-isNumeric (UnboxedColumn _ (_vec :: VU.Vector a)) = case sNumeric @a of-    STrue -> True-    _ -> False-isNumeric (BoxedColumn _ (_vec :: VB.Vector a)) = case testEquality (typeRep @a) (typeRep @Integer) of-    Nothing -> False-    Just Refl -> True-isNumeric (PackedText _ _) = False--{- | Whether the column stores element type @a@. For nullable columns, also-'True' when @a = Maybe b@ and the column stores @b@ internally.--}-hasElemType :: forall a. (Columnable a) => Column -> Bool-hasElemType = \case-    BoxedColumn bm (_column :: VB.Vector b) -> checkBoxed bm (typeRep @b)-    UnboxedColumn bm (_column :: VU.Vector b) -> checkUnboxed bm (typeRep @b)-    PackedText bm _ -> checkBoxed bm (typeRep @T.Text)-    c@(MergedColumn _ _) -> hasElemType @a (mergedHead c)-  where-    directMatch :: forall (b :: Type). TypeRep b -> Bool-    directMatch = isJust . testEquality (typeRep @a)-    checkMaybe :: forall (b :: Type). TypeRep b -> Bool-    checkMaybe tb = case typeRep @a of-        App tMaybe tInner -> case eqTypeRep tMaybe (typeRep @Maybe) of-            Just HRefl -> isJust (testEquality tInner tb)-            Nothing -> False-        _ -> False-    checkBoxed :: forall (b :: Type). Maybe Bitmap -> TypeRep b -> Bool-    checkBoxed bm tb = directMatch tb || (isJust bm && checkMaybe tb)-    checkUnboxed :: forall (b :: Type). Maybe Bitmap -> TypeRep b -> Bool-    checkUnboxed bm tb = directMatch tb || (isJust bm && checkMaybe tb)---- | An internal/debugging function to get the column type of a column.-columnVersionString :: Column -> String-columnVersionString column = case column of-    BoxedColumn Nothing _ -> "Boxed"-    BoxedColumn (Just _) _ -> "NullableBoxed"-    UnboxedColumn Nothing _ -> "Unboxed"-    UnboxedColumn (Just _) _ -> "NullableUnboxed"-    PackedText Nothing _ -> "Boxed"-    PackedText (Just _) _ -> "NullableBoxed"-    MergedColumn _ _ -> columnVersionString (mergedHead column)--{- | An internal/debugging function to get the type stored in the outermost vector-of a column.--}-columnTypeString :: Column -> String-columnTypeString column = case column of-    BoxedColumn Nothing (_ :: VB.Vector a) -> show (typeRep @a)-    BoxedColumn (Just _) (_ :: VB.Vector a) -> showMaybeType @a-    UnboxedColumn Nothing (_ :: VU.Vector a) -> show (typeRep @a)-    UnboxedColumn (Just _) (_ :: VU.Vector a) -> showMaybeType @a-    PackedText Nothing _ -> show (typeRep @T.Text)-    PackedText (Just _) _ -> showMaybeType @T.Text-    MergedColumn _ _ -> columnTypeString (mergedHead column)-  where-    showMaybeType :: forall a. (Typeable a) => String-    showMaybeType =-        let s = show (typeRep @a)-         in "Maybe " ++ if ' ' `elem` s then "(" ++ s ++ ")" else s--instance (Show a) => Show (TypedColumn a) where-    show :: (Show a) => TypedColumn a -> String-    show (TColumn col) = show col--{- | Force evaluation of all elements in a column. Replacement for the removed-@instance NFData Column@; used by the IO and lazy-executor strict paths.--}-forceColumn :: Column -> ()-forceColumn (BoxedColumn Nothing (v :: VB.Vector a)) = VB.foldl' (const (`seq` ())) () v-forceColumn (BoxedColumn (Just bm) (v :: VB.Vector a)) =-    let n = VB.length v-        go !i-            | i >= n = ()-            | bitmapTestBit bm i = VB.unsafeIndex v i `seq` go (i + 1)-            | otherwise = go (i + 1)-     in go 0-forceColumn (UnboxedColumn _ v) = v `seq` ()-forceColumn (PackedText _ (PackedTextData arr offs sel _)) = arr `seq` offs `seq` sel `seq` ()-forceColumn (MergedColumn a b) =-    forceColumn a `seq` forceColumn b `seq` checkMergedNoBothNull a b--instance Show Column where-    show :: Column -> String-    show c@(MergedColumn _ _) = show (materializeMerged c)-    show (BoxedColumn Nothing column) = show column-    show (BoxedColumn (Just bm) column) =-        let n = VB.length column-            elems =-                [ if bitmapTestBit bm i then show (VB.unsafeIndex column i) else "null"-                | i <- [0 .. n - 1]-                ]-         in "[" ++ foldl (\acc e -> if null acc then e else acc ++ "," ++ e) "" elems ++ "]"-    show (UnboxedColumn Nothing column) = show column-    show (UnboxedColumn (Just bm) column) =-        let n = VU.length column-            elems =-                [ if bitmapTestBit bm i then show (VU.unsafeIndex column i) else "null"-                | i <- [0 .. n - 1]-                ]-         in "[" ++ foldl (\acc e -> if null acc then e else acc ++ "," ++ e) "" elems ++ "]"-    show c@(PackedText _ _) = show (materializePacked c)--{- | Compare two nullable boxed columns element by element, skipping null slots.-Uses a manual loop to avoid stream fusion forcing null-slot error thunks.--}-eqBoxedCols ::-    (Eq a) => Maybe Bitmap -> VB.Vector a -> Maybe Bitmap -> VB.Vector a -> Bool-eqBoxedCols bm1 a bm2 b-    | VB.length a /= VB.length b = False-    | otherwise = go 0-  where-    !n = VB.length a-    go !i-        | i >= n = True-        | nullA || nullB = (nullA == nullB) && go (i + 1)-        | VB.unsafeIndex a i == VB.unsafeIndex b i = go (i + 1)-        | otherwise = False-      where-        nullA = maybe False (\bm -> not (bitmapTestBit bm i)) bm1-        nullB = maybe False (\bm -> not (bitmapTestBit bm i)) bm2-{-# INLINE eqBoxedCols #-}--instance Eq Column where-    (==) :: Column -> Column -> Bool-    (==) (BoxedColumn bm1 (a :: VB.Vector t1)) (BoxedColumn bm2 (b :: VB.Vector t2)) =-        case testEquality (typeRep @t1) (typeRep @t2) of-            Nothing -> False-            Just Refl -> eqBoxedCols bm1 a bm2 b-    (==) (UnboxedColumn bm1 (a :: VU.Vector t1)) (UnboxedColumn bm2 (b :: VU.Vector t2)) =-        case testEquality (typeRep @t1) (typeRep @t2) of-            Nothing -> False-            Just Refl ->-                VU.length a == VU.length b-                    && VU.and-                        ( VU.imap-                            ( \i x ->-                                let nullA = maybe False (\bm -> not (bitmapTestBit bm i)) bm1-                                    nullB = maybe False (\bm -> not (bitmapTestBit bm i)) bm2-                                 in if nullA || nullB then nullA == nullB else x == VU.unsafeIndex b i-                            )-                            a-                        )-    (==) lhs@(MergedColumn _ _) rhs = materializeMerged lhs == rhs-    (==) lhs rhs@(MergedColumn _ _) = lhs == materializeMerged rhs-    (==) (PackedText bm1 p1) (PackedText bm2 p2) = eqPackedCols bm1 p1 bm2 p2-    (==) lhs@(PackedText _ _) rhs = materializePacked lhs == rhs-    (==) lhs rhs@(PackedText _ _) = lhs == materializePacked rhs-    (==) _ _ = False--{- | Byte-slice equality of two packed-text columns, skipping null slots-(a null compares equal only to a null), mirroring 'eqBoxedCols'.--}-eqPackedCols ::-    Maybe Bitmap -> PackedTextData -> Maybe Bitmap -> PackedTextData -> Bool-eqPackedCols bm1 p1 bm2 p2-    | packedLength p1 /= packedLength p2 = False-    | otherwise = go 0-  where-    !n = packedLength p1-    go !i-        | i >= n = True-        | nullA || nullB = (nullA == nullB) && go (i + 1)-        | otherwise =-            let (a1, o1, l1) = packedSlice p1 i-                (a2, o2, l2) = packedSlice p2 i-             in sliceEqBytes a1 o1 l1 a2 o2 l2 && go (i + 1)-      where-        nullA = maybe False (\bm -> not (bitmapTestBit bm i)) bm1-        nullB = maybe False (\bm -> not (bitmapTestBit bm i)) bm2-{-# INLINE eqPackedCols #-}--{- | A class for converting a vector to a column of the appropriate type.-Given each Rep we tell the `toColumnRep` function which Column type to pick.--}-class ColumnifyRep (r :: Rep) a where-    toColumnRep :: VB.Vector a -> Column---- | Constraint synonym for what we can put into columns.-type Columnable a =-    ( Columnable' a-    , ColumnifyRep (KindOf a) a-    , UnboxIf a-    , IntegralIf a-    , FloatingIf a-    , SBoolI (Unboxable a)-    , SBoolI (Numeric a)-    , SBoolI (IntegralTypes a)-    , SBoolI (FloatingTypes a)-    )--instance-    (Columnable a, VU.Unbox a) =>-    ColumnifyRep 'RUnboxed a-    where-    toColumnRep :: (Columnable a, VUM.Unbox a) => VB.Vector a -> Column-    toColumnRep v = UnboxedColumn Nothing (VU.convert v)--instance-    (Columnable a) =>-    ColumnifyRep 'RBoxed a-    where-    toColumnRep :: (Columnable a) => VB.Vector a -> Column-    toColumnRep = BoxedColumn Nothing--instance-    (Columnable a) =>-    ColumnifyRep 'RNullableBoxed (Maybe a)-    where-    toColumnRep :: (Columnable a) => VB.Vector (Maybe a) -> Column-    toColumnRep = fromMaybeVec--{- | O(n) Convert a vector to a column. Automatically picks the best representation of a vector to store the underlying data in.--__Examples:__--@-> import qualified Data.Vector as V-> fromVector (VB.fromList [(1 :: Int), 2, 3, 4])-[1,2,3,4]-@--}-fromVector ::-    forall a.-    (Columnable a, ColumnifyRep (KindOf a) a) =>-    VB.Vector a -> Column-fromVector = toColumnRep @(KindOf a)--{- | O(n) Convert an unboxed vector to a column. This avoids the extra conversion if you already have the data in an unboxed vector.--__Examples:__--@-> import qualified Data.Vector.Unboxed as V-> fromUnboxedVector (VB.fromList [(1 :: Int), 2, 3, 4])-[1,2,3,4]-@--}-fromUnboxedVector ::-    forall a. (Columnable a, VU.Unbox a) => VU.Vector a -> Column-fromUnboxedVector = UnboxedColumn Nothing--{- | O(n) Convert a list to a column. Automatically picks the best representation of a vector to store the underlying data in.--__Examples:__--@-> fromList [(1 :: Int), 2, 3, 4]-[1,2,3,4]-@--}-fromList ::-    forall a.-    (Columnable a, ColumnifyRep (KindOf a) a) =>-    [a] -> Column-fromList = toColumnRep @(KindOf a) . VB.fromList--{- | O(n) Create a column of random elements within a range.--Takes a random number generator, a length, and a lower and upper bound for the random values.--__Examples:__--@-> import System.Random (mkStdGen)-> mkRandom (mkStdGen 42) 4 0 10-[4,2,6,5]-@--}-mkRandom ::-    (RandomGen g, Columnable a, ColumnifyRep (KindOf a) a, UniformRange a) =>-    g -> Int -> a -> a -> Column-mkRandom pureGen k lo hi = fromList $ go pureGen k-  where-    go _g 0 = []-    go g n =-        let-            (!v, !g') = uniformR (lo, hi) g-         in-            v : go g' (n - 1)---- An internal helper for type errors-throwTypeMismatch ::-    forall (a :: Type) (b :: Type).-    (Typeable a, Typeable b) => Either DataFrameException Column-throwTypeMismatch =-    Left $-        TypeMismatchException-            MkTypeErrorContext-                { userType = Right (typeRep @b)-                , expectedType = Right (typeRep @a)-                , callingFunctionName = Nothing-                , errorColumnName = Nothing-                }---- | An internal function to map a function over the values of a column.-mapColumn ::-    forall b c.-    (Columnable b, Columnable c) =>-    (b -> c) -> Column -> Either DataFrameException Column-mapColumn f = \case-    BoxedColumn bm (col :: VB.Vector a) -> runBoxed bm col-    UnboxedColumn bm (col :: VU.Vector a) -> runUnboxed bm col-    c@(PackedText _ _) -> mapColumn f (materializePacked c)-    c@(MergedColumn _ _) -> mapColumn f (materializeMerged c)-  where-    runBoxed ::-        forall a.-        (Columnable a) =>-        Maybe Bitmap -> VB.Vector a -> Either DataFrameException Column-    runBoxed bm col = case testEquality (typeRep @b) (typeRep @(Maybe a)) of-        Just Refl ->-            let !n = VB.length col-             in Right $ case sUnbox @c of-                    STrue -> UnboxedColumn Nothing $-                        VU.generate n $ \i ->-                            f-                                ( if maybe True (`bitmapTestBit` i) bm-                                    then Just (VB.unsafeIndex col i)-                                    else Nothing-                                )-                    SFalse -> fromVector @c $-                        VB.generate n $ \i ->-                            f-                                ( if maybe True (`bitmapTestBit` i) bm-                                    then Just (VB.unsafeIndex col i)-                                    else Nothing-                                )-        Nothing -> case testEquality (typeRep @a) (typeRep @b) of-            Just Refl ->-                Right $ case sUnbox @c of-                    STrue -> UnboxedColumn bm (VU.generate (VB.length col) (f . VB.unsafeIndex col))-                    SFalse -> case bm of-                        Nothing -> fromVector @c (VB.map f col)-                        Just _ -> BoxedColumn bm (VB.map f col)-            Nothing -> throwTypeMismatch @a @b--    runUnboxed ::-        forall a.-        (Columnable a, VU.Unbox a) =>-        Maybe Bitmap -> VU.Vector a -> Either DataFrameException Column-    runUnboxed bm col = case testEquality (typeRep @b) (typeRep @(Maybe a)) of-        Just Refl ->-            let !n = VU.length col-             in Right $ case sUnbox @c of-                    STrue -> UnboxedColumn Nothing $-                        VU.generate n $ \i ->-                            f-                                ( if maybe True (`bitmapTestBit` i) bm-                                    then Just (VU.unsafeIndex col i)-                                    else Nothing-                                )-                    SFalse -> fromVector @c $-                        VB.generate n $ \i ->-                            f-                                ( if maybe True (`bitmapTestBit` i) bm-                                    then Just (VU.unsafeIndex col i)-                                    else Nothing-                                )-        Nothing -> case testEquality (typeRep @a) (typeRep @b) of-            Just Refl -> Right $ case sUnbox @c of-                STrue -> UnboxedColumn bm (VU.map f col)-                SFalse -> case bm of-                    Nothing -> fromVector @c (VB.generate (VU.length col) (f . VU.unsafeIndex col))-                    Just _ -> BoxedColumn bm (VB.generate (VU.length col) (f . VU.unsafeIndex col))-            Nothing -> throwTypeMismatch @a @b-{-# INLINEABLE mapColumn #-}---- | Applies a function that returns an unboxed result to an unboxed vector, storing the result in a column.-imapColumn ::-    forall b c.-    (Columnable b, Columnable c) =>-    (Int -> b -> c) -> Column -> Either DataFrameException Column-imapColumn f = \case-    BoxedColumn bm (col :: VB.Vector a) -> runBoxed bm col-    UnboxedColumn bm (col :: VU.Vector a) -> runUnboxed bm col-    c@(PackedText _ _) -> imapColumn f (materializePacked c)-    c@(MergedColumn _ _) -> imapColumn f (materializeMerged c)-  where-    runBoxed ::-        forall a.-        (Columnable a) =>-        Maybe Bitmap -> VB.Vector a -> Either DataFrameException Column-    runBoxed bm col = case testEquality (typeRep @a) (typeRep @b) of-        Just Refl -> Right $ case sUnbox @c of-            STrue ->-                UnboxedColumn-                    bm-                    (VU.generate (VB.length col) (\i -> f i (VB.unsafeIndex col i)))-            SFalse -> BoxedColumn bm (VB.imap f col)-        Nothing -> throwTypeMismatch @a @b--    runUnboxed ::-        forall a.-        (Columnable a, VU.Unbox a) =>-        Maybe Bitmap -> VU.Vector a -> Either DataFrameException Column-    runUnboxed bm col = case testEquality (typeRep @a) (typeRep @b) of-        Just Refl -> Right $ case sUnbox @c of-            STrue -> UnboxedColumn bm (VU.imap f col)-            SFalse -> BoxedColumn bm (VB.imap f (VG.convert col))-        Nothing -> throwTypeMismatch @a @b---- | O(1) Gets the number of elements in the column.-columnLength :: Column -> Int-columnLength (MergedColumn a b) = min (columnLength a) (columnLength b)-columnLength (BoxedColumn _ xs) = VB.length xs-columnLength (UnboxedColumn _ xs) = VU.length xs-columnLength (PackedText _ p) = packedLength p-{-# INLINE columnLength #-}---- | O(n) Gets the number of non-null elements in the column.-numElements :: Column -> Int-numElements (MergedColumn a b) = min (columnLength a) (columnLength b)-numElements (BoxedColumn Nothing xs) = VB.length xs-numElements (BoxedColumn (Just bm) _xs) = VU.foldl' (\acc b -> acc + popCount b) 0 bm-numElements (UnboxedColumn Nothing xs) = VU.length xs-numElements (UnboxedColumn (Just bm) _xs) = VU.foldl' (\acc b -> acc + popCount b) 0 bm-numElements (PackedText Nothing p) = packedLength p-numElements (PackedText (Just bm) _p) = VU.foldl' (\acc b -> acc + popCount b) 0 bm-{-# INLINE numElements #-}---- | O(n) Takes the first n values of a column.-takeColumn :: Int -> Column -> Column-takeColumn n (MergedColumn a b) = MergedColumn (takeColumn n a) (takeColumn n b)-takeColumn n (BoxedColumn bm xs) =-    BoxedColumn (fmap (bitmapSlice 0 n) bm) (VG.take n xs)-takeColumn n (UnboxedColumn bm xs) =-    UnboxedColumn (fmap (bitmapSlice 0 n) bm) (VG.take n xs)-takeColumn n (PackedText bm p) =-    PackedText (fmap (bitmapSlice 0 n) bm) (packedTake n p)-{-# INLINE takeColumn #-}---- | O(n) Takes the last n values of a column.-takeLastColumn :: Int -> Column -> Column-takeLastColumn n column = sliceColumn (columnLength column - n) n column-{-# INLINE takeLastColumn #-}---- | O(n) Takes n values after a given column index.-sliceColumn :: Int -> Int -> Column -> Column-sliceColumn start n (MergedColumn a b) =-    MergedColumn (sliceColumn start n a) (sliceColumn start n b)-sliceColumn start n (BoxedColumn bm xs) =-    BoxedColumn (fmap (bitmapSlice start n) bm) (VG.slice start n xs)-sliceColumn start n (UnboxedColumn bm xs) =-    UnboxedColumn (fmap (bitmapSlice start n) bm) (VG.slice start n xs)-sliceColumn start n c@(PackedText _ _) = sliceColumn start n (materializePacked c)-{-# INLINE sliceColumn #-}---- | O(n) Selects the elements at a given set of indices. Does not change the order.-atIndicesStable :: VU.Vector Int -> Column -> Column-atIndicesStable indexes (BoxedColumn bm column) =-    BoxedColumn-        ( fmap-            ( \bm0 ->-                buildBitmapFromValid $-                    VU.map (\i -> if bitmapTestBit bm0 i then 1 else 0) indexes-            )-            bm-        )-        ( VB.generate-            (VU.length indexes)-            ((column `VB.unsafeIndex`) . (indexes `VU.unsafeIndex`))-        )-atIndicesStable indexes (UnboxedColumn bm column) =-    UnboxedColumn-        ( fmap-            ( \bm0 ->-                buildBitmapFromValid $-                    VU.map (\i -> if bitmapTestBit bm0 i then 1 else 0) indexes-            )-            bm-        )-        (VU.unsafeBackpermute column indexes)-atIndicesStable indexes (MergedColumn a b) =-    MergedColumn (atIndicesStable indexes a) (atIndicesStable indexes b)-atIndicesStable indexes (PackedText bm p) =-    PackedText-        ( fmap-            ( \bm0 ->-                buildBitmapFromValid $-                    VU.map (\i -> if bitmapTestBit bm0 i then 1 else 0) indexes-            )-            bm-        )-        (packedGather indexes p)-{-# INLINE atIndicesStable #-}--{- | Like 'atIndicesStable' but treats negative indices as null.-Keeps the index vector fully unboxed (no @VB.Vector (Maybe Int)@).--}-gatherWithSentinel :: VU.Vector Int -> Column -> Column-gatherWithSentinel indices c@(MergedColumn _ _) =-    gatherWithSentinel indices (materializeMerged c)-gatherWithSentinel indices col =-    let !n = VU.length indices-        newBm = buildBitmapFromValid $ VU.generate n $ \i ->-            if VU.unsafeIndex indices i < 0 then 0 else 1-     in case col of-            PackedText srcBm p ->-                let bm = case srcBm of-                        Nothing -> Just newBm-                        Just sb ->-                            Just-                                ( mergeBitmaps-                                    newBm-                                    ( buildBitmapFromValid $ VU.generate n $ \i ->-                                        let idx = VU.unsafeIndex indices i-                                         in if idx >= 0 && bitmapTestBit sb idx then 1 else 0-                                    )-                                )-                 in PackedText bm (packedGather indices p)-            BoxedColumn srcBm v ->-                let dat = VB.generate n $ \i ->-                        let !idx = VU.unsafeIndex indices i-                         in if idx < 0 then VB.unsafeIndex v 0 else VB.unsafeIndex v idx-                    bm = case srcBm of-                        Nothing -> Just newBm-                        Just sb ->-                            Just-                                ( mergeBitmaps-                                    newBm-                                    ( buildBitmapFromValid $ VU.generate n $ \i ->-                                        let idx = VU.unsafeIndex indices i-                                         in if idx >= 0 && bitmapTestBit sb idx then 1 else 0-                                    )-                                )-                 in BoxedColumn bm dat-            UnboxedColumn srcBm v ->-                let dat = runST $ do-                        mv <- VUM.new n-                        VG.iforM_ indices $ \i idx ->-                            when (idx >= 0) $ VUM.unsafeWrite mv i (VU.unsafeIndex v idx)-                        VU.unsafeFreeze mv-                    bm = case srcBm of-                        Nothing -> Just newBm-                        Just sb ->-                            Just-                                ( mergeBitmaps-                                    newBm-                                    ( buildBitmapFromValid $ VU.generate n $ \i ->-                                        let idx = VU.unsafeIndex indices i-                                         in if idx >= 0 && bitmapTestBit sb idx then 1 else 0-                                    )-                                )-                 in UnboxedColumn bm dat-{-# INLINE gatherWithSentinel #-}---- | Internal helper to get indices in a boxed vector.-getIndices :: VU.Vector Int -> VB.Vector a -> VB.Vector a-getIndices indices xs = VB.generate (VU.length indices) (\i -> xs VB.! (indices VU.! i))-{-# INLINE getIndices #-}---- | Internal helper to get indices in an unboxed vector.-getIndicesUnboxed :: (VU.Unbox a) => VU.Vector Int -> VU.Vector a -> VU.Vector a-getIndicesUnboxed indices xs = VU.generate (VU.length indices) (\i -> xs VU.! (indices VU.! i))-{-# INLINE getIndicesUnboxed #-}--findIndices ::-    forall a.-    (Columnable a) =>-    (a -> Bool) ->-    Column ->-    Either DataFrameException (VU.Vector Int)-findIndices predicate = \case-    BoxedColumn _ (v :: VB.Vector b) -> run v VG.convert-    UnboxedColumn _ (v :: VU.Vector b) -> run v id-    c@(PackedText _ _) -> findIndices predicate (materializePacked c)-    c@(MergedColumn _ _) -> findIndices predicate (materializeMerged c)-  where-    run ::-        forall b v.-        (Typeable b, VG.Vector v b, VG.Vector v Int) =>-        v b ->-        (v Int -> VU.Vector Int) ->-        Either DataFrameException (VU.Vector Int)-    run column finalize = case testEquality (typeRep @a) (typeRep @b) of-        Just Refl -> Right . finalize $ VG.findIndices predicate column-        Nothing ->-            Left $-                TypeMismatchException-                    MkTypeErrorContext-                        { userType = Right (typeRep @a)-                        , expectedType = Right (typeRep @b)-                        , callingFunctionName = Just "findIndices"-                        , errorColumnName = Nothing-                        }---- | Fold (right) column with index.-ifoldrColumn ::-    forall a b.-    (Columnable a, Columnable b) =>-    (Int -> a -> b -> b) -> b -> Column -> Either DataFrameException b-ifoldrColumn f acc = \case-    BoxedColumn _ column -> foldrWorker column-    UnboxedColumn _ column -> foldrWorker column-    c@(PackedText _ _) -> ifoldrColumn f acc (materializePacked c)-    c@(MergedColumn _ _) -> ifoldrColumn f acc (materializeMerged c)-  where-    foldrWorker ::-        forall c v.-        (Typeable c, VG.Vector v c) =>-        v c ->-        Either DataFrameException b-    foldrWorker vec = case testEquality (typeRep @a) (typeRep @c) of-        Just Refl -> pure $ VG.ifoldr f acc vec-        Nothing ->-            Left $-                TypeMismatchException-                    ( MkTypeErrorContext-                        { userType = Right (typeRep @a)-                        , expectedType = Right (typeRep @c)-                        , callingFunctionName = Just "ifoldrColumn"-                        , errorColumnName = Nothing-                        }-                    )--foldlColumn ::-    forall a b.-    (Columnable a, Columnable b) =>-    (b -> a -> b) -> b -> Column -> Either DataFrameException b-foldlColumn f acc = \case-    BoxedColumn _ column -> foldlWorker column-    UnboxedColumn _ column -> foldlWorker column-    c@(PackedText _ _) -> foldlColumn f acc (materializePacked c)-    c@(MergedColumn _ _) -> foldlColumn f acc (materializeMerged c)-  where-    foldlWorker ::-        forall c v.-        (Typeable c, VG.Vector v c) =>-        v c ->-        Either DataFrameException b-    foldlWorker vec = case testEquality (typeRep @a) (typeRep @c) of-        Just Refl -> pure $ VG.foldl' f acc vec-        Nothing ->-            Left $-                TypeMismatchException-                    ( MkTypeErrorContext-                        { userType = Right (typeRep @a)-                        , expectedType = Right (typeRep @c)-                        , callingFunctionName = Just "ifoldrColumn"-                        , errorColumnName = Nothing-                        }-                    )--foldl1Column ::-    forall a.-    (Columnable a) =>-    (a -> a -> a) -> Column -> Either DataFrameException a-foldl1Column f = \case-    BoxedColumn _ column -> foldl1Worker column-    UnboxedColumn _ column -> foldl1Worker column-    c@(PackedText _ _) -> foldl1Column f (materializePacked c)-    c@(MergedColumn _ _) -> foldl1Column f (materializeMerged c)-  where-    foldl1Worker ::-        forall c v.-        (Typeable c, VG.Vector v c) =>-        v c ->-        Either DataFrameException a-    foldl1Worker vec = case testEquality (typeRep @a) (typeRep @c) of-        Just Refl -> pure $ VG.foldl1' f vec-        Nothing ->-            Left $-                TypeMismatchException-                    ( MkTypeErrorContext-                        { userType = Right (typeRep @a)-                        , expectedType = Right (typeRep @c)-                        , callingFunctionName = Just "foldl1Column"-                        , errorColumnName = Nothing-                        }-                    )--{- | O(n) Seedless fold over groups using the first element of each group as seed.-Like 'foldDirectGroups' but for the case where no initial accumulator is available.--}-foldl1DirectGroups ::-    forall a.-    (Columnable a) =>-    (a -> a -> a) ->-    Column ->-    VU.Vector Int ->-    VU.Vector Int ->-    Either DataFrameException Column-foldl1DirectGroups f col valueIndices offsets-    | VU.length offsets <= 1 = pure $ fromVector @a VB.empty-    | otherwise = case col of-        UnboxedColumn _ (vec :: VU.Vector d) -> UnboxedColumn Nothing <$> foldl1Worker vec-        BoxedColumn _ (vec :: VB.Vector d) -> BoxedColumn Nothing <$> foldl1Worker vec-        PackedText _ _ -> foldl1DirectGroups f (materializePacked col) valueIndices offsets-        MergedColumn _ _ -> foldl1DirectGroups f (materializeMerged col) valueIndices offsets-  where-    foldl1Worker ::-        forall c v.-        (Typeable c, VG.Vector v c) =>-        v c ->-        Either DataFrameException (v c)-    foldl1Worker vec = case testEquality (typeRep @a) (typeRep @c) of-        Just Refl ->-            Right $-                VG.generate (VU.length offsets - 1) foldGroup-          where-            foldGroup k =-                let !s = VU.unsafeIndex offsets k-                    !e = VU.unsafeIndex offsets (k + 1)-                    !seed = VG.unsafeIndex vec (VU.unsafeIndex valueIndices s)-                 in go (s + 1) e seed-            go !i !e !acc-                | i >= e = acc-                | otherwise =-                    go (i + 1) e $!-                        f acc (VG.unsafeIndex vec (VU.unsafeIndex valueIndices i))-        Nothing ->-            Left $-                TypeMismatchException-                    MkTypeErrorContext-                        { userType = Right (typeRep @a)-                        , expectedType = Right (typeRep @c)-                        , callingFunctionName = Just "foldl1DirectGroups"-                        , errorColumnName = Nothing-                        }-{-# INLINEABLE foldl1DirectGroups #-}--{- | O(n) fold over groups by scanning the column linearly (rowToGroup[i] = group-of row i). Random writes hit the small per-group accumulator array; when @acc@ is-unboxable that array is unboxed, avoiding pointer indirection.--}-foldLinearGroups ::-    forall b acc.-    (Columnable b, Columnable acc) =>-    (acc -> b -> acc) ->-    acc ->-    Column ->-    VU.Vector Int ->-    Int ->-    Either DataFrameException Column-foldLinearGroups f seed col rowToGroup nGroups-    | nGroups == 0 = Right (fromVector @acc VB.empty)-    | otherwise = case col of-        UnboxedColumn _ (vec :: VU.Vector d) -> foldLinearWorker vec-        BoxedColumn _ (vec :: VB.Vector d) -> foldLinearWorker vec-        PackedText _ _ ->-            foldLinearGroups f seed (materializePacked col) rowToGroup nGroups-        MergedColumn _ _ ->-            foldLinearGroups f seed (materializeMerged col) rowToGroup nGroups-  where-    foldLinearWorker ::-        forall c v.-        (Typeable c, VG.Vector v c) =>-        v c ->-        Either DataFrameException Column-    foldLinearWorker vec = case testEquality (typeRep @b) (typeRep @c) of-        Just Refl ->-            Right $-                unsafePerformIO $-                    runWith-                        ( \readAt writeAt ->-                            VG.iforM_ vec $ \row x -> do-                                let !k = VG.unsafeIndex rowToGroup row-                                cur <- readAt k-                                writeAt k $! f cur x-                        )-        Nothing ->-            Left $-                TypeMismatchException-                    MkTypeErrorContext-                        { userType = Right (typeRep @b)-                        , expectedType = Right (typeRep @c)-                        , callingFunctionName = Just "foldLinearGroups"-                        , errorColumnName = Nothing-                        }--    runWith :: ((Int -> IO acc) -> (Int -> acc -> IO ()) -> IO ()) -> IO Column-    runWith body = case sUnbox @acc of-        STrue -> do-            accs <- VUM.replicate nGroups seed-            body (VUM.unsafeRead accs) (VUM.unsafeWrite accs)-            UnboxedColumn Nothing <$> VU.unsafeFreeze accs-        SFalse -> do-            accs <- VBM.replicate nGroups seed-            body (VBM.unsafeRead accs) (VBM.unsafeWrite accs)-            fromVector @acc <$> VB.unsafeFreeze accs-    {-# INLINE runWith #-}-{-# INLINEABLE foldLinearGroups #-}--headColumn :: forall a. (Columnable a) => Column -> Either DataFrameException a-headColumn = \case-    BoxedColumn _ col -> headWorker col-    UnboxedColumn _ col -> headWorker col-    c@(PackedText _ _) -> headColumn (materializePacked c)-    c@(MergedColumn _ _) -> headColumn (mergedHead c)-  where-    headWorker ::-        forall c v.-        (Typeable c, VG.Vector v c) =>-        v c ->-        Either DataFrameException a-    headWorker vec = case testEquality (typeRep @a) (typeRep @c) of-        Just Refl ->-            if VG.null vec-                then Left (EmptyDataSetException "headColumn")-                else pure (VG.head vec)-        Nothing ->-            Left $-                TypeMismatchException-                    ( MkTypeErrorContext-                        { userType = Right (typeRep @a)-                        , expectedType = Right (typeRep @c)-                        , callingFunctionName = Just "headColumn"-                        , errorColumnName = Nothing-                        }-                    )---- | An internal, column version of zip.-zipColumns :: Column -> Column -> Column-zipColumns l@(MergedColumn _ _) r = zipColumns (materializeMerged l) r-zipColumns l r@(MergedColumn _ _) = zipColumns l (materializeMerged r)-zipColumns l@(PackedText _ _) r = zipColumns (materializePacked l) r-zipColumns l r@(PackedText _ _) = zipColumns l (materializePacked r)-zipColumns (BoxedColumn _ column) (BoxedColumn _ other) = BoxedColumn Nothing (VG.zip column other)-zipColumns (BoxedColumn _ column) (UnboxedColumn _ other) =-    BoxedColumn-        Nothing-        ( VB.generate-            (min (VG.length column) (VG.length other))-            (\i -> (column VG.! i, other VG.! i))-        )-zipColumns (UnboxedColumn _ column) (BoxedColumn _ other) =-    BoxedColumn-        Nothing-        ( VB.generate-            (min (VG.length column) (VG.length other))-            (\i -> (column VG.! i, other VG.! i))-        )-zipColumns (UnboxedColumn _ column) (UnboxedColumn _ other) = UnboxedColumn Nothing (VG.zip column other)-{-# INLINE zipColumns #-}--{- | Merge two columns using `These`. O(1): the sides are kept in their-native representation and 'These' values materialize on element access.--}-mergeColumns :: Column -> Column -> Column-mergeColumns = MergedColumn-{-# INLINE mergeColumns #-}---- | Decode a 'MergedColumn' into the eager @BoxedColumn (These a b)@ form.-materializeMerged :: Column -> Column-materializeMerged (MergedColumn colA colB) =-    mergeEager (materializeMerged colA) (materializeMerged colB)-materializeMerged c = c--mergedHead :: Column -> Column-mergedHead (MergedColumn a b) =-    materializeMerged (MergedColumn (takeColumn 1 a) (takeColumn 1 b))-mergedHead c = c--{- | The eager element-wise merge ('These' per row, boxed). Bitmaps are-honored for every representation pair: a null side yields 'This'/'That',-both-null is an error (the join kernels never produce such a row).--}-mergeEager :: Column -> Column -> Column-mergeEager colA colB = case (colA, colB) of-    (MergedColumn a b, _) -> mergeEager (mergeEager a b) colB-    (_, MergedColumn a b) -> mergeEager colA (mergeEager a b)-    (PackedText _ _, _) -> mergeEager (materializePacked colA) colB-    (_, PackedText _ _) -> mergeEager colA (materializePacked colB)-    (BoxedColumn bmA c1, BoxedColumn bmB c2) ->-        merged bmA bmB (VG.length c1) (VG.length c2) (c1 VG.!) (c2 VG.!)-    (BoxedColumn bmA c1, UnboxedColumn bmB c2) ->-        merged bmA bmB (VG.length c1) (VG.length c2) (c1 VG.!) (c2 VG.!)-    (UnboxedColumn bmA c1, BoxedColumn bmB c2) ->-        merged bmA bmB (VG.length c1) (VG.length c2) (c1 VG.!) (c2 VG.!)-    (UnboxedColumn bmA c1, UnboxedColumn bmB c2) ->-        merged bmA bmB (VG.length c1) (VG.length c2) (c1 VG.!) (c2 VG.!)-  where-    merged ::-        (Columnable a, Columnable b) =>-        Maybe Bitmap ->-        Maybe Bitmap ->-        Int ->-        Int ->-        (Int -> a) ->-        (Int -> b) ->-        Column-    merged bmA bmB lenA lenB atA atB =-        BoxedColumn Nothing $ VB.generate (min lenA lenB) $ \i ->-            case (validAt bmA i, validAt bmB i) of-                (True, True) -> These (atA i) (atB i)-                (True, False) -> This (atA i)-                (False, True) -> That (atB i)-                (False, False) -> error "mergeColumns: both null"-    validAt mbm i = maybe True (`bitmapTestBit` i) mbm-    {-# INLINE validAt #-}---- | An internal, column version of zipWith.-zipWithColumns ::-    forall a b c.-    (Columnable a, Columnable b, Columnable c) =>-    (a -> b -> c) -> Column -> Column -> Either DataFrameException Column-zipWithColumns f (UnboxedColumn bmL (column :: VU.Vector d)) (UnboxedColumn bmR (other :: VU.Vector e)) = case testEquality (typeRep @a) (typeRep @d) of-    Just Refl -> case testEquality (typeRep @b) (typeRep @e) of-        Just Refl-            | isNothing bmL-            , isNothing bmR ->-                pure $ case sUnbox @c of-                    STrue -> UnboxedColumn Nothing (VU.zipWith f column other)-                    SFalse -> fromVector $ VB.zipWith f (VG.convert column) (VG.convert other)-        _ -> zipWithColumnsGeneral f (UnboxedColumn bmL column) (UnboxedColumn bmR other)-    Nothing -> zipWithColumnsGeneral f (UnboxedColumn bmL column) (UnboxedColumn bmR other)--- TODO: mchavinda - reuse pattern from interpret where we augment the--- error at the end.-zipWithColumns f left right = zipWithColumnsGeneral f left right--zipWithColumnsGeneral ::-    forall a b c.-    (Columnable a, Columnable b, Columnable c) =>-    (a -> b -> c) -> Column -> Column -> Either DataFrameException Column-zipWithColumnsGeneral f left right = case toVector @a left of-    Left (TypeMismatchException context) ->-        Left $-            TypeMismatchException (context{callingFunctionName = Just "zipWithColumns"})-    Left e -> Left e-    Right left' -> case toVector @b right of-        Left (TypeMismatchException context) ->-            Left $-                TypeMismatchException (context{callingFunctionName = Just "zipWithColumns"})-        Left e -> Left e-        Right right' -> pure $ fromVector $ VB.zipWith f left' right'-{-# INLINE zipWithColumnsGeneral #-}-{-# INLINE zipWithColumns #-}---- writeColumn and freezeColumn' (CSV-ingest helpers) moved to--- DataFrame.IO.Internal.MutableColumn so the core column module does not--- need to depend on DataFrame.Internal.Parsing.--{- | Freeze a mutable column into an @Either Text a@ column: every recorded-null position becomes @Left rawText@ (preserving the original input), every-other position becomes @Right v@. Used by CSV readers under 'EitherRead' mode.--}-freezeColumnEither :: [(Int, T.Text)] -> MutableColumn -> IO Column-freezeColumnEither nulls (MBoxedColumn col) = do-    frozen <- VB.unsafeFreeze col-    let nullMap = nulls-    pure $-        BoxedColumn Nothing $-            VB.imap-                ( \i v -> case lookup i nullMap of-                    Just t -> Left t-                    Nothing -> Right v-                )-                frozen-freezeColumnEither nulls (MUnboxedColumn col) = do-    c <- VU.unsafeFreeze col-    let nullMap = nulls-    pure $-        BoxedColumn Nothing $-            VB.generate (VU.length c) $ \i ->-                case lookup i nullMap of-                    Just t -> Left t-                    Nothing -> Right (c VU.! i)-{-# INLINE freezeColumnEither #-}--{- | Promote a non-nullable column to a nullable one (add an all-valid bitmap).-No-op when already nullable.--}-ensureOptional :: Column -> Column-ensureOptional c@(MergedColumn _ _) = ensureOptional (materializeMerged c)-ensureOptional c@(BoxedColumn (Just _) _) = c-ensureOptional (BoxedColumn Nothing col) =-    BoxedColumn (Just (allValidBitmap (VB.length col))) col-ensureOptional c@(UnboxedColumn (Just _) _) = c-ensureOptional (UnboxedColumn Nothing col) =-    UnboxedColumn (Just (allValidBitmap (VU.length col))) col-ensureOptional c@(PackedText (Just _) _) = c-ensureOptional (PackedText Nothing p) =-    PackedText (Just (allValidBitmap (packedLength p))) p---- | Fills the end of a column, up to n, with null rows. Does nothing if column has length >= n.-expandColumn :: Int -> Column -> Column-expandColumn n c@(MergedColumn a b)-    | n <= min (columnLength a) (columnLength b) = c-    | otherwise = expandColumn n (materializeMerged c)-expandColumn n c@(PackedText _ p)-    | n <= packedLength p = c-    | otherwise = expandColumn n (materializePacked c)-expandColumn n column@(BoxedColumn bm col)-    | n <= VG.length col = column-    | otherwise =-        let extra = n - VG.length col-            newBm = case bm of-                Nothing -> Just (buildBitmapFromNulls n [VG.length col .. n - 1])-                Just b ->-                    Just-                        (bitmapConcat (VG.length col) b extra (VU.replicate ((extra + 7) `shiftR` 3) 0))-            newCol = col <> VB.replicate extra (errorWithoutStackTrace "expandColumn: null slot")-         in BoxedColumn newBm newCol-expandColumn n column@(UnboxedColumn bm col)-    | n <= VG.length col = column-    | otherwise =-        let extra = n - VG.length col-            newBm = case bm of-                Nothing -> Just (buildBitmapFromNulls n [VG.length col .. n - 1])-                Just b ->-                    Just-                        (bitmapConcat (VG.length col) b extra (VU.replicate ((extra + 7) `shiftR` 3) 0))-            newCol = runST $ do-                mv <- VUM.new n-                VU.imapM_ (VUM.unsafeWrite mv) col-                VU.unsafeFreeze mv-         in UnboxedColumn newBm newCol---- | Fills the beginning of a column, up to n, with null rows. Does nothing if column has length >= n.-leftExpandColumn :: Int -> Column -> Column-leftExpandColumn n c@(MergedColumn a b)-    | n <= min (columnLength a) (columnLength b) = c-    | otherwise = leftExpandColumn n (materializeMerged c)-leftExpandColumn n c@(PackedText _ p)-    | n <= packedLength p = c-    | otherwise = leftExpandColumn n (materializePacked c)-leftExpandColumn n column@(BoxedColumn bm col)-    | n <= VG.length col = column-    | otherwise =-        let extra = n - VG.length col-            origLen = VG.length col-            newBm = case bm of-                Nothing -> Just (buildBitmapFromNulls n [0 .. extra - 1])-                Just b ->-                    let nullPart = VU.replicate ((extra + 7) `shiftR` 3) 0-                     in Just (bitmapConcat extra nullPart origLen b)-            newCol =-                VB.replicate extra (errorWithoutStackTrace "leftExpandColumn: null slot") <> col-         in BoxedColumn newBm newCol-leftExpandColumn n column@(UnboxedColumn bm col)-    | n <= VG.length col = column-    | otherwise =-        let extra = n - VG.length col-            origLen = VG.length col-            newBm = case bm of-                Nothing -> Just (buildBitmapFromNulls n [0 .. extra - 1])-                Just b ->-                    let nullPart = VU.replicate ((extra + 7) `shiftR` 3) 0-                     in Just (bitmapConcat extra nullPart origLen b)-            newCol = runST $ do-                mv <- VUM.new n-                VU.imapM_ (\i x -> VUM.unsafeWrite mv (extra + i) x) col-                VU.unsafeFreeze mv-         in UnboxedColumn newBm newCol--{- | Concatenates two columns.-Returns Nothing if the columns are of different types.--}-concatColumns :: Column -> Column -> Either DataFrameException Column-concatColumns left right = case (left, right) of-    (MergedColumn _ _, _) -> concatColumns (materializeMerged left) right-    (_, MergedColumn _ _) -> concatColumns left (materializeMerged right)-    (PackedText _ _, _) -> concatColumns (materializePacked left) right-    (_, PackedText _ _) -> concatColumns left (materializePacked right)-    (BoxedColumn bmL l, BoxedColumn bmR r) -> case testEquality (typeOf l) (typeOf r) of-        Just Refl ->-            let newBm = case (bmL, bmR) of-                    (Nothing, Nothing) -> Nothing-                    (Just bl, Nothing) ->-                        Just-                            (bitmapConcat (VB.length l) bl (VB.length r) (allValidBitmap (VB.length r)))-                    (Nothing, Just br) ->-                        Just-                            (bitmapConcat (VB.length l) (allValidBitmap (VB.length l)) (VB.length r) br)-                    (Just bl, Just br) -> Just (bitmapConcat (VB.length l) bl (VB.length r) br)-             in pure (BoxedColumn newBm (l <> r))-        Nothing -> Left (mismatchErr (typeOf r) (typeOf l))-    (UnboxedColumn bmL l, UnboxedColumn bmR r) -> case testEquality (typeOf l) (typeOf r) of-        Just Refl ->-            let newBm = case (bmL, bmR) of-                    (Nothing, Nothing) -> Nothing-                    (Just bl, Nothing) ->-                        Just-                            (bitmapConcat (VU.length l) bl (VU.length r) (allValidBitmap (VU.length r)))-                    (Nothing, Just br) ->-                        Just-                            (bitmapConcat (VU.length l) (allValidBitmap (VU.length l)) (VU.length r) br)-                    (Just bl, Just br) -> Just (bitmapConcat (VU.length l) bl (VU.length r) br)-             in pure (UnboxedColumn newBm (l <> r))-        Nothing -> Left (mismatchErr (typeOf r) (typeOf l))-    _ -> Left (mismatchErr (typeOf right) (typeOf left))-  where-    mismatchErr ::-        forall (x :: Type) (y :: Type). TypeRep x -> TypeRep y -> DataFrameException-    mismatchErr ta tb =-        withTypeable ta $-            withTypeable tb $-                TypeMismatchException-                    ( MkTypeErrorContext-                        { userType = Right ta-                        , expectedType = Right tb-                        , callingFunctionName = Just "concatColumns"-                        , errorColumnName = Nothing-                        }-                    )--{- | Like 'concatColumns' but also combines columns of different types by wrapping-values in 'Either' (e.g. @[1,2]@ and @["a","b"]@ become-@[Left 1, Left 2, Right "a", Right "b"]@).--}--{- | O(n) Concatenate a list of same-type columns in a single allocation.-All columns must have the same constructor and element type (as they will-within a single Parquet column). Calls 'error' on mismatch.--}-concatManyColumns :: [Column] -> Column-concatManyColumns [] = fromList ([] :: [Maybe Int])-concatManyColumns [c] = c-concatManyColumns all'-    | any isMergedColumn all' =-        concatManyColumns (map materializeMerged all')-    | any isPackedText all' =-        concatManyColumns (map materializePacked all')-concatManyColumns (c0 : cs) = case c0 of-    BoxedColumn bm0 v0 ->-        let getCol (BoxedColumn bm v) = case testEquality (typeOf v0) (typeOf v) of-                Just Refl -> (bm, v)-                Nothing -> error "concatManyColumns: BoxedColumn type mismatch"-            getCol _ = error "concatManyColumns: column constructor mismatch"-            rest = map getCol cs-            allVecs = v0 : map snd rest-            allBms = bm0 : map fst rest-            newBm-                | all isNothing allBms = Nothing-                | otherwise =-                    let pairs = zip allVecs allBms-                        expandedBms = map (\(v, mb) -> fromMaybe (allValidBitmap (VB.length v)) mb) pairs-                        go b1 n1 b2 n2 = bitmapConcat n1 b1 n2 b2-                        concatBms [] = VU.empty-                        concatBms [(b, _v)] = b-                        concatBms ((b1, v1) : (b2, v2) : rest') =-                            let merged = go b1 (VB.length v1) b2 (VB.length v2)-                             in concatBms ((merged, v1 <> v2) : rest')-                     in Just $ concatBms (zip expandedBms allVecs)-         in BoxedColumn newBm (VB.concat allVecs)-    UnboxedColumn bm0 v0 ->-        let getCol (UnboxedColumn bm v) = case testEquality (typeOf v0) (typeOf v) of-                Just Refl -> (bm, v)-                Nothing -> error "concatManyColumns: UnboxedColumn type mismatch"-            getCol _ = error "concatManyColumns: column constructor mismatch"-            rest = map getCol cs-            allVecs = v0 : map snd rest-            allBms = bm0 : map fst rest-            newBm-                | all isNothing allBms = Nothing-                | otherwise =-                    let pairs = zip allVecs allBms-                        expandedBms = map (\(v, mb) -> fromMaybe (allValidBitmap (VU.length v)) mb) pairs-                        go b1 n1 b2 n2 = bitmapConcat n1 b1 n2 b2-                        concatBms [] = VU.empty-                        concatBms [(b, _)] = b-                        concatBms ((b1, v1) : (b2, v2) : rest') =-                            let merged = go b1 (VU.length v1) b2 (VU.length v2)-                             in concatBms ((merged, v1 <> v2) : rest')-                     in Just $ concatBms (zip expandedBms allVecs)-         in UnboxedColumn newBm (VU.concat allVecs)-    PackedText _ _ -> concatManyColumns (map materializePacked (c0 : cs))-    MergedColumn _ _ -> concatManyColumns (map materializeMerged (c0 : cs))--concatColumnsEither :: Column -> Column -> Column-concatColumnsEither l@(MergedColumn _ _) r =-    concatColumnsEither (materializeMerged l) r-concatColumnsEither l r@(MergedColumn _ _) =-    concatColumnsEither l (materializeMerged r)-concatColumnsEither l@(PackedText _ _) r = concatColumnsEither (materializePacked l) r-concatColumnsEither l r@(PackedText _ _) = concatColumnsEither l (materializePacked r)-concatColumnsEither (BoxedColumn bmL left) (BoxedColumn bmR right) = case testEquality (typeOf left) (typeOf right) of-    Nothing ->-        BoxedColumn Nothing $ fmap Left left <> fmap Right right-    Just Refl ->-        let newBm = case (bmL, bmR) of-                (Nothing, Nothing) -> Nothing-                (Just bl, Nothing) ->-                    Just-                        ( bitmapConcat-                            (VB.length left)-                            bl-                            (VB.length right)-                            (allValidBitmap (VB.length right))-                        )-                (Nothing, Just br) ->-                    Just-                        ( bitmapConcat-                            (VB.length left)-                            (allValidBitmap (VB.length left))-                            (VB.length right)-                            br-                        )-                (Just bl, Just br) -> Just (bitmapConcat (VB.length left) bl (VB.length right) br)-         in BoxedColumn newBm $ left <> right-concatColumnsEither (UnboxedColumn bmL left) (UnboxedColumn bmR right) = case testEquality (typeOf left) (typeOf right) of-    Nothing ->-        BoxedColumn Nothing $-            fmap Left (VG.convert left) <> fmap Right (VG.convert right)-    Just Refl ->-        let newBm = case (bmL, bmR) of-                (Nothing, Nothing) -> Nothing-                (Just bl, Nothing) ->-                    Just-                        ( bitmapConcat-                            (VU.length left)-                            bl-                            (VU.length right)-                            (allValidBitmap (VU.length right))-                        )-                (Nothing, Just br) ->-                    Just-                        ( bitmapConcat-                            (VU.length left)-                            (allValidBitmap (VU.length left))-                            (VU.length right)-                            br-                        )-                (Just bl, Just br) -> Just (bitmapConcat (VU.length left) bl (VU.length right) br)-         in UnboxedColumn newBm $ left <> right-concatColumnsEither (BoxedColumn _ left) (UnboxedColumn _ right) =-    BoxedColumn Nothing $ fmap Left left <> fmap Right (VG.convert right)-concatColumnsEither (UnboxedColumn _ left) (BoxedColumn _ right) =-    BoxedColumn Nothing $ fmap Left (VG.convert left) <> fmap Right right---- | Allocate a mutable column of size @n@ matching the constructor/type of the given column.-newMutableColumn :: Int -> Column -> IO MutableColumn-newMutableColumn n (BoxedColumn _ (_ :: VB.Vector a)) =-    MBoxedColumn <$> (VBM.new n :: IO (VBM.IOVector a))-newMutableColumn n (UnboxedColumn _ (_ :: VU.Vector a)) =-    MUnboxedColumn <$> (VUM.new n :: IO (VUM.IOVector a))-newMutableColumn n c@(PackedText _ _) = newMutableColumn n (materializePacked c)-newMutableColumn n c@(MergedColumn _ _) = newMutableColumn n (materializeMerged c)---- | Copy a column chunk into a mutable column starting at offset @off@.-copyIntoMutableColumn :: MutableColumn -> Int -> Column -> IO ()-copyIntoMutableColumn mv off c@(MergedColumn _ _) =-    copyIntoMutableColumn mv off (materializeMerged c)-copyIntoMutableColumn (MBoxedColumn (mv :: VBM.IOVector b)) off (BoxedColumn _ (v :: VB.Vector a)) =-    case testEquality (typeRep @a) (typeRep @b) of-        Just Refl -> VG.imapM_ (\i x -> VBM.unsafeWrite mv (off + i) x) v-        Nothing -> error "copyIntoMutableColumn: Boxed type mismatch"-copyIntoMutableColumn (MUnboxedColumn (mv :: VUM.IOVector b)) off (UnboxedColumn _ (v :: VU.Vector a)) =-    case testEquality (typeRep @a) (typeRep @b) of-        Just Refl -> VG.imapM_ (\i x -> VUM.unsafeWrite mv (off + i) x) v-        Nothing -> error "copyIntoMutableColumn: Unboxed type mismatch"-copyIntoMutableColumn mc off c@(PackedText _ _) =-    copyIntoMutableColumn mc off (materializePacked c)-copyIntoMutableColumn _ _ _ =-    error "copyIntoMutableColumn: constructor mismatch"---- | Freeze a mutable column into an immutable column.-freezeMutableColumn :: MutableColumn -> IO Column-freezeMutableColumn (MBoxedColumn mv) = BoxedColumn Nothing <$> VB.unsafeFreeze mv-freezeMutableColumn (MUnboxedColumn mv) = UnboxedColumn Nothing <$> VU.unsafeFreeze mv--{- | O(n) Converts a column to a list. Throws an exception if the wrong type is specified.--__Examples:__--@-> column = fromList [(1 :: Int), 2, 3, 4]-> toList @Int column-[1,2,3,4]-> toList @Double column-exception: ...-@--}-toList :: forall a. (Columnable a) => Column -> [a]-toList xs = case toVector @a xs of-    Left err -> throw err-    Right val -> VB.toList val--{- | Type-safe conversion of a column to a vector of element type @a@ (specify via-type application); 'Left' 'TypeMismatchException' when the column's type differs.-->>> toVector @Int @VU.Vector column-Right (unboxed vector of Ints)-->>> toVector @Text @VB.Vector column-Right (boxed vector of Text)--}-toVector ::-    forall a v.-    (VG.Vector v a, Columnable a) => Column -> Either DataFrameException (v a)-toVector col = case col of-    PackedText _ _ -> toVector (materializePacked col)-    MergedColumn _ _ -> toVector (materializeMerged col)-    BoxedColumn bm (inner :: VB.Vector c) ->-        -- Check if user wants Maybe c (nullable) or c directly-        case testEquality (typeRep @a) (typeRep @c) of-            Just Refl -> Right $ VG.convert inner-            Nothing ->-                -- Try: a = Maybe c-                case testEquality (typeRep @a) (typeRep @(Maybe c)) of-                    Just Refl ->-                        -- Use VB.generate to avoid fusion forcing null slots-                        let !n = VB.length inner-                            maybeVec = case bm of-                                Nothing -> VB.generate n (Just . VB.unsafeIndex inner)-                                Just bitmap -> VB.generate n $ \i ->-                                    if bitmapTestBit bitmap i then Just (VB.unsafeIndex inner i) else Nothing-                         in Right $ VG.convert maybeVec-                    Nothing ->-                        Left $-                            TypeMismatchException-                                ( MkTypeErrorContext-                                    { userType = Right (typeRep @a)-                                    , expectedType = Right (typeRep @c)-                                    , callingFunctionName = Just "toVector"-                                    , errorColumnName = Nothing-                                    }-                                )-    UnboxedColumn bm (inner :: VU.Vector c) ->-        case testEquality (typeRep @a) (typeRep @c) of-            Just Refl -> Right $ VG.convert inner-            Nothing ->-                case testEquality (typeRep @a) (typeRep @(Maybe c)) of-                    Just Refl ->-                        let maybeVec = case bm of-                                Nothing -> VB.generate (VU.length inner) (Just . VU.unsafeIndex inner)-                                Just bitmap -> VB.generate (VU.length inner) $ \i ->-                                    if bitmapTestBit bitmap i then Just (VU.unsafeIndex inner i) else Nothing-                         in Right $ VG.convert maybeVec-                    Nothing ->-                        Left $-                            TypeMismatchException-                                ( MkTypeErrorContext-                                    { userType = Right (typeRep @a)-                                    , expectedType = Right (typeRep @c)-                                    , callingFunctionName = Just "toVector"-                                    , errorColumnName = Nothing-                                    }-                                )---- Some common types we will use for numerical computing.--{- | Convert a column to an unboxed 'Double' vector, coercing numeric types-('realToFrac' for floats, 'fromIntegral' for integrals; nulls become @NaN@).-'Left' 'TypeMismatchException' when the column is not numeric.--}-toDoubleVector :: Column -> Either DataFrameException (VU.Vector Double)-toDoubleVector column =-    case column of-        PackedText _ _ -> toDoubleVector (materializePacked column)-        MergedColumn _ _ -> toDoubleVector (materializeMerged column)-        UnboxedColumn bm (f :: VU.Vector a) -> case testEquality (typeRep @a) (typeRep @Double) of-            Just Refl -> case bm of-                Nothing -> Right f-                Just bitmap -> Right $ VU.imap (\i x -> if bitmapTestBit bitmap i then x else read "NaN") f-            Nothing -> case sFloating @a of-                STrue ->-                    Right-                        ( VU.imap-                            ( \i x -> case bm of-                                Just bitmap | not (bitmapTestBit bitmap i) -> read "NaN"-                                _ -> realToFrac x-                            )-                            f-                        )-                SFalse -> case sIntegral @a of-                    STrue ->-                        Right-                            ( VU.imap-                                ( \i x -> case bm of-                                    Just bitmap | not (bitmapTestBit bitmap i) -> read "NaN"-                                    _ -> fromIntegral x-                                )-                                f-                            )-                    SFalse ->-                        Left $-                            TypeMismatchException-                                ( MkTypeErrorContext-                                    { userType = Right (typeRep @Double)-                                    , expectedType = Right (typeRep @a)-                                    , callingFunctionName = Just "toDoubleVector"-                                    , errorColumnName = Nothing-                                    }-                                )-        BoxedColumn bm (f :: VB.Vector a) -> case testEquality (typeRep @a) (typeRep @Integer) of-            Just Refl ->-                Right-                    ( VB.convert $-                        VB.imap-                            ( \i x -> case bm of-                                Just bitmap | not (bitmapTestBit bitmap i) -> read "NaN"-                                _ -> fromIntegral x-                            )-                            f-                    )-            Nothing ->-                Left $-                    TypeMismatchException-                        ( MkTypeErrorContext-                            { userType = Right (typeRep @Double)-                            , expectedType = Left (columnTypeString column) :: Either String (TypeRep ())-                            , callingFunctionName = Just "toDoubleVector"-                            , errorColumnName = Nothing-                            }-                        )--{- | Convert a column to an unboxed 'Float' vector, coercing numeric types (nulls-become @NaN@); 'Left' 'TypeMismatchException' when not numeric. Converting from-'Double' may lose precision.--}-toFloatVector :: Column -> Either DataFrameException (VU.Vector Float)-toFloatVector column =-    case column of-        PackedText _ _ -> toFloatVector (materializePacked column)-        MergedColumn _ _ -> toFloatVector (materializeMerged column)-        UnboxedColumn bm (f :: VU.Vector a) -> case testEquality (typeRep @a) (typeRep @Float) of-            Just Refl -> case bm of-                Nothing -> Right f-                Just bitmap -> Right $ VU.imap (\i x -> if bitmapTestBit bitmap i then x else read "NaN") f-            Nothing -> case sFloating @a of-                STrue ->-                    Right-                        ( VU.imap-                            ( \i x -> case bm of-                                Just bitmap | not (bitmapTestBit bitmap i) -> read "NaN"-                                _ -> realToFrac x-                            )-                            f-                        )-                SFalse -> case sIntegral @a of-                    STrue ->-                        Right-                            ( VU.imap-                                ( \i x -> case bm of-                                    Just bitmap | not (bitmapTestBit bitmap i) -> read "NaN"-                                    _ -> fromIntegral x-                                )-                                f-                            )-                    SFalse ->-                        Left $-                            TypeMismatchException-                                ( MkTypeErrorContext-                                    { userType = Right (typeRep @Float)-                                    , expectedType = Right (typeRep @a)-                                    , callingFunctionName = Just "toFloatVector"-                                    , errorColumnName = Nothing-                                    }-                                )-        BoxedColumn bm (f :: VB.Vector a) -> case testEquality (typeRep @a) (typeRep @Integer) of-            Just Refl ->-                Right-                    ( VB.convert $-                        VB.imap-                            ( \i x -> case bm of-                                Just bitmap | not (bitmapTestBit bitmap i) -> read "NaN"-                                _ -> fromIntegral x-                            )-                            f-                    )-            Nothing ->-                Left $-                    TypeMismatchException-                        ( MkTypeErrorContext-                            { userType = Right (typeRep @Float)-                            , expectedType = Left (columnTypeString column) :: Either String (TypeRep ())-                            , callingFunctionName = Just "toFloatVector"-                            , errorColumnName = Nothing-                            }-                        )--{- | Convert a column to an unboxed 'Int' vector, coercing numeric types-(floats are 'round'ed via banker's rounding); 'Left' 'TypeMismatchException'-when the column is not numeric. Does not support nullable columns.--}-toIntVector :: Column -> Either DataFrameException (VU.Vector Int)-toIntVector column =-    case column of-        PackedText _ _ -> toIntVector (materializePacked column)-        MergedColumn _ _ -> toIntVector (materializeMerged column)-        UnboxedColumn _ (f :: VU.Vector a) -> case testEquality (typeRep @a) (typeRep @Int) of-            Just Refl -> Right f-            Nothing -> case sFloating @a of-                STrue -> Right (VU.map (round . (realToFrac :: a -> Double)) f)-                SFalse -> case sIntegral @a of-                    STrue -> Right (VU.map fromIntegral f)-                    SFalse ->-                        Left $-                            TypeMismatchException-                                ( MkTypeErrorContext-                                    { userType = Right (typeRep @Int)-                                    , expectedType = Right (typeRep @a)-                                    , callingFunctionName = Just "toIntVector"-                                    , errorColumnName = Nothing-                                    }-                                )-        BoxedColumn _ (f :: VB.Vector a) -> case testEquality (typeRep @a) (typeRep @Integer) of-            Just Refl -> Right (VB.convert $ VB.map fromIntegral f)-            Nothing ->-                Left $-                    TypeMismatchException-                        ( MkTypeErrorContext-                            { userType = Right (typeRep @Int)-                            , expectedType = Left (columnTypeString column) :: Either String (TypeRep ())-                            , callingFunctionName = Just "toIntVector"-                            , errorColumnName = Nothing-                            }-                        )--toUnboxedVector ::-    forall a.-    (Columnable a, VU.Unbox a) => Column -> Either DataFrameException (VU.Vector a)-toUnboxedVector column =-    case column of-        UnboxedColumn _ (f :: VU.Vector b) -> case testEquality (typeRep @a) (typeRep @b) of-            Just Refl -> Right f-            Nothing ->-                Left $-                    TypeMismatchException-                        ( MkTypeErrorContext-                            { userType = Right (typeRep @a)-                            , expectedType = Right (typeRep @b)-                            , callingFunctionName = Just "toUnboxedVector"-                            , errorColumnName = Nothing-                            }-                        )-        _ ->-            Left $-                TypeMismatchException-                    ( MkTypeErrorContext-                        { userType = Right (typeRep @a)-                        , expectedType = Left (columnTypeString column) :: Either String (TypeRep ())-                        , callingFunctionName = Just "toUnboxedVector"-                        , errorColumnName = Nothing-                        }-                    )-{-# INLINE toUnboxedVector #-}---- Shared finaliser for the two parseUnboxedColumn* helpers.  Freezes--- the mutable data vector, and only materialises the bitmap when the--- column actually had nulls.-{-# INLINE finalizeParseResult #-}-finalizeParseResult ::-    (VU.Unbox a) =>-    VUM.STVector s a ->-    VUM.STVector s Word8 ->-    Bool ->-    ST s (Maybe (Maybe Bitmap, VU.Vector a))-finalizeParseResult values vmask anyNull-    | anyNull = do-        vs <- VU.unsafeFreeze values-        vm <- VU.unsafeFreeze vmask-        return (Just (Just (buildBitmapFromValid vm), vs))-    | otherwise = do-        vs <- VU.unsafeFreeze values-        return (Just (Nothing, vs))+{- |+Umbrella re-export of the column implementation. The module is split by+concern:++* "DataFrame.Internal.Column.Types" — type-level machinery ('Rep', 'SBool', ...)+* "DataFrame.Internal.Column.Base" — the 'Column' GADT and core definitions+* "DataFrame.Internal.Column.Properties" — predicates and introspection+* "DataFrame.Internal.Column.Conversion" — vector\/list conversions+* "DataFrame.Internal.Column.Operations" — bulk transformations++Import this module to get the whole surface; import a submodule directly when+you only need one layer.+-}+module DataFrame.Internal.Column (+    module DataFrame.Internal.Column.Base,+    module DataFrame.Internal.Column.Conversion,+    module DataFrame.Internal.Column.Operations,+    module DataFrame.Internal.Column.Properties,+    module DataFrame.Internal.Column.Types,+) where++import DataFrame.Internal.Column.Base+import DataFrame.Internal.Column.Conversion+import DataFrame.Internal.Column.Operations+import DataFrame.Internal.Column.Properties+import DataFrame.Internal.Column.Types