Fix some Fold/Scan bugs, minor doc issues

.github/workflows/haskell.yml

@@ -371,16 +371,28 @@ jobs:
     # and SDK come from the same toolchain. packcheck clears the environment,
     # so SDKROOT or DEVELOPER_DIR cannot be used instead.
     #
-    # GHC records the SDK library path when it is installed, so the GHC and
-    # dependency cache keys include the Xcode version and build. A runner
-    # image with a different Xcode then installs GHC afresh.
+    # The GHC bindists for macOS hardcode the CLT SDK path from their build
+    # machine, e.g. the rts and ghci package confs of GHC 9.14.1 have
+    # /Library/Developer/CommandLineTools/SDKs/MacOSX.sdk/usr/lib in
+    # library-dirs, so the linker uses the CLT SDK even when Xcode is
+    # selected. Point the CLT MacOSX.sdk symlink at the Xcode SDK.
+    #
+    # The GHC and dependency cache keys include the Xcode version and build,
+    # so a runner image with a different Xcode installs GHC afresh.
     - name: Select default Xcode (macOS)
       id: xcode
       if: runner.os == 'macOS'
       run: |
         sudo xcode-select -s /Applications/Xcode.app/Contents/Developer
         xcode-select -p
-        xcrun --show-sdk-path
+        sdk=$(xcrun --show-sdk-path)
+        echo "$sdk"
+        clt=/Library/Developer/CommandLineTools/SDKs/MacOSX.sdk
+        if test -L "$clt"; then
+          ls -ld "$clt"
+          sudo ln -sfn "$sdk" "$clt"
+          ls -ld "$clt"
+        fi
         ver=$(xcodebuild -version | awk 'NR==1{v=$2} /Build version/{b=$3} END{print v "-" b}')
         echo "toolchain=-xcode-$ver" >> "$GITHUB_OUTPUT"
 

core/docs/Changelog.md

@@ -23,6 +23,9 @@
   separator, e.g. separator `"abc"` and input `"abcab"`.
 * Fix `chunksOf` in the array modules emitting a trailing empty array when
   the stream length is a multiple of the chunk size.
+* Fix `Fold.classifyScan`, `classifyScanIO`, `classify` and `classifyIO`
+  restarting the fold for a key that terminated on its first input.
+* Fix the `Arrow` instance of `Scan` failing when a scan skips an input.
 
 * On POSIX platforms, the `Path` based APIs that open an existing file in
   `WriteMode`, e.g. `write` and `withFile` in `Streamly.FileSystem.FileIO`,
@@ -69,6 +72,8 @@
   and not running the release action if the stream ends there.
 * Fix `Fold.finallyIO` running the action only on GC when the inner fold
   terminates.
+* Fix `Fold.distributeScan` losing the result of a fold that terminates on
+  the last input.
 
 ## 0.3.1 (May 2026)
 

core/src/Streamly/Data/Scan.hs

@@ -28,7 +28,7 @@
 -- >>> scan1 = Scan.function (\x -> x * x)
 -- >>> scan2 = Scan.function (\x -> 3 * x)
 -- >>> scan3 = Scan.teeWith (+) scan1 scan2 -- Compute x^2 + 3x
--- >>> scan4 = Scan.compose scan1 scan3     -- compute x^2 then pass it to scan3
+-- >>> scan4 = Scan.compose scan3 scan1     -- compute x^2 then pass it to scan3
 --
 -- >>> :{
 -- main =
@@ -38,6 +38,11 @@
 --       & Stream.fold (Fold.drainMapM print) -- IO ()
 -- :}
 --
+-- >>> main
+-- 8
+-- 32
+-- 112
+--
 -- @scan3@ splits the computation into two parts, one part computes @x^2@ using
 -- @scan1@ and the other part computes @3x@ using @scan2@ and then it zips the
 -- two parts back into a single stream by summing them. @scan4@ first passes an

core/src/Streamly/Internal/Data/Fold/Combinators.hs

@@ -617,12 +617,12 @@ the = fromScanl Scanl.the
 ------------------------------------------------------------------------------
 
 -- | Determine the sum of all elements of a stream of numbers. Returns additive
--- identity (@0@) when the stream is empty. Note that this is not numerically
--- stable for floating point numbers.
+-- identity (@0@) when the stream is empty. Uses Kahan-Babuska-Neumaier
+-- summation, which is numerically stable for floating point numbers.
 --
 -- >>> sum = Fold.fromScanl (Scanl.cumulativeScan Scanl.incrSum)
 --
--- Same as following but numerically stable:
+-- Same as the following, except that these are not numerically stable:
 --
 -- >>> sum = Fold.foldl' (+) 0
 -- >>> sum = fmap Data.Monoid.getSum $ Fold.foldMap Data.Monoid.Sum
@@ -699,9 +699,9 @@ minimumBy cmp = foldl1' min'
 -- >>> minimum = Fold.foldl1' min
 --
 -- Same as the following but without a default minimum. The 'Min' Monoid uses the
--- 'maxBound' as the default maximum:
+-- 'maxBound' as the default minimum:
 --
--- >>> maximum = fmap Data.Semigroup.getMin $ Fold.foldMap Data.Semigroup.Min
+-- >>> minimum = fmap Data.Semigroup.getMin $ Fold.foldMap Data.Semigroup.Min
 --
 {-# INLINE minimum #-}
 minimum :: (Monad m, Ord a) => Fold m a (Maybe a)
@@ -1098,7 +1098,7 @@ elemIndex a = findIndex (== a)
 --
 -- Definition:
 --
--- >>> null = fmap isJust Fold.one
+-- >>> null = fmap (not . isJust) Fold.one
 --
 {-# INLINE null #-}
 null :: Monad m => Fold m a Bool
@@ -1778,7 +1778,7 @@ distribute = Prelude.foldr (teeWith (:)) (fromPure [])
 -- >>> ref <- newIORef [Fold.take 2 Fold.sum, Fold.take 2 Fold.length :: Fold IO Int Int]
 -- >>> gen = atomicModifyIORef ref (\xs -> ([], xs))
 -- >>> Stream.toList $ Stream.scanl (Fold.distributeScan gen) (Stream.enumerateFromTo 1 10)
--- [[],[],[],[2,3],[],[],[],[],[],[],[]]
+-- [[],[],[3,2],[],[],[],[],[],[],[],[]]
 --
 {-# INLINE distributeScan #-}
 distributeScan :: Monad m => m [Fold m a b] -> Scanl m a [b]
@@ -1794,7 +1794,11 @@ distributeScan getFolds = Scanl consume initial extract final
         case res of
             Partial fs -> do
               r <- step fs a
-              run (Tuple' (Fold step (return r) extr fin : ys) zs) xs a
+              case r of
+                  Partial fs1 ->
+                      let fld = Fold step (return (Partial fs1)) extr fin
+                       in run (Tuple' (fld : ys) zs) xs a
+                  Done b -> run (Tuple' ys (b : zs)) xs a
             Done b -> do
               run (Tuple' ys (b : zs)) xs a
 
@@ -2127,7 +2131,7 @@ with f comb g = f . comb g . lmap snd
 {-# INLINE sampleFromthen #-}
 sampleFromthen :: Monad m => Int -> Int -> Fold m a b -> Fold m a b
 sampleFromthen offset size =
-    with indexed filter (\(i, _) -> (i + offset) `mod` size == 0)
+    with indexed filter (\(i, _) -> i >= offset && (i - offset) `mod` size == 0)
 
 ------------------------------------------------------------------------------
 -- Nesting

core/src/Streamly/Internal/Data/Fold/Container.hs

@@ -880,7 +880,7 @@ demuxKvToContainer f = demuxerToContainer fst (fmap (fmap (lmap snd)) . f)
 --
 -- Definition:
 --
--- >>> demuxKvToMap f = Fold.demuxerToContainer fst (Fold.lmap snd . f)
+-- >>> demuxKvToMap f = Fold.demuxerToContainer fst (fmap (fmap (Fold.lmap snd)) . f)
 --
 -- Example:
 --
@@ -938,7 +938,7 @@ classifyGeneric f (Fold step1 initial1 extract1 final1) =
                           Partial s1 ->
                             Tuple3' (IsMap.mapInsert k s1 kv) set Nothing
                           Done b ->
-                            Tuple3' kv set (Just (k, b))
+                            Tuple3' kv (Set.insert k set) (Just (k, b))
               Done b -> return (Tuple3' kv (Set.insert k set) (Just (k, b)))
 
     step (Tuple3' kv set _) a = do
@@ -1047,7 +1047,7 @@ classifyScanGeneric f (Fold step1 initial1 extract1 final1) =
                           Partial s1 ->
                             Tuple3' (IsMap.mapInsert k s1 kv) set Nothing
                           Done b ->
-                            Tuple3' kv set (Just (k, b))
+                            Tuple3' kv (Set.insert k set) (Just (k, b))
               Done b -> return (Tuple3' kv (Set.insert k set) (Just (k, b)))
 
     step (Tuple3' kv set _) a = do
@@ -1137,7 +1137,7 @@ classifyGenericIO f (Fold step1 initial1 extract1 final1) =
                         ref <- liftIO $ newIORef s1
                         return $ Tuple3' (IsMap.mapInsert k ref kv) set Nothing
                       Done b ->
-                        return $ Tuple3' kv set (Just (k, b))
+                        return $ Tuple3' kv (Set.insert k set) (Just (k, b))
               Done b -> return (Tuple3' kv (Set.insert k set) (Just (k, b)))
 
     step (Tuple3' kv set _) a = do
@@ -1257,7 +1257,7 @@ classifyScanGenericIO f (Fold step1 initial1 extract1 final1) =
                         ref <- liftIO $ newIORef s1
                         return $ Tuple3' (IsMap.mapInsert k ref kv) set Nothing
                       Done b ->
-                        return $ Tuple3' kv set (Just (k, b))
+                        return $ Tuple3' kv (Set.insert k set) (Just (k, b))
               Done b -> return (Tuple3' kv (Set.insert k set) (Just (k, b)))
 
     step (Tuple3' kv set _) a = do

core/src/Streamly/Internal/Data/Fold/Window.hs

@@ -294,7 +294,7 @@ windowRange n = Fold step initial extract extract
 
     initial =
         if n <= 0
-        then error "ringsOf: window size must be > 0"
+        then error "windowRange: window size must be > 0"
         else do
             arr :: MutArray.MutArray a <- liftIO $ MutArray.emptyOf n
             return $ Partial $ Tuple3Fused' (MutArray.arrContents arr) 0 0

core/src/Streamly/Internal/Data/Scan.hs

@@ -435,7 +435,11 @@ unzipMay (Scan stepL initialL) (Scan stepR initialR) =
 -- output. Other outputs are filtered out.
 {-# INLINE_NORMAL unzip #-}
 unzip :: Monad m => Scan m a x -> Scan m b y -> Scan m (a, b) (x, y)
-unzip s1 s2 = fmap (fromJust Prelude.. f) $ unzipMay s1 s2
+unzip s1 s2 =
+    fmap fromJust
+        $ compose (filter isJust)
+        $ fmap f
+        $ unzipMay s1 s2
 
     where
 

test/Streamly/Test/Data/Fold/Combinators.hs

@@ -408,14 +408,29 @@ takeEndBySeq_ =
         (Stream.fromList [1,2,0,0,3,4])
     `shouldReturn` [1,2]
 
+sampleFromthenOffset :: Expectation
+sampleFromthenOffset =
+    Stream.fold (Fold.sampleFromthen 1 3 Fold.toList)
+        (Stream.fromList [0..9 :: Int])
+    `shouldReturn` [1,4,7]
+
 distributeScan :: Expectation
 distributeScan = do
     ref <- newIORef [Fold.take 2 Fold.sum, Fold.take 2 Fold.length :: Fold.Fold IO Int Int]
     let gen = atomicModifyIORef ref (\xs -> ([], xs))
     r <- Stream.fold Fold.toList
              $ Stream.postscanl (Fold.distributeScan gen)
              $ Stream.fromList [1..5 :: Int]
-    r `shouldBe` [[], [], [2, 3], [], []]
+    r `shouldBe` [[], [3, 2], [], [], []]
+
+-- The results of the folds that terminate on the last input must not be lost.
+distributeScanLast :: Expectation
+distributeScanLast = do
+    ref <- newIORef [Fold.take 2 Fold.sum, Fold.take 2 Fold.length :: Fold.Fold IO Int Int]
+    let gen = atomicModifyIORef ref (\xs -> ([], xs))
+    Stream.fold (Fold.fromScanl (Fold.distributeScan gen))
+        (Stream.fromList [1, 2 :: Int])
+        `shouldReturn` [3, 2]
 
 unzipWithFstM :: Property
 unzipWithFstM =
@@ -499,6 +514,8 @@ main = hspec $ do
         it "takeEndBySeq" takeEndBySeq
         it "takeEndBySeq_" takeEndBySeq_
         it "distributeScan" distributeScan
+        it "distributeScan last input" distributeScanLast
+        it "sampleFromthen offset" sampleFromthenOffset
         prop "unzipWithFstM" unzipWithFstM
         prop "unzipWithMinM" unzipWithMinM
         it "partitionBy" partitionBy

test/Streamly/Test/Data/Fold/Container.hs

@@ -101,6 +101,29 @@ classifyScan =
         (Stream.postscanlMaybe (Fold.classifyScan getKey innerFold) input)
         `shouldReturn` [("ONE", 4), ("TWO", 6)]
 
+-- The inner fold terminates on the first input of a key, further inputs of
+-- that key must be ignored.
+classifyScanDoneOnFirst :: Expectation
+classifyScanDoneOnFirst = do
+    let innerFold = Fold.lmap snd (Fold.take 1 Fold.sum)
+        input = [("A", 1::Int), ("B", 5), ("A", 2), ("B", 6)]
+        expected = [("A", 1), ("B", 5)]
+        run scan = Stream.fold Fold.toList
+            $ Stream.postscanlMaybe scan (Stream.fromList input)
+        snd2 = fmap snd
+    run (Fold.classifyScan fst innerFold) `shouldReturn` expected
+    run (Fold.classifyScanIO fst innerFold) `shouldReturn` expected
+    Stream.fold Fold.toList
+        (Stream.catMaybes
+            $ Stream.postscan (snd2 (Fold.classify fst innerFold))
+            $ Stream.fromList input)
+        `shouldReturn` expected
+    Stream.fold Fold.toList
+        (Stream.catMaybes
+            $ Stream.postscan (snd2 (Fold.classifyIO fst innerFold))
+            $ Stream.fromList input)
+        `shouldReturn` expected
+
 frequency :: Expectation
 frequency =
     Stream.fold Fold.frequency
@@ -183,6 +206,7 @@ main = hspec $ do
         it "classifyWith" classifyWith
         it "classify" classify
         it "classifyScan" classifyScan
+        it "classifyScanDoneOnFirst" classifyScanDoneOnFirst
         it "frequency" frequency
         it "demuxerToMap" demuxerToMap
         it "demuxerToContainer" demuxerToContainer

test/Streamly/Test/Data/Stream/Transform.hs

@@ -8,6 +8,7 @@
 
 module Streamly.Test.Data.Stream.Transform (main) where
 
+import Control.Arrow ((***))
 import Control.Exception (ErrorCall(..), catch)
 import Data.IORef (IORef, modifyIORef, newIORef, readIORef, writeIORef)
 import System.Timeout (timeout)
@@ -213,6 +214,15 @@ testScanr =
     toList (Stream.scanr Scan.sum (Stream.fromList [1, 2, 3 :: Int]))
         `shouldReturn` [1, 3, 6]
 
+-- An input is dropped when either scan skips it.
+testScanrUnzip :: Expectation
+testScanrUnzip =
+    toList
+        (Stream.scanr
+            (Scan.filter odd *** Scan.filter (> 1))
+            (Stream.fromList [(1, 1), (2, 2), (3, 3) :: (Int, Int)]))
+        `shouldReturn` [(3, 3)]
+
 testPipe :: Expectation
 testPipe =
     toList (Stream.pipe (Pipe.map (*2)) (Stream.fromList [1, 2, 3 :: Int]))
@@ -639,6 +649,7 @@ main = hspec
         prop "filter law: id (driver)" $ filterLawScanModifier id
         prop "filter law: sum" $ filterLawScan Scanl.sum
         it "scanr" testScanr
+        it "scanr (***)" testScanrUnzip
         it "pipe map" testPipe
         it "pipe filter" testPipeFilter