Implementing streaming support

.github/workflows/run-tests.yml

@@ -10,7 +10,7 @@ jobs:
   build:
     strategy:
       matrix:
-        ghc: ['8.8.4', '8.10.7', '9.2.8', '9.4.8', '9.6.7', '9.12.1']
+        ghc: ['8.8.4', '8.10.7', '9.2.8', '9.4.8', '9.6.7', '9.12.2']
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v2

ChangeLog.hs

@@ -10,13 +10,21 @@
 
 changeLog :: ChangeLog
 changeLog =
-  [ Version "0.5.2.0" "2022-09-18"
+  [ Version "0.5.3.0" "2025-10-24"
+    "Streaming support and maintenance"
+    [ "Add support streaming of large results (#200)"
+    , "Publish the website via GitHub pages (#201)"
+    , "Properly use cascading deletes with foreign keys (#191)"
+    , "Schema query validation fix for PostgreSQL (#183)"
+    , "Many small version compatibility fixes"
+    ]
+  , Version "0.5.2.0" "2022-09-18"
     "Quality of life improvements"
     [ "Add support for GHC versions 8.10-9.2"
     , "Add typed UUIDs"
     , "Allow literal rows in update queries. (#139)"
     , "Add support for UNION/UNION ALL. (#140)"
-    , "Suppor raw PostgreSQL connetion strings. (#136)"
+    , "Support raw PostgreSQL connetion strings. (#136)"
     , "Move to Docker and GitHub actions for testing."
     , "Drop support for GHC versions <8.8."
     , "Various bugfixes."
@@ -251,15 +259,15 @@
 
 -- | Get the current version.
 currentVersion :: [Version] -> String
 currentVersion = vVersion . head
 
 -- | Get the release date of the current version.
 currentDate :: [Version] -> String
 currentDate = vDate . head
 
 -- | Get the release summary of the current version.
 currentSummary :: [Version] -> String
 currentSummary = vDate . head
 
 -- | Get the git tag message for the current version, as a list of @-m <msg>@
 --   flags.
@@ -293,7 +301,7 @@
 
 -- | Get the latest entry in the changelog.
 latestEntry :: ChangeLog -> Version
 latestEntry = head
 
 -- | Get the release date from the latest entry in the changelog.
 latestReleaseDate :: ChangeLog -> String
@@ -359,7 +367,7 @@
       putStrLn "Changelog has no list of changes."
       exitFailure
 
     when (head version /= '0') $ do
       putStrLn "Major version is not 0; make upload will complain."
       exitFailure
 

README.md

@@ -80,7 +80,7 @@ If you want to contribute code, please consult the following checklist before
 sending a pull request:
 
 * Does the code build with a recent version of GHC?
-* Do all the tests pass?
+* Do all the tests pass? Run `make test` in the root of the repo.
 * Have you added any tests covering your code?
 
 If you want to contribute code but don't really know where to begin,

selda-build-tools.cabal

@@ -10,7 +10,7 @@ executable selda-changelog
   main-is: ChangeLog.hs
   build-depends:
     base     >=4.8 && <5,
-    time     >=1.5 && <1.10,
-    filepath >=1.4 && <1.5,
+    time     >=1.5 && <1.15,
+    filepath >=1.4 && <1.6,
     process  >=1.5 && <1.7
   default-language: Haskell2010

selda-postgresql/selda-postgresql.cabal

@@ -1,5 +1,5 @@
 name:                selda-postgresql
-version:             0.1.8.2
+version:             0.1.9.0
 synopsis:            PostgreSQL backend for the Selda database EDSL.
 description:         PostgreSQL backend for the Selda database EDSL.
                      Requires the PostgreSQL @libpq@ development libraries to be
@@ -23,16 +23,16 @@ library
     OverloadedStrings
     CPP
   build-depends:
-      base              >=4.9  && <5
-    , bytestring        >=0.9  && <0.13
-    , exceptions        >=0.8  && <0.11
-    , selda             >=0.5  && <0.6
-    , selda-json        >=0.1  && <0.2
-    , text              >=1.0  && <2.2
-    , postgresql-binary >=0.12 && <0.15
-    , postgresql-libpq  >=0.9  && <0.12
-    , time              >=1.5  && <1.16
-    , uuid-types        >=1.0  && <1.1
+      base              >=4.9   && <5
+    , bytestring        >=0.9   && <0.13
+    , exceptions        >=0.8   && <0.11
+    , selda             >=0.5.3 && <0.6
+    , selda-json        >=0.1   && <0.2
+    , text              >=1.0   && <2.2
+    , postgresql-binary >=0.12  && <0.15
+    , postgresql-libpq  >=0.9   && <0.12
+    , time              >=1.5   && <1.16
+    , uuid-types        >=1.0   && <1.1
   hs-source-dirs:
     src
   default-language:

selda-postgresql/src/Database/Selda/PostgreSQL.hs

@@ -19,7 +19,7 @@ import Control.Monad.Catch
 import Control.Monad.IO.Class
 
 #ifndef __HASTE__
-import Control.Monad (void)
+import Control.Monad (void, unless)
 import qualified Data.ByteString as BS (foldl')
 import qualified Data.ByteString.Char8 as BS (pack, unpack)
 import Data.Dynamic
@@ -200,14 +200,17 @@ pgPPConfig = defPPConfig
 pgBackend :: Connection   -- ^ PostgreSQL connection object.
           -> SeldaBackend PG
 pgBackend c = SeldaBackend
-  { runStmt         = \q ps -> right <$> pgQueryRunner c False q ps
-  , runStmtWithPK   = \q ps -> left <$> pgQueryRunner c True q ps
-  , prepareStmt     = pgPrepare c
-  , runPrepared     = pgRun c
-  , getTableInfo    = pgGetTableInfo c . rawTableName
-  , backendId       = PostgreSQL
-  , ppConfig        = pgPPConfig
-  , closeConnection = \_ -> finish c
+  { runStmt          = \q ps -> right <$> pgQueryRunner c False q ps
+  , runStmtStreaming = \q ps k -> pgQueryRunnerStream c q ps k
+  , runStmtWithPK    = \q ps -> left <$> pgQueryRunner c True q ps
+  , prepareStmt      = pgPrepare c
+  , runPrepared      = pgRun c
+  , runPreparedStreaming
+                     = pgRunStream c
+  , getTableInfo     = pgGetTableInfo c . rawTableName
+  , backendId        = PostgreSQL
+  , ppConfig         = pgPPConfig
+  , closeConnection  = \_ -> finish c
   , disableForeignKeys = disableFKs c
   }
   where
@@ -398,6 +401,64 @@ getRows res = do
   where
     bsToPositiveInt = BS.foldl' (\a x -> a*10+fromIntegral x-48) 0
 
+pgQueryRunnerStream ::
+     (MonadIO m, MonadMask m, Monoid r)
+  => Connection
+  -> T.Text
+  -> [Param]
+  -> ([[SqlValue]] -> m r)
+  -> m r
+pgQueryRunnerStream c q ps k = do
+  qsent <-
+    liftIO
+      $ sendQueryParams c (encodeUtf8 q) [fromSqlValue p | Param p <- ps] Binary
+  unless qsent . throwM $ DbError "sendQueryParams failed"
+  msent <- liftIO $ setSingleRowMode c -- TODO: chunked mode, but the library doesn't wrap setChunkedRowsMode. See https://github.com/haskellari/postgresql-libpq/pull/79 for progress.
+  unless msent . throwM $ DbError "setSingleRowMode failed"
+  streamRows c k mempty
+
+-- TODO this is prepared for streaming of prepared statements (not in frontend yet)
+pgRunStream ::
+     (MonadIO m, MonadMask m, Monoid r)
+  => Connection
+  -> Dynamic
+  -> [Param]
+  -> ([[SqlValue]] -> m r)
+  -> m r
+pgRunStream c hdl ps k = do
+  let Just sid = fromDynamic hdl :: Maybe StmtID
+  qsent <-
+    liftIO $ sendQueryPrepared c (BS.pack $ show sid) (map mkParam ps) Binary
+  unless qsent . throwM $ DbError "sendQueryParams failed"
+  msent <- liftIO $ setSingleRowMode c
+  unless msent . throwM $ DbError "setSingleRowMode failed"
+  streamRows c k mempty
+  where
+    mkParam (Param p) =
+      case fromSqlValue p of
+        Just (_, val, fmt) -> Just (val, fmt)
+        Nothing -> Nothing
+
+streamRows ::
+     (MonadIO m, MonadMask m, Monoid r)
+  => Connection
+  -> ([[SqlValue]] -> m r)
+  -> r
+  -> m r
+streamRows c k r = do
+  mres <- liftIO $ getResult c
+  case mres of
+    Just res -> do
+      result <-
+        liftIO $ do
+          rows <- ntuples res
+          cols <- nfields res
+          types <- mapM (ftype res) [0 .. cols - 1]
+          mapM (getRow res types cols) [0 .. rows - 1]
+      if null result
+        then pure r
+        else k result >>= streamRows c k . mappend r
+    Nothing -> throwM $ DbError "streaming getResult failed"
 
 -- | Get all columns for the given row.
 getRow :: Result -> [Oid] -> Column -> Row -> IO [SqlValue]

selda-sqlite/selda-sqlite.cabal

@@ -1,5 +1,5 @@
 name:                selda-sqlite
-version:             0.1.7.2
+version:             0.1.8.0
 synopsis:            SQLite backend for the Selda database EDSL.
 description:         Allows the Selda database EDSL to be used with SQLite
                      databases.
@@ -20,14 +20,14 @@ library
     GADTs
     CPP
   build-depends:
-      base          >=4.9 && <5
-    , selda         >=0.5 && <0.6
-    , text          >=1.0 && <2.2
+      base          >=4.9   && <5
+    , selda         >=0.5.3 && <0.6
+    , text          >=1.0   && <2.2
     , bytestring    >=0.10  && <0.13
     , direct-sqlite >=2.2   && <2.4
     , directory     >=1.2.2 && <1.4
     , exceptions    >=0.8   && <0.11
-    , time          >=1.5   && <1.13
+    , time          >=1.5   && <1.15
     , uuid-types    >=1.0   && <1.1
   hs-source-dirs:
     src

selda-sqlite/src/Database/Selda/SQLite.hs

@@ -3,6 +3,7 @@
 module Database.Selda.SQLite
   ( SQLite
   , withSQLite
+  , withSQLiteStreaming
   , sqliteOpen, seldaClose
   , sqliteBackend
   ) where
@@ -56,6 +57,10 @@ sqliteBackend _ = error "sqliteBackend called in JS context"
 #else
 withSQLite file m = bracket (sqliteOpen file) seldaClose (runSeldaT m)
 
+-- S.yield :: Monad m => a -> S.Stream (S.Of a) m ()
+withSQLiteStreaming :: (MonadMask m, MonadIO m, Monad m, Monoid (m b)) => FilePath -> SeldaT SQLite m b -> m b
+withSQLiteStreaming file m = bracket (sqliteOpen file) seldaClose (runSeldaT m)
+
 -- | Create a Selda backend using an already open database handle.
 --   This is useful for situations where you want to use some SQLite-specific
 --   functionality alongside Selda.
@@ -66,14 +71,17 @@ withSQLite file m = bracket (sqliteOpen file) seldaClose (runSeldaT m)
 --   Proceed with extreme caution.
 sqliteBackend :: Database -> SeldaBackend SQLite
 sqliteBackend db = SeldaBackend
-  { runStmt         = \q ps -> snd <$> sqliteQueryRunner db q ps
-  , runStmtWithPK   = \q ps -> fst <$> sqliteQueryRunner db q ps
-  , prepareStmt     = \_ _ -> sqlitePrepare db
-  , runPrepared     = sqliteRunPrepared db
-  , getTableInfo    = sqliteGetTableInfo db . fromTableName
-  , ppConfig        = defPPConfig {ppMaxInsertParams = Just 999}
-  , backendId       = SQLite
-  , closeConnection = \conn -> do
+  { runStmt          = \q ps -> snd <$> sqliteQueryRunner db q ps
+  , runStmtStreaming = \q ps -> sqliteQueryRunnerStreaming db q ps
+  , runStmtWithPK    = \q ps -> fst <$> sqliteQueryRunner db q ps
+  , prepareStmt      = \_ _ -> sqlitePrepare db
+  , runPrepared      = sqliteRunPrepared db
+  , runPreparedStreaming
+                     = sqliteRunPreparedStreaming db
+  , getTableInfo     = sqliteGetTableInfo db . fromTableName
+  , ppConfig         = defPPConfig {ppMaxInsertParams = Just 999}
+  , backendId        = SQLite
+  , closeConnection  = \conn -> do
       stmts <- allStmts conn
       flip mapM_ stmts $ \(_, stm) -> do
         finalize $ fromDyn stm (error "BUG: non-statement SQLite statement")
@@ -180,6 +188,24 @@ sqliteRunPrepared db hdl params = do
     Left e@(SQLError{}) -> throwM (SqlError (show e))
     Right res           -> return (snd res)
 
+sqliteRunPreparedStreaming ::
+     (MonadIO m, MonadMask m, Monoid r)
+  => Database
+  -> Dynamic
+  -> [Param]
+  -> ([[SqlValue]] -> m r)
+  -> m r
+sqliteRunPreparedStreaming _ hdl params k = do
+  -- note for self: this does not need the `db` handle because it does not
+  -- return the last-inserted row ID.
+  eres <- try $ do
+    let Just stm = fromDynamic hdl
+    sqliteRunStmtStreaming stm params k `finally` do
+      liftIO $ clearBindings stm >> reset stm
+  case eres of
+    Left e@(SQLError{}) -> throwM (SqlError (show e))
+    Right res           -> return res
+
 sqliteQueryRunner :: Database -> QueryRunner (Int64, (Int, [[SqlValue]]))
 sqliteQueryRunner db qry params = do
     eres <- try $ do
@@ -207,6 +233,48 @@ getRows s acc = do
       getRows s (cs : acc)
     _ -> do
       return $ reverse acc
+sqliteQueryRunnerStreaming ::
+     (MonadIO m, MonadMask m, MonadIO m, Monoid r)
+  => Database
+  -> Text
+  -> [Param]
+  -> ([[SqlValue]] -> m r)
+  -> m r
+sqliteQueryRunnerStreaming db qry params k = do
+  eres <-
+    try $ do
+      stm <- liftIO $ prepare db qry
+      sqliteRunStmtStreaming stm params k `finally` do
+        liftIO $ finalize stm
+  case eres of
+    Left e@(SQLError {}) -> throwM (SqlError (show e))
+    Right res -> return res
+
+sqliteRunStmtStreaming ::
+     (MonadIO m, MonadMask m, Monoid r)
+  => Statement
+  -> [Param]
+  -> ([[SqlValue]] -> m r)
+  -> m r
+sqliteRunStmtStreaming stm params k = do
+  liftIO $ bind stm [toSqlData p | Param p <- params]
+  streamRows stm 1024 $ k . map (map fromSqlData)
+
+streamRows ::
+     (MonadIO m, Monoid r) => Statement -> Int -> ([[SQLData]] -> m r) -> m r
+streamRows s n k = cont mempty
+  where
+    cont r = go r [] 0
+    go r acc i
+      | i < n = do
+        res <- liftIO $ step s
+        case res of
+          Row -> do
+            cs <- liftIO $ columns s
+            go r (cs : acc) (succ i)
+          _ -> mappend r <$> send acc
+      | otherwise = mappend r <$> send acc >>= cont
+    send = k . reverse
 
 toSqlData :: Lit a -> SQLData
 toSqlData (LInt32 i)    = SQLInteger $ fromIntegral i

selda-tests/selda-tests.cabal

@@ -33,7 +33,7 @@ test-suite selda-testsuite
     Tests.Validation
   build-depends:
       aeson
-    , base       >=4.8  && <5
+    , base       >=4.10  && <5
     , bytestring >=0.10 && <0.13
     , directory  >=1.2  && <1.4
     , exceptions >=0.8  && <0.11

selda-tests/test/Tests/Query.hs

@@ -67,6 +67,7 @@ queryTests run = test
   , "nonNull" ~: run nonNullYieldsEmptyResult
   , "rawQuery1" ~: run rawQuery1Works
   , "rawQuery" ~: run rawQueryWorks
+  , "rawQueryStreaming" ~: run rawQueryStreamingWorks
   , "union" ~: run unionWorks
   , "union discards dupes" ~: run unionDiscardsDupes
   , "union works for whole rows" ~: run unionWorksForWholeRows
@@ -620,6 +621,13 @@ rawQueryWorks = do
   let correct = [p | p <- peopleItems, name p == "Link"]
   assEq "wrong name list returned" correct ppl
 
+rawQueryStreamingWorks = do
+  let q = rawQuery ["name", "age", "pet", "cash"]
+                  ("SELECT * FROM people WHERE name = " <> injLit ("Link"::Text))
+  ppl <- forQuery q $ pure . (:[])
+  let correct = [p | p <- peopleItems, name p == "Link"]
+  assEq "wrong name list returned" correct ppl
+
 unionWorks = assQueryEq "wrong name list returned" correct $ do
     let ppl = pName `from` select people
         pets = (pPet `from` select people) >>= nonNull

selda/selda.cabal

@@ -1,5 +1,5 @@
 name:                selda
-version:             0.5.2.1
+version:             0.5.3.0
 synopsis:            Multi-backend, high-level EDSL for interacting with SQL databases.
 description:         This package provides an EDSL for writing portable, type-safe, high-level
                      database code. Its feature set includes querying and modifying databases,
@@ -18,7 +18,8 @@ maintainer:          anton@ekblad.cc
 category:            Database
 build-type:          Simple
 cabal-version:       >=1.10
-tested-with:         GHC == 8.8.2, GHC == 8.10.1, GHC == 9.2.5, GHC == 9.4.4, GHC == 9.6.7, GHC == 9.12.1
+tested-with:         GHC == 8.8.2, GHC == 8.10.1, GHC == 9.2.5, GHC == 9.4.4, GHC == 9.6.7, GHC == 9.12.2
+
 source-repository head
   type:     git
   location: https://github.com/valderman/selda.git