duckdb-simple 0.1.5.2 → 0.2.0.0

raw patch · 34 files changed

+3769/−1300 lines, 34 filesdep +dataframe-arrow-bridgedep +dataframe-coredep ~duckdb-ffidep ~textdep ~timePVP ok

version bump matches the API change (PVP)

Dependencies added: dataframe-arrow-bridge, dataframe-core

Dependency ranges changed: duckdb-ffi, text, time

API changes (from Hackage documentation)

Files

• CHANGELOG.md +137/−0
• README.md +130/−15
• bench/Main.hs +62/−0
• dataframe-test/Main.hs +105/−0
• duckdb-simple.cabal +74/−16
• leaktest/Main.hs +217/−16
• src/Database/DuckDB/Simple.hs +96/−448
• src/Database/DuckDB/Simple/Arrow.hs +51/−0
• src/Database/DuckDB/Simple/Arrow/Internal.hs +91/−0
• src/Database/DuckDB/Simple/Callback.hs +97/−0
• src/Database/DuckDB/Simple/Catalog.hs +36/−34
• src/Database/DuckDB/Simple/Config.hs +35/−34
• src/Database/DuckDB/Simple/Copy.hs +43/−87
• src/Database/DuckDB/Simple/Deprecated/Streaming.hs +87/−0
• src/Database/DuckDB/Simple/FileSystem.hs +48/−32
• src/Database/DuckDB/Simple/FromField.hs +73/−36
• src/Database/DuckDB/Simple/Function.hs +94/−167
• src/Database/DuckDB/Simple/Generic.hs +49/−26
• src/Database/DuckDB/Simple/Internal.hs +175/−36
• src/Database/DuckDB/Simple/Logging.hs +32/−39
• src/Database/DuckDB/Simple/LogicalRep.hs +67/−86
• src/Database/DuckDB/Simple/Materialize.hs +33/−62
• src/Database/DuckDB/Simple/Result.hs +240/−0
• src/Database/DuckDB/Simple/Time.hs +27/−0
• src/Database/DuckDB/Simple/ToField.hs +173/−141
• test/ArrowTests.hs +241/−0
• test/CancellationTests.hs +198/−0
• test/CoreRegressionTests.hs +153/−0
• test/ExtensionRegressionTests.hs +304/−0
• test/Properties.hs +15/−16
• test/Spec.hs +24/−9
• test/StreamingTests.hs +162/−0
• test/TimeTests.hs +105/−0
• test/ValueRegressionTests.hs +295/−0

@@ -1,5 +1,142 @@ # Changelog +## 0.2.0.0++### Query execution and resource lifetime++- Add `Database.DuckDB.Simple.Deprecated.Streaming` for callers that need native+  streaming. It provides row folds, cursors, and Arrow folds with a deprecation+  warning. Both execution modes share decoding and resource cleanup. Native+  chunk fetching is interruptible, including cleanup of a chunk fetched just+  before cancellation. Default APIs continue to use materialized execution.+- Previously, a long native query could defer Ctrl-C until execution finished.+  Query preparation and execution now run in a worker so the caller can receive+  asynchronous exceptions. Cancellation interrupts DuckDB and waits for the+  worker before releasing its resources. Prompt cancellation requires the+  threaded RTS. Native code and Haskell callbacks must return before cleanup+  can finish.+- Add scoped Arrow batch export through `foldArrow` and `foldArrow_`. The+  callback receives a separate schema and array for each batch. Consumers may+  release or move them under the Arrow C Data Interface. The fold releases+  remaining contents on every exit path. This uses the supported schema/chunk+  conversion API and retains a materialized native result while the fold runs.+- The initial Arrow fold shared one borrowed schema across callbacks. Consumers+  such as `dataframe-arrow-bridge` release that schema when they import a batch,+  which made subsequent batches unusable. Each batch now has its own schema.+  Moved root objects remain usable after query and connection close; their+  consumer must release them.++- Previously, cursors decoded rows with prepare-time column types. Parameter+  binding or schema rebinding could change those types and cause truncated+  values or invalid memory access. Cursors now read types from the executed+  result. This addresses the streaming failures in #18.+- Cursors use the supported materialized execution API. DuckDB 1.5 provides+  native streaming only through deprecated entry points, which the default+  interface does not use. Folds decode one row at a time, but native result+  memory still depends on the result size. Fetch failures now produce SQL+  errors; previously, they were indistinguishable from end of input.+- Previously, reading past EOF could execute the statement again, including+  INSERT statements. Cursors now remain exhausted until an explicit reset.+  Clearing bindings also destroys any active result.+- Previously, exceptions during result decoding could skip native destruction.+  Results, chunks, and nested logical types now have exception-safe cleanup.+  Cursor state records chunk ownership before decoding and clears it before+  destruction, so cancellation cannot cause a leak or a second destruction.+- Previously, GC could finalize a connection or statement during its last+  native call. The binding now keeps the Haskell owner alive until the call+  returns. Reads through a closed parent connection fail before native access.+- Previously, streaming rejected STRUCT and UNION columns even though the+  shared decoder supported them. Eager queries and cursors now use the same+  row decoder, including nested collections and NULLs.+- Previously, a rollback failure could replace the exception from the user's+  transaction. The original exception is now preserved. A failed commit also+  attempts rollback.++### Callbacks and native helpers++- Previously, failed scalar, COPY, or logging registration could free callback+  resources twice. Registration now transfers ownership once and releases+  resources acquired before a failure. Static destructors avoid allocating+  a separate destructor callback for each registration.+- Previously, callback closures or query state could remain live on a long-lived+  connection after replacement or execution. Cleanup now releases scalar+  worker state, COPY state, and replaced closures before connection close.+- Previously, exceptions during callback initialization or error formatting+  could escape into native code. Scalar and COPY callbacks now report SQL+  errors. Logging callbacks contain exceptions because their API has no error+  channel. This includes asynchronous exceptions raised inside a callback.+- Previously, DuckDB could skip a scalar callback when an argument was NULL.+  Callbacks now receive those arguments. Use `Maybe` to accept NULL; a+  non-nullable Haskell argument produces a conversion error.+- Previously, Word and Word64 scalar results used signed BIGINT storage and+  could overflow. They now use UBIGINT. Float callbacks preserve NaN, infinity,+  and negative zero without depending on compiler optimization rules.+- Catalog, configuration, and filesystem helpers now bracket native allocations+  on failure paths. File reads reject sizes that cannot fit a Haskell buffer.+  Unsupported catalog entry kinds fail before the native lookup.++### Value conversion++- Previously, temporal infinity could abort the process or decode as an+  unrelated finite value. `Database.DuckDB.Simple.Time` now provides `Unbounded`,+  `Date`, `LocalTimestamp`, and `UTCTimestamp` to read and bind both infinities.+  Ordinary `Day`, `LocalTime`, and `UTCTime` report a conversion error for+  infinity. Floating-point NaN and infinities remain supported.+- `FieldDate`, `FieldTimestamp`, and `FieldTimestampTZ` now hold `Unbounded`+  payloads. Wrap existing finite payloads in `Finite`. Custom `FromField`+  instances can inspect infinity before conversion. Generic composites and+  nested collections preserve infinity and NULL separately.+- Previously, binding dates outside native storage limits could abort in C++.+  Finite date/time conversions now use checked epoch arithmetic and reject+  out-of-range inputs with Haskell exceptions.+- Previously, TIMESTAMP_S and TIMESTAMP_MS decoding multiplied Int64 values+  into microseconds and could overflow. Each timestamp family now retains its+  own units. Composite TIMESTAMP_S/MS/NS and TIME_NS values can be rebound+  without losing units, nanoseconds, or typed NULLs.+- Previously, UTCTime parameters had SQL type TIMESTAMP and could change their+  meaning under a non-UTC session timezone. They now have type TIMESTAMPTZ.+- Previously, Float parameters used DOUBLE, which hid a missing REAL decoder.+  Float parameters now use FLOAT, and REAL results decode to Float or Double.+  NaN, infinity, and negative zero remain supported. Narrowing a finite Double+  that exceeds Float's range now fails instead of producing infinity.+- Previously, Int8 and unsigned narrowing conversions could wrap out-of-range+  values. They now report conversion errors. Intermediate signed and unsigned+  values remain 64-bit until the target bounds have been checked.+- Previously, text parameters were terminated at an embedded NUL. Text and+  String parameters now pass their UTF-8 byte length and preserve NULs. SQL,+  native names, paths, and configuration strings reject NUL to prevent silent+  truncation. Native names and error messages are decoded as UTF-8.+- DECIMAL values retain their exact integer representation. Invalid width,+  scale, or magnitude now fails before native construction. Invalid ENUM+  indexes, UNION tags, and composite constructors also produce controlled errors.+- Previously, BIT padding could give incorrect SQL bit counts. Padding now+  follows DuckDB's representation; unsupported empty or malformed inputs fail+  before native use.+- Previously, generic records and sums decoded by position. They now match+  field and member names and reject incompatible schemas. NULL non-nullable+  products report a conversion failure instead of reaching a partial `error`.+- Previously, a NULL UNION payload could lose its declared member type. Typed+  NULL payloads now retain that type through native construction.+- TIMETZ offsets containing seconds now fail when decoding to Haskell's+  minute-based TimeZone. Invalid clock components and offsets fail before+  binding instead of being rounded or narrowed silently.++### Testing and compatibility++- Add an optional DataFrame integration suite with `-fdataframe-tests`. It+  imports several Arrow batches through `dataframe-arrow-bridge` and checks+  values, NULLs, column order, consumer failures, and use after connection close.+  CI runs it on Linux and macOS. The ordinary suite also checks consumption,+  ownership transfer, and cleanup after a consumer has released its objects.+- Add crash reproductions for #18, real native ownership tests, and sustained+  checks for long-lived connections, callback release, and cancellation.+  Property tests now include embedded NUL rather than filtering it out.+- Add repeatable benchmarks with checked results. Linux CI checks callback+  and cancellation workloads under Valgrind.+- Raise the minimum native DuckDB version to 1.5.3.+- Use GHC 9.14.1 by default. Test the latest stable patch release in each GHC+  series from 9.6 to 9.14.+ ## 0.1.5.2 - Fix a connection leak: `close` and the connection finalizer built the close action but then discarded it, so the DuckDB connection and database handles stayed open. Every leaked database instance also kept its own DuckDB thread pool alive. (Reported by @winitzki, see #15.) - Fix the same defect in `closeStatement`, which discarded the action that destroys the prepared statement. (Fixed by @bgamari in #15.)

@@ -1,7 +1,8 @@ # duckdb-simple  `duckdb-simple` provides a high-level Haskell interface to DuckDB inspired by-the ergonomics of [`sqlite-simple`](https://hackage.haskell.org/package/sqlite-simple).+the APIs of [`sqlite-simple`](https://hackage.haskell.org/package/sqlite-simple) and+[`postgresql-simple`](https://hackage.haskell.org/package/postgresql-simple). It builds on the low-level bindings exposed by [`duckdb-ffi`](../duckdb-ffi) and provides a focused API for opening connections, running queries, binding parameters, and decoding typed results—including the full set of DuckDB scalar@@ -79,8 +80,7 @@  DuckDB does not allow mixing positional and named placeholders within the same SQL statement; the library preserves DuckDB’s error message in that situation.-Savepoints are currently rejected by DuckDB, so `withSavepoint` raises an-`SQLError` describing the limitation.+DuckDB does not support savepoints. This library does not provide `withSavepoint`.  If the number of supplied parameters does not match the statement’s declared placeholders—or if you attempt to bind named arguments to a positional-only@@ -194,8 +194,34 @@   fmap fromOnly <$> query_ conn "SELECT vals FROM lists" ``` +### Infinite dates and timestamps++Use `Database.DuckDB.Simple.Time` when a column can contain temporal infinity.+Its `Date`, `LocalTimestamp`, and `UTCTimestamp` types wrap `Day`, `LocalTime`,+and `UTCTime` in `Unbounded`: `NegInfinity`, `Finite value`, or `PosInfinity`.+These types support parameters, results, and fields in generic composites.+`UTCTimestamp` binds as TIMESTAMPTZ.++```haskell+import Database.DuckDB.Simple.Time++infiniteDates :: Connection -> IO [Only Date]+infiniteDates conn = query conn "SELECT ?::DATE" (Only (PosInfinity :: Date))+```++The ordinary `Day`, `LocalTime`, and `UTCTime` instances reject infinity with+a conversion error. Use `Maybe Date` to distinguish SQL NULL from infinity.+Floating-point NaN and infinities remain valid `Float` and `Double` values.+ ### Manual STRUCT and UNION Handling +Temporal fields retain their SQL units when composite values are rebound.+The `FieldDate`, `FieldTimestamp`, and `FieldTimestampTZ` constructors hold+`Unbounded` values. Wrap finite payloads in `Finite` when constructing them.+For TIMESTAMP_S or TIMESTAMP_MS values outside the TIMESTAMP range, use an+explicit parameter cast, such as `SELECT ?::STRUCT(value TIMESTAMP_S)`.+DuckDB otherwise attempts to convert these parameters to microseconds.+ For more control, you can work directly with `StructValue` and `UnionValue` from `Database.DuckDB.Simple.LogicalRep`: @@ -220,23 +246,44 @@ - `withConnection` and `withStatement` wrap the open/close lifecycle and guard   against exceptions; use them whenever possible to avoid leaking C handles. - All intermediate DuckDB objects (results, prepared statements, values) are-  released immediately after use. Long queries still materialise their result-  sets when using the eager helpers; reach for `fold`/`fold_`/`foldNamed` (or-  the lower-level `nextRow`) to stream results in constant space.+  released immediately after use. Query helpers return a Haskell list of all+  rows. Folds and cursors decode rows incrementally, while DuckDB retains the+  materialized native result until it is exhausted, reset, or closed. - `execute`/`query` variants reset statement bindings each run so prepared   statements can be reused safely. +For concurrent workers, use a separate connection per worker. A connection+can move between threads or be shared when the application serializes access.+Hold that lock for the whole transaction or cursor lifetime, including `close`.+DuckDB serializes native query calls, but this does not protect the Haskell+handle state or prevent another call from interfering with an active cursor.+Statements and connections do not provide their own lock. A callback must not+execute another query on its active connection or close that connection.++Link your executable with `ghc-options: -threaded` to allow prompt cancellation+of native queries, including Ctrl-C. On cancellation, the library interrupts+DuckDB and waits for the native call to return before it releases resources+and propagates the exception. Cancellation is cooperative: native code and+Haskell callbacks must return before cleanup can finish.++Close or reset an abandoned statement to release its result. DuckDB can retain+native result buffers until that result is destroyed.+Use `withStatement` for manual iteration. `fold` releases the result after+success or an exception. The accumulator determines Haskell memory use.+ ### Metadata helpers  - `columnCount` and `columnName` expose prepared-statement metadata so you can   inspect result shapes before executing a query.-- `rowsChanged` tracks the number of rows affected by the most recent mutation-  on a connection. DuckDB does not offer a `lastInsertRowId`; prefer SQL-  `RETURNING` clauses when you need generated identifiers.-### Streaming Results+- `execute` returns the number of affected rows. Use SQL `RETURNING` clauses+  when you need generated identifiers.+### Cursors and folds -`fold`, `fold_`, and `foldNamed` expose DuckDB’s chunked result API, letting you-aggregate or stream rows without materialising the entire result set:+`fold`, `fold_`, and `foldNamed` decode one row at a time from DuckDB's result+chunks. DuckDB 1.5 materializes the native result before the first row is+returned. These functions avoid a complete Haskell row list, but native memory+use still depends on the result size. The native API for starting a streaming+result is deprecated; the default interface uses the supported execution API.  ```haskell import Database.DuckDB.Simple.Types (Only (..))@@ -250,12 +297,80 @@ For manual cursor-style iteration, use `nextRow`/`nextRowWith` on an open `Statement` to pull rows one at a time and decide when to stop. +Cursors support the same column types as eager queries, including STRUCT+and UNION values with nested collections and NULLs.++#### Optional native streaming++`Database.DuckDB.Simple.Deprecated.Streaming` provides `fold`, `fold_`,+`foldNamed`, `nextRow`, and `nextRowWith` with native streaming enabled.+Import it qualified:++```haskell+import qualified Database.DuckDB.Simple.Deprecated.Streaming as Streaming++streamSum :: Connection -> IO Int+streamSum conn =+  Streaming.fold_ conn "SELECT i FROM range(1000000) t(i)" 0 $ \acc (Only n) ->+    pure (acc + n)+```++The import emits a deprecation warning because DuckDB has deprecated the+execution entry point. DuckDB can still materialize some queries. Streaming+does not bound the memory used by query operators.++The first cursor fetch selects the execution mode until an explicit reset.+Switching between default and streaming `nextRow` calls retains that mode.+Keep the connection dedicated to the active stream; another query on that+connection can invalidate it. Cancellation interrupts native chunk fetching+as well as execution.++This module also provides `foldArrow` and `foldArrow_` for streaming Arrow+batches. They use the same supported Arrow conversion and scoped ownership+as `Database.DuckDB.Simple.Arrow`.++### Arrow batches++`Database.DuckDB.Simple.Arrow` provides `foldArrow` and `foldArrow_` for clients+that consume the Arrow C Data Interface. Each callback receives a separate+schema and array batch. It can read them or pass them to an Arrow consumer+that releases or moves them. The fold releases any remaining contents on+success, failure, or cancellation. The consumer owns any contents it moves.++The original pointers are valid only during the callback. To retain contents,+a consumer must move the root structs into its own storage and set the source+release fields to NULL. Moved contents remain valid after the query and+connection close. Empty results do not produce a callback.++For example, the `dataframe-arrow-bridge` package can copy each batch into a+Haskell `DataFrame` and release the Arrow objects:++```haskell+import qualified DataFrame.IO.Arrow as DataFrame+import qualified Database.DuckDB.Simple.Arrow as Arrow+import Foreign.Ptr (castPtr)++frames <- Arrow.foldArrow_ conn "SELECT id::BIGINT, name::VARCHAR FROM people" [] $ \acc schema array -> do+  frame <- DataFrame.arrowToDataframe (castPtr schema) (castPtr array)+  pure (frame : acc)+-- Reverse frames to recover the query's batch order.+```++The bridge currently imports signed 32-bit and 64-bit integers, Float, Double,+and text columns. It is a test dependency of this repository; applications+that use it must declare their own dependency on `dataframe-arrow-bridge`.++Arrow export uses DuckDB's schema and chunk conversion API. DuckDB materializes+the native result before callbacks start, so its memory use depends on the+result size. The older Arrow query and scan bindings remain available through+`Database.DuckDB.FFI.Deprecated` and emit deprecation warnings.+ ### Feature Coverage  - Connections, prepared statements, positional/named parameter binding. - High-level execution (`execute*`) and eager queries (`query*`, `queryNamed`).-- Streaming helpers (`fold`, `foldNamed`, `fold_`, `nextRow`) for constant-space-  result processing.+- Cursor and fold helpers (`fold`, `foldNamed`, `fold_`, `nextRow`) that decode+  native result chunks one row at a time. - Comprehensive scalar type support: signed/unsigned integers, HUGEINT/UHUGEINT,   decimals (with width/scale), intervals, precise and timezone-aware temporals,   enums, bit strings, blobs, bignums, and UUIDs.@@ -267,7 +382,7 @@ - User-defined scalar functions backed by Haskell functions (including IO and   nullable arguments). - Transaction helpers (`withTransaction`) and metadata accessors (`columnCount`,-  `columnName`, `rowsChanged`).+  `columnName`).  ## User-Defined Functions 

@@ -0,0 +1,62 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Repeatable end-to-end benchmarks with checked results.+module Main (main) where++import Control.Exception (evaluate)+import Control.Monad (forM_, replicateM, unless)+import Data.Int (Int64)+import qualified Data.List as List+import qualified Data.Text as Text+import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)+import Data.Time.LocalTime (LocalTime, localTimeToUTC, utc)+import Database.DuckDB.Simple+import GHC.Clock (getMonotonicTimeNSec)+import System.Environment (getArgs)+import System.Mem (performMajorGC)+import Text.Printf (printf)++-- | Run one workload seven times after a warm-up run.+main :: IO ()+main = do+    args <- getArgs+    let workload = case args of name : _ -> name; _ -> "eager"+        count = case args of _ : n : _ -> read n; _ -> 100000 :: Int64+        expected = count * (count - 1) `div` 2+    withConnectionWithConfig ":memory:" [("threads", "1")] $ \conn -> do+        createFunction conn "bench_identity" (id :: Int64 -> Int64)+        let sql = Query ("SELECT i FROM range(" <> Text.pack (show count) <> ") t(i)")+            action = case workload of+                "eager" -> do+                    rows <- query_ conn sql+                    evaluate (List.foldl' (\acc (Only n) -> acc + n) 0 rows)+                "fold" -> fold_ conn sql 0 (\acc (Only n) -> pure (acc + n))+                "scalar" -> do+                    rows <- query_ conn (Query ("SELECT bench_identity(i) FROM range(" <> Text.pack (show count) <> ") t(i)"))+                    evaluate (List.foldl' (\acc (Only n) -> acc + n) 0 rows)+                "text" -> do+                    rows <- query_ conn (Query ("SELECT repeat('duckdb λ text', 4) FROM range(" <> Text.pack (show count) <> ")"))+                    evaluate (List.foldl' (\acc (Only value) -> acc + fromIntegral (Text.length value)) 0 rows)+                "timestamp" -> do+                    rows <- query_ conn (Query ("SELECT TIMESTAMP '2000-01-01' + i * INTERVAL 1 SECOND FROM range(" <> Text.pack (show count) <> ") t(i)"))+                    evaluate (List.foldl' (\acc (Only (value :: LocalTime)) -> acc + floor (utcTimeToPOSIXSeconds (localTimeToUTC utc value))) 0 rows)+                "parameters" -> do+                    rows <- replicateM (fromIntegral count) (query conn "SELECT ?::BIGINT" (Only (1 :: Int64)))+                    evaluate (sum [n | [Only n] <- rows])+                _ -> fail "Expected eager, fold, scalar, text, timestamp, or parameters"+            expectedResult = case workload of+                "parameters" -> count+                "text" -> count * fromIntegral (Text.length (Text.replicate 4 "duckdb λ text"))+                "timestamp" -> count * 946684800 + expected+                _ -> expected+            check actual = unless (actual == expectedResult) (fail "benchmark result mismatch")+        action >>= check+        putStrLn "workload,rows,run,milliseconds,checksum"+        forM_ [1 .. 7 :: Int] $ \run -> do+            performMajorGC+            start <- getMonotonicTimeNSec+            result <- action+            end <- getMonotonicTimeNSec+            check result+            printf "%s,%d,%d,%.3f,%d\n" workload count run (fromIntegral (end - start) / 1000000 :: Double) result

@@ -0,0 +1,105 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-deprecations #-}++-- | Consume DuckDB Arrow batches with the dataframe library.+module Main (main) where++import Control.Exception (ErrorCall, displayException, try)+import Control.Monad (forM_)+import Data.Int (Int64)+import qualified Data.Text as Text+import qualified DataFrame.Core as DataFrame+import qualified DataFrame.IO.Arrow as DataFrameArrow+import Database.DuckDB.FFI (ArrowArray (..), ArrowSchema (..))+import Database.DuckDB.Simple+import qualified Database.DuckDB.Simple.Arrow as Arrow+import qualified Database.DuckDB.Simple.Deprecated.Streaming as Streaming+import Foreign.Ptr (Ptr, castPtr, nullFunPtr)+import Foreign.Storable (peek)+import Test.Tasty (TestTree, defaultMain, testGroup)+import Test.Tasty.HUnit++-- | Test both execution modes against the same Arrow consumer.+main :: IO ()+main = defaultMain $ testGroup "DataFrame Arrow import" [dataframeTests False, dataframeTests True]++-- | Check data and ownership across successful and failed imports.+dataframeTests :: Bool -> TestTree+dataframeTests streaming =+    testGroup+        (if streaming then "deprecated streaming" else "materialized")+        [ testCase "several batches remain readable after connection close" do+            (rowCount, batches) <- withDb \conn ->+                foldArrow+                    conn+                    "SELECT i::INTEGER AS id, CASE WHEN i % 3 = 0 THEN NULL ELSE i - 2500 END AS \"íslenska_λ\", (i::FLOAT / 4)::FLOAT AS small, CASE WHEN i % 5 = 0 THEN NULL ELSE i::DOUBLE / 4 END AS real, CASE WHEN i % 7 = 0 THEN NULL WHEN i % 7 = 1 THEN '' ELSE ? || i::VARCHAR END AS label, NULL::BIGINT AS missing FROM range(?::BIGINT) t(i)"+                    ("λ\0雪" :: Text.Text, 5000 :: Int64)+                    (0 :: Int, [])+                    \(offset, acc) schema array -> do+                        assertLiveSchema schema+                        count <- fromIntegral . arrowArrayLength <$> peek array+                        frame <- DataFrameArrow.arrowToDataframe (castPtr schema) (castPtr array)+                        assertConsumed schema array+                        pure (offset + count, (offset, count, frame) : acc)+            rowCount @?= 5000+            assertBool "multiple native batches" (length batches > 1)+            forM_ batches \(offset, count, frame) -> do+                let expected = expectedFrame [offset .. offset + count - 1]+                DataFrame.columnNames frame @?= DataFrame.columnNames expected+                frame @?= expected+        , testCase "an unsupported type releases the failed export" $+            withDb \conn -> do+                -- Import one supported column before the unsupported column fails.+                result <- try $ foldArrow conn "SELECT 42::BIGINT AS n, true AS unsupported" () () \() schema array -> do+                    _ <- DataFrameArrow.arrowToDataframe (castPtr schema) (castPtr array)+                    pure ()+                case result of+                    Left (err :: ErrorCall) ->+                        assertBool "DataFrame format error" ("unsupported format" `Text.isInfixOf` Text.pack (displayException err))+                    Right () -> assertFailure "expected an unsupported Arrow type"+                query_ conn "SELECT 42" >>= (@?= [Only (42 :: Int64)])+        , testCase "an empty result does not call the consumer" $+            withDb \conn -> do+                count <- foldArrow conn "SELECT 1::BIGINT AS n WHERE false" () (0 :: Int) \_ schema array -> do+                    _ <- DataFrameArrow.arrowToDataframe (castPtr schema) (castPtr array)+                    assertFailure "unexpected batch"+                count @?= 0+        ]+  where+    foldArrow :: (ToRow q) => Connection -> Query -> q -> a -> (a -> Ptr ArrowSchema -> Ptr ArrowArray -> IO a) -> IO a+    foldArrow = if streaming then Streaming.foldArrow else Arrow.foldArrow++-- | Construct the expected values independently of the Arrow buffer layout.+expectedFrame :: [Int] -> DataFrame.DataFrame+expectedFrame rows =+    DataFrame.fromNamedColumns+        [ ("id", DataFrame.fromList rows)+        , ("íslenska_λ", DataFrame.fromList [if i `rem` 3 == 0 then Nothing else Just (i - 2500) | i <- rows])+        , ("small", DataFrame.fromList [fromIntegral i / 4 :: Double | i <- rows])+        , ("real", DataFrame.fromList [if i `rem` 5 == 0 then Nothing else Just (fromIntegral i / 4 :: Double) | i <- rows])+        , ("label", DataFrame.fromList [label i | i <- rows])+        , ("missing", DataFrame.fromList [Nothing :: Maybe Int | _ <- rows])+        ]+  where+    label i+        | i `rem` 7 == 0 = Nothing+        | i `rem` 7 == 1 = Just ""+        | otherwise = Just ("λ\0雪" <> Text.pack (show i))++-- | Fail before the consumer can dereference a previously released schema.+assertLiveSchema :: Ptr ArrowSchema -> Assertion+assertLiveSchema ptr = do+    schema <- peek ptr+    assertBool "each batch needs an unreleased schema" (arrowSchemaRelease schema /= nullFunPtr)++-- | The DataFrame importer releases both objects after copying their contents.+assertConsumed :: Ptr ArrowSchema -> Ptr ArrowArray -> Assertion+assertConsumed schema array = do+    arrowSchemaRelease <$> peek schema >>= (@?= nullFunPtr)+    arrowArrayRelease <$> peek array >>= (@?= nullFunPtr)++-- | Keep native batch ordering deterministic in both execution modes.+withDb :: (Connection -> IO a) -> IO a+withDb = withConnectionWithConfig ":memory:" [("threads", "1")]

@@ -1,6 +1,6 @@ cabal-version: 3.4 name: duckdb-simple-version: 0.1.5.2+version: 0.2.0.0 license: MPL-2.0 license-file: LICENSE author: Matthias Pall Gissurarson@@ -8,22 +8,21 @@ category: Database build-type: Simple tested-with:-  ghc ==9.6.*-  ghc ==9.8.*-  ghc ==9.10.*-  ghc ==9.12.*-  ghc ==9.14.*+  ghc ==9.10.3+  ghc ==9.12.4+  ghc ==9.14.1+  ghc ==9.6.7+  ghc ==9.8.4 -extra-doc-files: README.md homepage: https://github.com/Tritlo/duckdb-haskell bug-reports: https://github.com/Tritlo/duckdb-haskell/issues-synopsis: Haskell FFI bindings for DuckDB+synopsis: High-level DuckDB interface inspired by sqlite-simple and postgresql-simple description:-  This library provides a mid-level interface for interacting with DuckDB,-  in the style of other "simple" libraries such as sqlite-simple and-  postgresql-simple.+  A high-level DuckDB interface with typed parameters and results,+  prepared statements, chunked folds, transactions, and Haskell scalar+  functions. The API follows the style of sqlite-simple and postgresql-simple.   .-  Tested with DuckDB version 1.5.0.+  Supports native DuckDB >= 1.5.3 and < 1.6. Tested with version 1.5.3.  extra-doc-files:   CHANGELOG.md@@ -34,12 +33,19 @@   location: https://github.com/Tritlo/duckdb-haskell.git   subdir: duckdb-simple +flag dataframe-tests+  description: Test Arrow export with the dataframe library.+  default: False+  manual: True+ library   exposed-modules:     Database.DuckDB.Simple+    Database.DuckDB.Simple.Arrow     Database.DuckDB.Simple.Catalog     Database.DuckDB.Simple.Config     Database.DuckDB.Simple.Copy+    Database.DuckDB.Simple.Deprecated.Streaming     Database.DuckDB.Simple.FileSystem     Database.DuckDB.Simple.FromField     Database.DuckDB.Simple.FromRow@@ -49,12 +55,16 @@     Database.DuckDB.Simple.Logging     Database.DuckDB.Simple.LogicalRep     Database.DuckDB.Simple.Ok+    Database.DuckDB.Simple.Time     Database.DuckDB.Simple.ToField     Database.DuckDB.Simple.ToRow     Database.DuckDB.Simple.Types    other-modules:+    Database.DuckDB.Simple.Arrow.Internal+    Database.DuckDB.Simple.Callback     Database.DuckDB.Simple.Materialize+    Database.DuckDB.Simple.Result    hs-source-dirs: src   default-language: Haskell2010@@ -63,7 +73,7 @@     base >=4.14 && <5,     bytestring >=0.11 && <0.13,     containers >=0.6 && <0.9,-    duckdb-ffi >=1.5.0.0 && <1.6,+    duckdb-ffi >=1.5.3.0 && <1.6,     text >=2.0 && <2.2,     time >=1.12 && <1.16,     transformers >=0.6 && <0.7,@@ -73,17 +83,30 @@   type: exitcode-stdio-1.0   hs-source-dirs: test   main-is: Spec.hs-  other-modules: Properties+  other-modules:+    ArrowTests+    CancellationTests+    CoreRegressionTests+    ExtensionRegressionTests+    Properties+    StreamingTests+    TimeTests+    ValueRegressionTests+   default-language: Haskell2010+  ghc-options:+    -threaded+    -rtsopts+   build-depends:-    QuickCheck >=2.14 && <2.18,     array >=0.5 && <0.6,     base >=4.14 && <5,     bytestring,     containers >=0.6 && <0.9,     directory >=1.3 && <1.4,-    duckdb-ffi >=1.5.0.0 && <1.6,+    duckdb-ffi >=1.5.3.0 && <1.6,     duckdb-simple,+    QuickCheck >=2.14 && <2.18,     tasty >=1.4 && <1.6,     tasty-expected-failure >=0.12 && <0.13,     tasty-hunit >=0.10 && <0.12,@@ -101,3 +124,38 @@   build-depends:     base >=4.14 && <5,     duckdb-simple,++test-suite duckdb-simple-dataframe-test+  type: exitcode-stdio-1.0+  hs-source-dirs: dataframe-test+  main-is: Main.hs+  default-language: Haskell2010+  ghc-options: -threaded++  if !flag(dataframe-tests)+    buildable: False+  build-depends:+    base >=4.14 && <5,+    dataframe-arrow-bridge >=1.0 && <1.1,+    dataframe-core >=2.5 && <2.6,+    duckdb-ffi >=1.5.3.0 && <1.6,+    duckdb-simple,+    tasty >=1.4 && <1.6,+    tasty-hunit >=0.10 && <0.12,+    text,++benchmark duckdb-simple-bench+  type: exitcode-stdio-1.0+  hs-source-dirs: bench+  main-is: Main.hs+  default-language: Haskell2010+  ghc-options:+    -O2+    -threaded+    -rtsopts++  build-depends:+    base >=4.14 && <5,+    duckdb-simple,+    text,+    time,

@@ -1,27 +1,39 @@ {-# LANGUAGE BlockArguments #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-deprecations #-}  {- | Regression check for leaked DuckDB handles.  The handles live in C memory, so the GHC heap statistics cannot see them.-Each leaked database instance keeps its own DuckDB thread pool alive, thus-this check compares the number of operating-system threads and the resident-set size of the process before and after many open\/close cycles.  Both-values are process-global, thus the check has its own test suite and does-not share a process with the other tests.+The open/close check counts native threads and resident memory. The sustained+checks keep one connection open through repeated queries, failures, cursor+resets, callback replacement, and cancellation. Weak references check callback+state release independently of resident memory. -The check reads @\/proc\/self\/status@.  On a system without @\/proc@ the-check reports a skip and exits with success.+Each check runs in this dedicated process. On systems without+@/proc/self/status@, the native resource counters are unavailable. The+functional checks and callback collection checks still run. -} module Main (main) where -import Control.Exception (IOException, evaluate, try)-import Control.Monad (forM_, when)+import Control.Concurrent (forkFinally, killThread, newEmptyMVar, putMVar, takeMVar, threadDelay, tryPutMVar)+import Control.Exception (AsyncException (ThreadKilled), IOException, SomeException, evaluate, fromException, try)+import Control.Monad (forM_, replicateM, unless, void, when)+import Data.IORef (atomicModifyIORef', atomicWriteIORef, mkWeakIORef, newIORef, readIORef) import Data.Int (Int64) import Data.List (isPrefixOf)-import Data.Maybe (listToMaybe)+import Data.Maybe (catMaybes, listToMaybe) import Database.DuckDB.Simple+import qualified Database.DuckDB.Simple.Copy as Copy+import qualified Database.DuckDB.Simple.Deprecated.Streaming as Streaming+import Database.DuckDB.Simple.FromField (FieldValue)+import qualified Database.DuckDB.Simple.Logging as Logging+import System.Environment (getArgs, lookupEnv) import System.Exit (exitFailure)+import System.Mem (performMajorGC)+import System.Mem.Weak (deRefWeak)+import Text.Read (readMaybe)  -- | The process-global resource counters that a leaked handle increases. data Usage = Usage@@ -69,10 +81,33 @@     stmt <- openStatement conn "SELECT 42"     closeStatement stmt     _ <- query_ conn "SELECT 42" :: IO [Only Int64]+    -- VARIANT decoding fails after DuckDB allocates the materialized result.+    -- The result and connection must still be destroyed.+    rejected <- try (query_ conn "SELECT i::VARIANT FROM range(100000) t(i)") :: IO (Either SomeException [Only FieldValue])+    case rejected of+        Left _ -> pure ()+        Right _ -> fail "expected unsupported VARIANT conversion"     close conn  main :: IO () main = do+    args <- getArgs+    let (mode, batches) = case args of+            [] -> ("all", 10)+            [name] -> (name, 10)+            [name, n] | Just count <- readMaybe n, count > 0 -> (name, count)+            _ -> ("invalid", 0)+    unless (mode `elem` ["all", "open-close", "long-lived", "callbacks", "cancel", "decode-failure"]) $+        fail "Expected [all|open-close|long-lived|callbacks|cancel|decode-failure] [positive batch count]"+    when (mode `elem` ["all", "open-close"]) checkOpenClose+    when (mode `elem` ["all", "long-lived"]) (checkLongLived batches)+    when (mode `elem` ["all", "callbacks"]) (checkCallbacks batches)+    when (mode `elem` ["all", "cancel"]) (checkCancellation batches)+    when (mode == "decode-failure") checkDecodeFailure++-- | Check native results and database handles across connection lifetimes.+checkOpenClose :: IO ()+checkOpenClose = do     -- The first cycle also does the one-time initialization, which must not     -- count as growth.     openCloseCycle@@ -84,12 +119,177 @@             after <- readUsage             case after of                 Nothing -> putStrLn "duckdb-simple leak check: /proc/self/status disappeared; skipped."-                Just final -> report baseline final+                Just final -> report ("open/close, " <> show cycles <> " cycles") baseline final +-- | Reuse one connection and statement through successful and failed operations.+checkLongLived :: Int -> IO ()+checkLongLived batches =+    withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+        withStatement conn "SELECT ?::BIGINT FROM range(10000)" \stmt -> do+            let cycleQuery = do+                    rows <- query_ conn "SELECT sum(i)::BIGINT FROM range(10000) t(i)"+                    unless (rows == [Only (49995000 :: Int64)]) (fail "wrong query result")+                    expectFailure (query_ conn "SELECT i::VARIANT FROM range(10000) t(i)" :: IO [Only FieldValue])+                    expectFailure (query_ conn "SELECT CAST('bad' AS BIGINT)" :: IO [Only Int64])+                    bind stmt [toField (42 :: Int64)]+                    nextRow stmt >>= \row -> unless (row == Just (Only (42 :: Int64))) (fail "wrong cursor result")+                    clearStatementBindings stmt+                    expectFailure (nextRow stmt :: IO (Maybe (Only Int64)))+                batch = forM_ [1 .. 100 :: Int] (const cycleQuery)+            batch+            performMajorGC+            before <- readUsage+            forM_ [1 .. batches] \n -> do+                batch+                performMajorGC+                after <- readUsage+                reportOptional ("open connection, " <> show (n * 100) <> " cycles") before after++-- | Isolate failed result cleanup for comparison with earlier package versions.+checkDecodeFailure :: IO ()+checkDecodeFailure =+    withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+        let rejected = expectFailure (query_ conn "SELECT i::VARIANT FROM range(100000) t(i)" :: IO [Only FieldValue])+        rejected+        performMajorGC+        before <- readUsage+        forM_ [1 .. cycles] (const rejected)+        performMajorGC+        after <- readUsage+        reportOptional ("open connection, " <> show cycles <> " decode failures") before after++-- | Check that callback closures and query state are released before close.+checkCallbacks :: Int -> IO ()+checkCallbacks batches =+    withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+        Logging.registerLogStorage conn "leak_log" (\_ -> pure ())+        copyStates <- newIORef []+        let newCopyState = do+                state <- newIORef (7 :: Int64)+                weak <- mkWeakIORef state (pure ())+                atomicModifyIORef' copyStates (\old -> (weak : old, ()))+                pure state+        Copy.registerCopyToFunction conn "leak_copy" (\_ -> newCopyState) (\_ -> newCopyState) (\_ _ -> pure ()) (\_ -> pure ())+        Copy.registerCopyToFunction conn "leak_copy_failure" (\_ -> newCopyState) (\_ -> fail "COPY init failure" :: IO ()) (\_ _ -> pure ()) (\_ -> pure ())+        let cycleCallback = do+                retained <- newIORef (42 :: Int64)+                weak <- mkWeakIORef retained (pure ())+                createFunction conn "leak_scalar" (readIORef retained)+                query_ conn "SELECT leak_scalar()" >>= \rows -> unless (rows == [Only (42 :: Int64)]) (fail "wrong callback result")+                states <- newIORef []+                createFunctionWithState+                    conn+                    "leak_state"+                    ( do+                        state <- newIORef (7 :: Int64)+                        stateWeak <- mkWeakIORef state (pure ())+                        -- The initializer returns each weak reference to the test.+                        atomicModifyIORef' states (\old -> (stateWeak : old, ()))+                        pure state+                    )+                    readIORef+                query_ conn "SELECT leak_state()" >>= \rows -> unless (rows == [Only (7 :: Int64)]) (fail "wrong state result")+                rejectedLog <- newIORef (0 :: Int64)+                logWeak <- mkWeakIORef rejectedLog (pure ())+                expectFailure (Logging.registerLogStorage conn "leak_log" (\_ -> void (readIORef rejectedLog)))+                void (execute_ conn "COPY (SELECT i FROM range(100) t(i)) TO 'unused' (FORMAT leak_copy)")+                expectFailure (execute_ conn "COPY (SELECT 1) TO 'unused' (FORMAT leak_copy_failure)")+                copyRefs <- atomicModifyIORef' copyStates (\refs -> ([], refs))+                stateRefs <- readIORef states+                pure (weak : logWeak : stateRefs <> copyRefs)+            batch = concat <$> replicateM 100 cycleCallback+            checkReleased refs = do+                -- Replace the last closure, then run a new transaction.+                createFunction conn "leak_scalar" (0 :: Int64)+                createFunction conn "leak_state" (0 :: Int64)+                void (query_ conn "SELECT 1" :: IO [Only Int64])+                performMajorGC+                performMajorGC+                live <- length . catMaybes <$> mapM deRefWeak refs+                unless (live == 0) (fail (show live <> " callback values remain live after replacement"))+        batch >>= checkReleased+        before <- readUsage+        forM_ [1 .. batches] \n -> do+            batch >>= checkReleased+            after <- readUsage+            reportOptional ("open connection, " <> show (n * 100) <> " callback cycles") before after++-- | Require an error without retaining the exception or its native resources.+expectFailure :: IO a -> IO ()+expectFailure action = do+    outcome <- try (void action) :: IO (Either SomeException ())+    case outcome of+        Left _ -> pure ()+        Right () -> fail "expected operation to fail"++-- | Reuse a connection after cancelling native execution and row decoding.+checkCancellation :: Int -> IO ()+checkCancellation batches =+    withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+        void (execute_ conn "SET streaming_buffer_size = '64KB'")+        let cancel action = do+                started <- newEmptyMVar+                done <- newEmptyMVar+                let signal = void (tryPutMVar started ())+                tid <- forkFinally (action signal) (\outcome -> putMVar done outcome >> signal)+                takeMVar started+                threadDelay 1000+                killThread tid+                outcome <- takeMVar done+                case outcome of+                    Left err | fromException err == Just ThreadKilled -> pure ()+                    Left err -> fail ("unexpected cancellation error: " <> show err)+                    Right () -> fail "query completed before cancellation"+                rows <- query_ conn "SELECT 42"+                unless (rows == [Only (42 :: Int64)]) (fail "connection failed after cancellation")+            batch = forM_ [1 .. 10 :: Int] \_ -> do+                cancel \signal -> do+                    createFunction conn "leak_cancel_started" (signal >> pure (1 :: Int64))+                    void+                        ( query_+                            conn+                            "WITH started AS MATERIALIZED (SELECT leak_cancel_started() AS seed) \+                            \SELECT sum(sin((a.i + b.j + started.seed)::DOUBLE)) \+                            \FROM started, range(1000000) a(i), range(1000000) b(j)" ::+                            IO [Only Double]+                        )+                cancel \signal -> do+                    signal+                    void (query_ conn "SELECT {'x': i, 'values': [i, i + 1]} FROM range(100000) t(i)" :: IO [Only FieldValue])+                cancel \signal -> do+                    blocked <- newEmptyMVar+                    -- Keep the first chunk live until the worker is cancelled.+                    void (fold_ conn "SELECT {'x': i, 'values': [i, NULL]} FROM range(100000) t(i)" (0 :: Int64) (\n (Only (_ :: FieldValue)) -> signal >> takeMVar blocked >> pure (n + 1)))+                forM_ [False, True] \arrow -> cancel \signal -> do+                    filtering <- newIORef False+                    createFunction conn "leak_stream_filter" \(_ :: Int64) -> do+                        active <- readIORef filtering+                        when active signal+                        pure active+                    let sql = "SELECT i FROM range(1000000000000) t(i) WHERE NOT leak_stream_filter(i)"+                        delivered = atomicWriteIORef filtering True+                    if arrow+                        then Streaming.foldArrow_ conn sql () (\() _ _ -> delivered)+                        else Streaming.fold_ conn sql () (\() (Only (_ :: Int64)) -> delivered)+        batch+        performMajorGC+        before <- readUsage+        forM_ [1 .. batches] \n -> do+            batch+            performMajorGC+            after <- readUsage+            reportOptional ("open connection, " <> show (n * 50) <> " cancellations") before after++-- | Report native counters when the operating system provides them.+reportOptional :: String -> Maybe Usage -> Maybe Usage -> IO ()+reportOptional label (Just before) (Just after) = report label before after+reportOptional label _ _ = putStrLn (label <> ": resource counters unavailable; functional checks passed")+ -- | Print both measurements and fail if either one grew too much.-report :: Usage -> Usage -> IO ()-report before after = do-    putStrLn $ "cycles: " <> show cycles+report :: String -> Usage -> Usage -> IO ()+report label before after = do+    checkRss <- (/= Just "0") <$> lookupEnv "DUCKDB_LEAK_RSS_CHECK"+    putStrLn label     putStrLn $         "threads: "             <> show (usageThreads before)@@ -106,8 +306,9 @@             <> " (slack "             <> show rssSlackKb             <> ")"-    when (threadGrowth > threadSlack || rssGrowth > rssSlackKb) do-        putStrLn "FAIL: open/close leaks DuckDB resources"+    unless checkRss (putStrLn "RSS assertion disabled for native memory instrumentation")+    when (threadGrowth > threadSlack || (checkRss && rssGrowth > rssSlackKb)) do+        putStrLn ("FAIL: " <> label <> " retains DuckDB resources")         exitFailure   where     threadGrowth = usageThreads after - usageThreads before

@@ -78,10 +78,11 @@     deleteFunction, ) where -import Control.Exception (SomeException, bracket, finally, mask, onException, throwIO, try)-import Control.Monad (forM, forM_, join, void, when, zipWithM, zipWithM_)-import Data.IORef (IORef, atomicModifyIORef', mkWeakIORef, newIORef, readIORef, writeIORef)+import Control.Exception (SomeException, bracket, finally, mask, mask_, onException, throwIO, try)+import Control.Monad (forM, forM_, join, void, when, zipWithM_)+import Data.IORef (atomicModifyIORef', mkWeakIORef, newIORef) import Data.Maybe (isJust, isNothing)+import qualified Data.Set as Set import Data.Text (Text) import qualified Data.Text as Text import qualified Data.Text.Foreign as TextForeign@@ -106,27 +107,27 @@     Connection (..),     ConnectionState (..),     Query (..),+    ResultMode (..),     SQLError (..),     Statement (..),     StatementState (..),-    StatementStream (..),-    StatementStreamChunk (..),-    StatementStreamChunkVector (..),-    StatementStreamColumn (..),     StatementStreamState (..),+    keepAlive,+    peekUtf8CString,+    runInterruptibleQuery,     withConnectionHandle,     withQueryCString,+    withResult,     withStatementHandle,  )-import Database.DuckDB.Simple.Materialize (-    materializeValue,- ) import Database.DuckDB.Simple.Ok (Ok (..))+import Database.DuckDB.Simple.Result (cleanupStatementStreamRef, collectRows, resetStatementStream)+import qualified Database.DuckDB.Simple.Result as Result import Database.DuckDB.Simple.ToField (DuckDBColumnType (..), FieldBinding, NamedParam (..), ToField (..), bindFieldBinding, duckdbColumnType, renderFieldBinding) import Database.DuckDB.Simple.ToRow (ToRow (..)) import Database.DuckDB.Simple.Types (FormatError (..), Null (..), Only (..), (:.) (..))-import Foreign.C.String (CString, peekCString, withCString)-import Foreign.Marshal.Alloc (alloca, free, malloc)+import Foreign.C.String (CString)+import Foreign.Marshal.Alloc (alloca) import Foreign.Ptr (Ptr, castPtr, nullPtr) import Foreign.Storable (peek, poke) @@ -137,10 +138,10 @@ -- | Open a DuckDB database with configuration flags applied before startup. openWithConfig :: FilePath -> [(Text, Text)] -> IO Connection openWithConfig path settings =-    mask \restore -> do-        db <- restore (openDatabaseWithConfig path settings)+    mask_ do+        db <- openDatabaseWithConfig path settings         conn <--            restore (connectDatabase db)+            connectDatabase db                 `onException` closeDatabaseHandle db         createConnection db conn             `onException` do@@ -150,11 +151,12 @@ -- | Close a connection.  The operation is idempotent. close :: Connection -> IO () close Connection{connectionState} =-    join $-        atomicModifyIORef' connectionState \case-            ConnectionClosed -> (ConnectionClosed, pure ())-            openState@(ConnectionOpen{}) ->-                (ConnectionClosed, closeHandles openState)+    mask_ $+        join $+            atomicModifyIORef' connectionState \case+                ConnectionClosed -> (ConnectionClosed, pure ())+                openState@(ConnectionOpen{}) ->+                    (ConnectionClosed, closeHandles openState)  -- | Run an action with a freshly opened connection, closing it afterwards. withConnection :: FilePath -> (Connection -> IO a) -> IO a@@ -167,26 +169,26 @@ -- | Prepare a SQL statement for execution. openStatement :: Connection -> Query -> IO Statement openStatement conn queryText =-    mask \restore -> do+    mask_ do         handle <--            restore $-                withConnectionHandle conn \connPtr ->-                    withQueryCString queryText \sql ->-                        alloca \stmtPtr -> do-                            rc <- c_duckdb_prepare connPtr sql stmtPtr+            withConnectionHandle conn \connPtr ->+                withQueryCString queryText \sql ->+                    alloca \stmtPtr -> do+                        poke stmtPtr nullPtr+                        flip onException (c_duckdb_destroy_prepare stmtPtr) do+                            rc <- runInterruptibleQuery conn (c_duckdb_prepare connPtr sql stmtPtr)                             stmt <- peek stmtPtr                             if rc == DuckDBSuccess                                 then pure stmt                                 else do                                     errMsg <- fetchPrepareError stmt-                                    c_duckdb_destroy_prepare stmtPtr                                     throwIO $ mkPrepareError queryText errMsg         createStatement conn handle queryText             `onException` destroyPrepared handle  -- | Finalise a prepared statement.  The operation is idempotent. closeStatement :: Statement -> IO ()-closeStatement stmt@Statement{statementState} = do+closeStatement stmt@Statement{statementState} = mask_ do     resetStatementStream stmt     finish <- atomicModifyIORef' statementState \case         StatementClosed -> (StatementClosed, pure ())@@ -248,6 +250,9 @@                 Just idx -> bindFieldBinding stmt (fromIntegral idx :: DuckDBIdx) binding      in do             resetStatementStream stmt+            let names = map (normalizeName . fst) bindings+            when (Set.size (Set.fromList names) /= length names) $+                throwFormatErrorNamed stmt (Text.pack "duckdb-simple: duplicate named parameter") bindings             withStatementHandle stmt \handle -> do                 let actual = length bindings                 expected <- fmap fromIntegral (c_duckdb_nparams handle)@@ -261,22 +266,22 @@  fetchParameterNames :: DuckDBPreparedStatement -> Int -> IO [Maybe Text] fetchParameterNames handle count =-    forM [1 .. count] \idx -> do-        namePtr <- c_duckdb_parameter_name handle (fromIntegral idx)-        if namePtr == nullPtr-            then pure Nothing-            else do-                name <- Text.pack <$> peekCString namePtr-                c_duckdb_free (castPtr namePtr)-                let normalized = normalizeName name-                if normalized == Text.pack (show idx)-                    then pure Nothing-                    else pure (Just name)+    forM [1 .. count] \idx ->+        bracket (c_duckdb_parameter_name handle (fromIntegral idx)) (c_duckdb_free . castPtr) \namePtr ->+            if namePtr == nullPtr+                then pure Nothing+                else do+                    name <- peekUtf8CString namePtr+                    let normalized = normalizeName name+                    if normalized == Text.pack (show idx)+                        then pure Nothing+                        else pure (Just name)  -- | Remove all parameter bindings associated with a prepared statement. clearStatementBindings :: Statement -> IO () clearStatementBindings stmt =     withStatementHandle stmt \handle -> do+        resetStatementStream stmt         rc <- c_duckdb_clear_bindings handle         when (rc /= DuckDBSuccess) $ do             err <- fetchPrepareError handle@@ -285,18 +290,20 @@ -- | Look up the 1-based index of a named placeholder. namedParameterIndex :: Statement -> Text -> IO (Maybe Int) namedParameterIndex stmt name =-    withStatementHandle stmt \handle ->+    withStatementHandle stmt \handle -> do+        when (Text.any (== '\0') name) $+            throwIO (mkPrepareError (statementQuery stmt) (Text.pack "duckdb-simple: parameter name contains NUL"))         let normalized = normalizeName name-         in TextForeign.withCString normalized \cName ->-                alloca \idxPtr -> do-                    rc <- c_duckdb_bind_parameter_index handle idxPtr cName-                    if rc == DuckDBSuccess-                        then do-                            idx <- peek idxPtr-                            if idx == 0-                                then pure Nothing-                                else pure (Just (fromIntegral idx))-                        else pure Nothing+        TextForeign.withCString normalized \cName ->+            alloca \idxPtr -> do+                rc <- c_duckdb_bind_parameter_index handle idxPtr cName+                if rc == DuckDBSuccess+                    then do+                        idx <- peek idxPtr+                        if idx == 0+                            then pure Nothing+                            else pure (Just (fromIntegral idx))+                    else pure Nothing  -- | Retrieve the number of columns produced by the supplied prepared statement. columnCount :: Statement -> IO Int@@ -313,13 +320,10 @@             total <- fmap fromIntegral (c_duckdb_prepared_statement_column_count handle)             when (columnIndex >= total) $                 throwIO (columnIndexError stmt columnIndex (Just total))-            namePtr <- c_duckdb_prepared_statement_column_name handle (fromIntegral columnIndex)-            if namePtr == nullPtr-                then throwIO (columnNameUnavailableError stmt columnIndex)-                else do-                    name <- Text.pack <$> peekCString namePtr-                    c_duckdb_free (castPtr namePtr)-                    pure name+            bracket (c_duckdb_prepared_statement_column_name handle (fromIntegral columnIndex)) (c_duckdb_free . castPtr) \namePtr ->+                if namePtr == nullPtr+                    then throwIO (columnNameUnavailableError stmt columnIndex)+                    else peekUtf8CString namePtr  {- | Execute a prepared statement and return the number of affected rows.   Resets any active result stream before running and raises an @SQLError@@@ -329,17 +333,7 @@ executeStatement stmt =     withStatementHandle stmt \handle -> do         resetStatementStream stmt-        alloca \resPtr -> do-            rc <- c_duckdb_execute_prepared handle resPtr-            if rc == DuckDBSuccess-                then do-                    changed <- resultRowsChanged resPtr-                    c_duckdb_destroy_result resPtr-                    pure changed-                else do-                    (errMsg, _) <- fetchResultError resPtr-                    c_duckdb_destroy_result resPtr-                    throwIO $ mkPrepareError (statementQuery stmt) errMsg+        withResult (statementConnection stmt) (statementQuery stmt) (c_duckdb_execute_prepared handle) resultRowsChanged  -- | Execute a query with positional parameters and return the affected row count. execute :: (ToRow q) => Connection -> Query -> q -> IO Int@@ -359,17 +353,7 @@ execute_ conn queryText =     withConnectionHandle conn \connPtr ->         withQueryCString queryText \sql ->-            alloca \resPtr -> do-                rc <- c_duckdb_query connPtr sql resPtr-                if rc == DuckDBSuccess-                    then do-                        changed <- resultRowsChanged resPtr-                        c_duckdb_destroy_result resPtr-                        pure changed-                    else do-                        (errMsg, errType) <- fetchResultError resPtr-                        c_duckdb_destroy_result resPtr-                        throwIO $ mkExecuteError queryText errMsg errType+            withResult conn queryText (c_duckdb_query connPtr sql) resultRowsChanged  -- | Execute a query that uses named parameters. executeNamed :: Connection -> Query -> [NamedParam] -> IO Int@@ -388,17 +372,8 @@     withStatement conn queryText \stmt -> do         bind stmt (toRow params)         withStatementHandle stmt \handle ->-            alloca \resPtr -> do-                rc <- c_duckdb_execute_prepared handle resPtr-                if rc == DuckDBSuccess-                    then do-                        rows <- collectRows resPtr-                        c_duckdb_destroy_result resPtr-                        convertRowsWith parser queryText rows-                    else do-                        (errMsg, errType) <- fetchResultError resPtr-                        c_duckdb_destroy_result resPtr-                        throwIO $ mkExecuteError queryText errMsg errType+            withResult conn queryText (c_duckdb_execute_prepared handle) \resPtr ->+                collectRows queryText resPtr >>= convertRowsWith parser queryText  -- | Run a query that uses named parameters and decode all rows eagerly. queryNamed :: (FromRow r) => Connection -> Query -> [NamedParam] -> IO [r]@@ -406,17 +381,8 @@     withStatement conn queryText \stmt -> do         bindNamed stmt params         withStatementHandle stmt \handle ->-            alloca \resPtr -> do-                rc <- c_duckdb_execute_prepared handle resPtr-                if rc == DuckDBSuccess-                    then do-                        rows <- collectRows resPtr-                        c_duckdb_destroy_result resPtr-                        convertRows queryText rows-                    else do-                        (errMsg, errType) <- fetchResultError resPtr-                        c_duckdb_destroy_result resPtr-                        throwIO $ mkExecuteError queryText errMsg errType+            withResult conn queryText (c_duckdb_execute_prepared handle) \resPtr ->+                collectRows queryText resPtr >>= convertRows queryText  -- | Run a query without supplying parameters and decode all rows eagerly. query_ :: (FromRow r) => Connection -> Query -> IO [r]@@ -427,282 +393,53 @@ queryWith_ parser conn queryText =     withConnectionHandle conn \connPtr ->         withQueryCString queryText \sql ->-            alloca \resPtr -> do-                rc <- c_duckdb_query connPtr sql resPtr-                if rc == DuckDBSuccess-                    then do-                        rows <- collectRows resPtr-                        c_duckdb_destroy_result resPtr-                        convertRowsWith parser queryText rows-                    else do-                        (errMsg, errType) <- fetchResultError resPtr-                        c_duckdb_destroy_result resPtr-                        throwIO $ mkExecuteError queryText errMsg errType+            withResult conn queryText (c_duckdb_query connPtr sql) \resPtr ->+                collectRows queryText resPtr >>= convertRowsWith parser queryText --- Streaming folds -----------------------------------------------------------+-- Cursors and folds --------------------------------------------------------- -{- | Stream a parameterised query through an accumulator without loading all rows.-  Bind the supplied parameters, start a streaming result, and apply the step-  function row by row to produce a final accumulator value.+{- | Fold a parameterised result without constructing a complete Haskell row list.+  DuckDB materializes the native result before decoding starts. Native memory+  use depends on the result size; the step function controls Haskell memory use. -} fold :: (FromRow row, ToRow params) => Connection -> Query -> params -> a -> (a -> row -> IO a) -> IO a fold conn queryText params initial step =     withStatement conn queryText \stmt -> do         resetStatementStream stmt         bind stmt (toRow params)-        foldStatementWith fromRow stmt initial step+        Result.foldStatementWith MaterializedResult fromRow stmt initial step --- | Stream a parameterless query through an accumulator without loading all rows.+-- | Fold a parameterless result. Native materialization follows 'fold'. fold_ :: (FromRow row) => Connection -> Query -> a -> (a -> row -> IO a) -> IO a fold_ conn queryText initial step =     withStatement conn queryText \stmt -> do         resetStatementStream stmt-        foldStatementWith fromRow stmt initial step+        Result.foldStatementWith MaterializedResult fromRow stmt initial step --- | Stream a query that uses named parameters through an accumulator.+-- | Fold a result with named parameters. Native materialization follows 'fold'. foldNamed :: (FromRow row) => Connection -> Query -> [NamedParam] -> a -> (a -> row -> IO a) -> IO a foldNamed conn queryText params initial step =     withStatement conn queryText \stmt -> do         resetStatementStream stmt         bindNamed stmt params-        foldStatementWith fromRow stmt initial step--foldStatementWith :: RowParser row -> Statement -> a -> (a -> row -> IO a) -> IO a-foldStatementWith parser stmt initial step =-    let loop acc = do-            nextVal <- nextRowWith parser stmt-            case nextVal of-                Nothing -> pure acc-                Just row -> do-                    acc' <- step acc row-                    acc' `seq` loop acc'-     in loop initial `finally` resetStatementStream stmt+        Result.foldStatementWith MaterializedResult fromRow stmt initial step --- | Fetch the next row from a streaming statement, stopping when no rows remain.+-- | Fetch the next row. The first call materializes the native result. nextRow :: (FromRow r) => Statement -> IO (Maybe r) nextRow = nextRowWith fromRow  -- | Fetch the next row using a custom parser, returning @Nothing@ once exhausted. nextRowWith :: RowParser r -> Statement -> IO (Maybe r)-nextRowWith parser stmt@Statement{statementStream} =-    mask \restore -> do-        state <- readIORef statementStream-        case state of-            StatementStreamIdle -> do-                newStream <- restore (startStatementStream stmt)-                case newStream of-                    Nothing -> pure Nothing-                    Just stream -> restore (consumeStream statementStream parser stmt stream)-            StatementStreamActive stream ->-                restore (consumeStream statementStream parser stmt stream)--resetStatementStream :: Statement -> IO ()-resetStatementStream Statement{statementStream} =-    cleanupStatementStreamRef statementStream--consumeStream :: IORef StatementStreamState -> RowParser r -> Statement -> StatementStream -> IO (Maybe r)-consumeStream streamRef parser stmt stream = do-    result <--        ( try (streamNextRow (statementQuery stmt) stream) ::-            IO (Either SomeException (Maybe [Field], StatementStream))-        )-    case result of-        Left err -> do-            finalizeStream stream-            writeIORef streamRef StatementStreamIdle-            throwIO err-        Right (maybeFields, updatedStream) ->-            case maybeFields of-                Nothing -> do-                    finalizeStream updatedStream-                    writeIORef streamRef StatementStreamIdle-                    pure Nothing-                Just fields ->-                    case parseRow parser fields of-                        Errors rowErr -> do-                            finalizeStream updatedStream-                            writeIORef streamRef StatementStreamIdle-                            throwIO $ rowErrorsToSqlError (statementQuery stmt) rowErr-                        Ok value -> do-                            writeIORef streamRef (StatementStreamActive updatedStream)-                            pure (Just value)--startStatementStream :: Statement -> IO (Maybe StatementStream)-startStatementStream stmt =-    withStatementHandle stmt \handle -> do-        columns <- collectStreamColumns handle-        resultPtr <- malloc-        rc <- c_duckdb_execute_prepared handle resultPtr-        if rc /= DuckDBSuccess-            then do-                (errMsg, errType) <- fetchResultError resultPtr-                c_duckdb_destroy_result resultPtr-                free resultPtr-                throwIO $ mkExecuteError (statementQuery stmt) errMsg errType-            else do-                resultType <- c_duckdb_result_return_type resultPtr-                if resultType /= DuckDBResultTypeQueryResult-                    then do-                        c_duckdb_destroy_result resultPtr-                        free resultPtr-                        pure Nothing-                    else pure (Just (StatementStream resultPtr columns Nothing))--collectStreamColumns :: DuckDBPreparedStatement -> IO [StatementStreamColumn]-collectStreamColumns handle = do-    rawCount <- c_duckdb_prepared_statement_column_count handle-    let cc = fromIntegral rawCount :: Int-    forM [0 .. cc - 1] \idx -> do-        namePtr <- c_duckdb_prepared_statement_column_name handle (fromIntegral idx)-        name <--            if namePtr == nullPtr-                then pure (Text.pack ("column" <> show idx))-                else Text.pack <$> peekCString namePtr-        dtype <- c_duckdb_prepared_statement_column_type handle (fromIntegral idx)-        pure-            StatementStreamColumn-                { statementStreamColumnIndex = idx-                , statementStreamColumnName = name-                , statementStreamColumnType = dtype-                }--streamNextRow :: Query -> StatementStream -> IO (Maybe [Field], StatementStream)-streamNextRow queryText stream@StatementStream{statementStreamChunk = Nothing} = do-    refreshed <- fetchChunk stream-    case statementStreamChunk refreshed of-        Nothing -> pure (Nothing, refreshed)-        Just chunk -> emitRow queryText refreshed chunk-streamNextRow queryText stream@StatementStream{statementStreamChunk = Just chunk} =-    emitRow queryText stream chunk--fetchChunk :: StatementStream -> IO StatementStream-fetchChunk stream@StatementStream{statementStreamResult} = do-    chunk <- c_duckdb_fetch_chunk statementStreamResult-    if chunk == nullPtr-        then pure stream-        else do-            rawSize <- c_duckdb_data_chunk_get_size chunk-            let rowCount = fromIntegral rawSize :: Int-            if rowCount <= 0-                then do-                    destroyDataChunk chunk-                    fetchChunk stream-                else do-                    vectors <- prepareChunkVectors chunk (statementStreamColumns stream)-                    let chunkState =-                            StatementStreamChunk-                                { statementStreamChunkPtr = chunk-                                , statementStreamChunkSize = rowCount-                                , statementStreamChunkIndex = 0-                                , statementStreamChunkVectors = vectors-                                }-                    pure stream{statementStreamChunk = Just chunkState}--prepareChunkVectors :: DuckDBDataChunk -> [StatementStreamColumn] -> IO [StatementStreamChunkVector]-prepareChunkVectors chunk columns =-    forM columns \StatementStreamColumn{statementStreamColumnIndex} -> do-        vector <- c_duckdb_data_chunk_get_vector chunk (fromIntegral statementStreamColumnIndex)-        dataPtr <- c_duckdb_vector_get_data vector-        validity <- c_duckdb_vector_get_validity vector-        pure-            StatementStreamChunkVector-                { statementStreamChunkVectorHandle = vector-                , statementStreamChunkVectorData = dataPtr-                , statementStreamChunkVectorValidity = validity-                }--emitRow :: Query -> StatementStream -> StatementStreamChunk -> IO (Maybe [Field], StatementStream)-emitRow queryText stream chunk@StatementStreamChunk{statementStreamChunkIndex, statementStreamChunkSize} = do-    fields <--        buildRow-            queryText-            (statementStreamColumns stream)-            (statementStreamChunkVectors chunk)-            statementStreamChunkIndex-    let nextIndex = statementStreamChunkIndex + 1-    if nextIndex < statementStreamChunkSize-        then-            let updatedChunk = chunk{statementStreamChunkIndex = nextIndex}-             in pure (Just fields, stream{statementStreamChunk = Just updatedChunk})-        else do-            destroyDataChunk (statementStreamChunkPtr chunk)-            pure (Just fields, stream{statementStreamChunk = Nothing})--buildRow :: Query -> [StatementStreamColumn] -> [StatementStreamChunkVector] -> Int -> IO [Field]-buildRow queryText columns vectors rowIdx =-    zipWithM (buildField queryText rowIdx) columns vectors--buildField :: Query -> Int -> StatementStreamColumn -> StatementStreamChunkVector -> IO Field-buildField queryText rowIdx column StatementStreamChunkVector{statementStreamChunkVectorHandle, statementStreamChunkVectorData, statementStreamChunkVectorValidity} = do-    let dtype = statementStreamColumnType column-    value <--        case dtype of-            DuckDBTypeStruct ->-                throwIO (streamingUnsupportedTypeError queryText column)-            DuckDBTypeUnion ->-                throwIO (streamingUnsupportedTypeError queryText column)-            _ ->-                materializeValue-                    dtype-                    statementStreamChunkVectorHandle-                    statementStreamChunkVectorData-                    statementStreamChunkVectorValidity-                    rowIdx-    pure-        Field-            { fieldName = statementStreamColumnName column-            , fieldIndex = statementStreamColumnIndex column-            , fieldValue = value-            }--cleanupStatementStreamRef :: IORef StatementStreamState -> IO ()-cleanupStatementStreamRef ref = do-    state <- atomicModifyIORef' ref (StatementStreamIdle,)-    finalizeStreamState state--finalizeStreamState :: StatementStreamState -> IO ()-finalizeStreamState = \case-    StatementStreamIdle -> pure ()-    StatementStreamActive stream -> finalizeStream stream--finalizeStream :: StatementStream -> IO ()-finalizeStream StatementStream{statementStreamResult, statementStreamChunk} = do-    maybe (pure ()) finalizeChunk statementStreamChunk-    c_duckdb_destroy_result statementStreamResult-    free statementStreamResult--finalizeChunk :: StatementStreamChunk -> IO ()-finalizeChunk StatementStreamChunk{statementStreamChunkPtr} =-    destroyDataChunk statementStreamChunkPtr--destroyDataChunk :: DuckDBDataChunk -> IO ()-destroyDataChunk chunk =-    alloca \ptr -> do-        poke ptr chunk-        c_duckdb_destroy_data_chunk ptr--streamingUnsupportedTypeError :: Query -> StatementStreamColumn -> SQLError-streamingUnsupportedTypeError queryText StatementStreamColumn{statementStreamColumnName, statementStreamColumnType} =-    SQLError-        { sqlErrorMessage =-            Text.concat-                [ Text.pack "duckdb-simple: streaming does not yet support column "-                , statementStreamColumnName-                , Text.pack " with DuckDB type "-                , Text.pack (show statementStreamColumnType)-                ]-        , sqlErrorType = Nothing-        , sqlErrorQuery = Just queryText-        }+nextRowWith = Result.nextRowWith MaterializedResult  -- | Run an action inside a transaction. withTransaction :: Connection -> IO a -> IO a withTransaction conn action =     mask \restore -> do         void (execute_ conn begin)-        let rollbackAction = void (execute_ conn rollback)+        let rollbackAction = void (try (execute_ conn rollback) :: IO (Either SomeException Int))         result <- restore action `onException` rollbackAction-        void (execute_ conn commit)+        void (execute_ conn commit) `onException` rollbackAction         pure result   where     begin = Query (Text.pack "BEGIN TRANSACTION")@@ -729,7 +466,7 @@     streamRef <- newIORef StatementStreamIdle     _ <-         mkWeakIORef ref $-            do+            keepAlive parent do                 join $                     atomicModifyIORef' ref $ \case                         StatementClosed -> (StatementClosed, pure ())@@ -748,7 +485,9 @@             }  openDatabaseWithConfig :: FilePath -> [(Text, Text)] -> IO DuckDBDatabase-openDatabaseWithConfig path settings =+openDatabaseWithConfig path settings = do+    when ('\0' `elem` path || any (\(name, value) -> Text.any (== '\0') name || Text.any (== '\0') value) settings) $+        throwIO (mkOpenError (Text.pack "duckdb-simple: database path or configuration contains NUL"))     alloca \dbPtr ->         alloca \configPtr ->             alloca \errPtr -> do@@ -766,14 +505,14 @@                             TextForeign.withCString name \cName ->                                 TextForeign.withCString value \cValue -> do                                     rcSet <- c_duckdb_set_config config cName cValue-                                    when (rcSet /= DuckDBSuccess) $-                                        throwIO $-                                            mkOpenError $-                                                Text.concat-                                                    [ Text.pack "duckdb-simple: failed to set config option "-                                                    , name-                                                    ]-                        withCString path \cPath -> do+                                    when (rcSet /= DuckDBSuccess)+                                        $ throwIO+                                        $ mkOpenError+                                        $ Text.concat+                                            [ Text.pack "duckdb-simple: failed to set config option "+                                            , name+                                            ]+                        TextForeign.withCString (Text.pack path) \cPath -> do                             rc <- c_duckdb_open_ext cPath dbPtr config errPtr                             if rc == DuckDBSuccess                                 then do@@ -816,21 +555,7 @@     msgPtr <- c_duckdb_prepare_error stmt     if msgPtr == nullPtr         then pure (Text.pack "duckdb-simple: prepare failed")-        else Text.pack <$> peekCString msgPtr--fetchResultError :: Ptr DuckDBResult -> IO (Text, Maybe DuckDBErrorType)-fetchResultError resultPtr = do-    msgPtr <- c_duckdb_result_error resultPtr-    msg <--        if msgPtr == nullPtr-            then pure (Text.pack "duckdb-simple: query failed")-            else Text.pack <$> peekCString msgPtr-    errType <- c_duckdb_result_error_type resultPtr-    let classified =-            if errType == DuckDBErrorInvalid-                then Nothing-                else Just errType-    pure (msg, classified)+        else peekUtf8CString msgPtr  mkOpenError :: Text -> SQLError mkOpenError msg =@@ -856,14 +581,6 @@         , sqlErrorQuery = Just queryText         } -mkExecuteError :: Query -> Text -> Maybe DuckDBErrorType -> SQLError-mkExecuteError queryText msg errType =-    SQLError-        { sqlErrorMessage = msg-        , sqlErrorType = errType-        , sqlErrorQuery = Just queryText-        }- throwFormatError :: Statement -> Text -> [String] -> IO a throwFormatError Statement{statementQuery} message params =     throwIO@@ -939,82 +656,13 @@         Errors err -> throwIO (rowErrorsToSqlError queryText err)         Ok ok -> pure ok -collectRows :: Ptr DuckDBResult -> IO [[Field]]-collectRows resPtr = do-    columns <- collectResultColumns resPtr-    collectChunks columns []-  where-    collectChunks columns acc = do-        chunk <- c_duckdb_fetch_chunk resPtr-        if chunk == nullPtr-            then pure (concat (reverse acc))-            else do-                rows <--                    finally-                        (decodeChunk columns chunk)-                        (destroyDataChunk chunk)-                let acc' = maybe acc (: acc) rows-                collectChunks columns acc'--    decodeChunk columns chunk = do-        rawSize <- c_duckdb_data_chunk_get_size chunk-        let rowCount = fromIntegral rawSize :: Int-        if rowCount <= 0-            then pure Nothing-            else-                if null columns-                    then pure (Just (replicate rowCount []))-                    else do-                        vectors <- prepareChunkVectors chunk columns-                        rows <- mapM (buildMaterializedRow columns vectors) [0 .. rowCount - 1]-                        pure (Just rows)--collectResultColumns :: Ptr DuckDBResult -> IO [StatementStreamColumn]-collectResultColumns resPtr = do-    rawCount <- c_duckdb_column_count resPtr-    let cc = fromIntegral rawCount :: Int-    forM [0 .. cc - 1] \columnIndex -> do-        namePtr <- c_duckdb_column_name resPtr (fromIntegral columnIndex)-        name <--            if namePtr == nullPtr-                then pure (Text.pack ("column" <> show columnIndex))-                else Text.pack <$> peekCString namePtr-        dtype <- c_duckdb_column_type resPtr (fromIntegral columnIndex)-        pure-            StatementStreamColumn-                { statementStreamColumnIndex = columnIndex-                , statementStreamColumnName = name-                , statementStreamColumnType = dtype-                }--buildMaterializedRow :: [StatementStreamColumn] -> [StatementStreamChunkVector] -> Int -> IO [Field]-buildMaterializedRow columns vectors rowIdx =-    zipWithM (buildMaterializedField rowIdx) columns vectors--buildMaterializedField :: Int -> StatementStreamColumn -> StatementStreamChunkVector -> IO Field-buildMaterializedField rowIdx column StatementStreamChunkVector{statementStreamChunkVectorHandle, statementStreamChunkVectorData, statementStreamChunkVectorValidity} = do-    value <--        materializeValue-            (statementStreamColumnType column)-            statementStreamChunkVectorHandle-            statementStreamChunkVectorData-            statementStreamChunkVectorValidity-            rowIdx-    pure-        Field-            { fieldName = statementStreamColumnName column-            , fieldIndex = statementStreamColumnIndex column-            , fieldValue = value-            }- peekError :: Ptr CString -> IO Text peekError ptr = do     errPtr <- peek ptr     if errPtr == nullPtr         then pure (Text.pack "duckdb-simple: failed to open database")         else do-            message <- peekCString errPtr-            pure (Text.pack message)+            peekUtf8CString errPtr  maybeFreeErr :: Ptr CString -> IO () maybeFreeErr ptr = do

@@ -0,0 +1,51 @@+{- |+Module      : Database.DuckDB.Simple.Arrow+Description : Scoped Arrow export through the DuckDB C Data Interface.++These functions execute a query and visit its Arrow batches. DuckDB materializes+the native result before the first callback. The Haskell code converts and+releases one batch at a time. Native result memory can grow with the query size.++Each callback receives a separate schema and array. It may read them or pass+them to a consumer that releases or moves them under the Arrow C Data Interface.+The fold releases objects that the consumer has not released or moved, including+when the callback throws. The consumer must release any contents it moves.++The pointers themselves are valid during the callback only. To retain contents,+the consumer must move the root structs into its own storage and set the source+release fields to NULL. Do not retain the original pointers or move children+separately. Moved contents remain valid after the query and connection close.+Haskell consumers must mask asynchronous exceptions while moving a root and+register its cleanup before restoring exceptions.++This module uses the schema and chunk conversion API. The older query and scan+functions in @Database.DuckDB.FFI.Deprecated@ are deprecated by DuckDB.+-}+module Database.DuckDB.Simple.Arrow (+    foldArrow,+    foldArrow_,+) where++import Database.DuckDB.FFI (ArrowArray, ArrowSchema)+import Database.DuckDB.Simple.Arrow.Internal (foldArrowWith)+import Database.DuckDB.Simple.Internal (Connection, Query, ResultMode (MaterializedResult))+import Database.DuckDB.Simple.ToRow (ToRow)+import Foreign.Ptr (Ptr)++{- | Execute a parameterized query and fold over Arrow batches.++The schema includes the executed result's column names and types. The callback+runs once per nonempty batch and never runs for an empty result. The schema and+array pointers are valid only during the callback. A consumer may release or+move their contents as described in the module documentation. The accumulator+is evaluated to weak head normal form after each callback.++DuckDB materializes the native result before this fold starts. This function+does not provide bounded-memory query execution.+-}+foldArrow :: (ToRow q) => Connection -> Query -> q -> a -> (a -> Ptr ArrowSchema -> Ptr ArrowArray -> IO a) -> IO a+foldArrow = foldArrowWith MaterializedResult++-- | Fold over Arrow batches from a query without parameters.+foldArrow_ :: Connection -> Query -> a -> (a -> Ptr ArrowSchema -> Ptr ArrowArray -> IO a) -> IO a+foldArrow_ conn queryText = foldArrow conn queryText ()

@@ -0,0 +1,91 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE OverloadedStrings #-}++{- |+Module      : Database.DuckDB.Simple.Arrow.Internal+Description : Shared ownership and conversion for Arrow result batches.+-}+module Database.DuckDB.Simple.Arrow.Internal (+    foldArrowWith,+) where++import Control.Exception (bracket, bracket_, throwIO)+import Control.Monad (forM, when)+import Database.DuckDB.FFI+import Database.DuckDB.Simple (bind, withStatement)+import Database.DuckDB.Simple.Internal (Connection, Query, ResultMode, SQLError (..), destroyDataChunk, destroyLogicalType, executePreparedResult, fetchResultChunk, peekUtf8CString, throwResultError, withResult, withStatementHandle)+import Database.DuckDB.Simple.ToRow (ToRow (..))+import Foreign.Marshal.Alloc (alloca)+import Foreign.Marshal.Array (withArray)+import Foreign.Marshal.Utils (fillBytes, withMany)+import Foreign.Ptr (Ptr, nullPtr)+import Foreign.Storable (Storable (sizeOf), poke)++-- | Fold over Arrow batches with the selected native execution mode.+foldArrowWith :: (ToRow q) => ResultMode -> Connection -> Query -> q -> a -> (a -> Ptr ArrowSchema -> Ptr ArrowArray -> IO a) -> IO a+foldArrowWith mode conn queryText params initial step =+    withStatement conn queryText \stmt -> do+        bind stmt (toRow params)+        withStatementHandle stmt \handle ->+            withResult conn queryText (executePreparedResult mode handle) \result ->+                bracket (c_duckdb_result_get_arrow_options result) destroyArrowOptions \options ->+                    loop result options initial+  where+    loop result options acc = do+        next <- bracket (fetchResultChunk mode conn result) destroyDataChunk \chunk ->+            if chunk == nullPtr+                then do+                    throwResultError queryText result+                    pure Nothing+                else do+                    rowCount <- c_duckdb_data_chunk_get_size chunk+                    if rowCount == 0+                        then pure (Just acc)+                        else withResultSchema queryText result options \schema ->+                            withArrowArray \array -> do+                                checkArrowError queryText (c_duckdb_data_chunk_to_arrow options chunk array)+                                nextAcc <- step acc schema array+                                nextAcc `seq` pure (Just nextAcc)+        case next of+            Nothing -> pure acc+            Just nextAcc -> loop result options nextAcc++-- | Convert the executed result's schema and release it after the action.+withResultSchema :: Query -> Ptr DuckDBResult -> DuckDBArrowOptions -> (Ptr ArrowSchema -> IO a) -> IO a+withResultSchema queryText result options action = do+    count <- c_duckdb_column_count result+    let indices = if count == 0 then [] else [0 .. count - 1]+    withMany (\idx -> bracket (c_duckdb_column_logical_type result idx) destroyLogicalType) indices \types -> do+        names <- forM indices (c_duckdb_column_name result)+        withArray types \typeArray ->+            withArray names \nameArray ->+                alloca \schema ->+                    bracket_ (fillBytes schema 0 (sizeOf (undefined :: ArrowSchema))) (releaseArrowSchema schema) do+                        checkArrowError queryText (c_duckdb_to_arrow_schema options typeArray nameArray count schema)+                        action schema++-- | Allocate an empty Arrow array and release its contents after the action.+withArrowArray :: (Ptr ArrowArray -> IO a) -> IO a+withArrowArray action =+    alloca \array ->+        bracket_ (fillBytes array 0 (sizeOf (undefined :: ArrowArray))) (releaseArrowArray array) (action array)++-- | Convert and release an owned Arrow conversion error.+checkArrowError :: Query -> IO DuckDBErrorData -> IO ()+checkArrowError queryText makeError =+    bracket makeError destroyError \err ->+        when (err /= nullPtr) do+            failed <- c_duckdb_error_data_has_error err+            when (failed /= 0) do+                messagePtr <- c_duckdb_error_data_message err+                message <- if messagePtr == nullPtr then pure "DuckDB Arrow conversion failed" else peekUtf8CString messagePtr+                errorType <- c_duckdb_error_data_error_type err+                throwIO (SQLError message (Just errorType) (Just queryText))++-- | Destroy the result's Arrow conversion options.+destroyArrowOptions :: DuckDBArrowOptions -> IO ()+destroyArrowOptions options = alloca \ptr -> poke ptr options >> c_duckdb_destroy_arrow_options ptr++-- | Destroy an Arrow conversion error handle.+destroyError :: DuckDBErrorData -> IO ()+destroyError err = alloca \ptr -> poke ptr err >> c_duckdb_destroy_error_data ptr

@@ -0,0 +1,97 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Manage callback resources at the DuckDB ownership boundary.+module Database.DuckDB.Simple.Callback (+    withCallbackResources,+    transferCallbackState,+    runCallback,+    ignoreCallbackExceptions,+) where++import Control.Exception (SomeException, catch, displayException, finally, mask, mask_, onException, try)+import Data.IORef (modifyIORef', newIORef, readIORef)+import qualified Data.Text as Text+import qualified Data.Text.Foreign as TextForeign+import Database.DuckDB.FFI (DuckDBDeleteCallback)+import Foreign.C.String (CString, withCString)+import Foreign.Ptr (FunPtr, Ptr, freeHaskellFunPtr, nullPtr)+import Foreign.StablePtr (StablePtr, castPtrToStablePtr, castStablePtrToPtr, deRefStablePtr, freeStablePtr, newStablePtr)++{- | Acquire callbacks and transfer their cleanup to a DuckDB object.+The object must be non-null. Its destructor must run after the action.+-}+withCallbackResources ::+    ((forall a. IO (FunPtr a) -> IO (FunPtr a)) -> IO r) ->+    (Ptr () -> DuckDBDeleteCallback -> IO ()) ->+    (r -> IO b) ->+    IO b+withCallbackResources acquire attach action = mask \restore -> do+    cleanups <- newIORef []+    let cleanup = readIORef cleanups >>= sequence_+        allocate make = do+            ptr <- make+            modifyIORef' cleanups (freeHaskellFunPtr ptr :)+                `onException` freeHaskellFunPtr ptr+            pure ptr+    resources <- acquire allocate `onException` cleanup+    stable <- newStablePtr cleanup `onException` cleanup+    attach (castStablePtrToPtr stable) callbackResourcesDestructor+        `onException` (freeStablePtr stable >> cleanup)+    restore (action resources)++-- | Transfer one state value to a DuckDB callback state slot.+transferCallbackState :: (Ptr () -> DuckDBDeleteCallback -> IO ()) -> a -> IO ()+transferCallbackState attach state = mask_ do+    stable <- newStablePtr state+    attach (castStablePtrToPtr stable) callbackStateDestructor+        `onException` freeStablePtr stable++-- | Convert callback exceptions to DuckDB errors.+runCallback :: (CString -> IO ()) -> IO () -> IO ()+runCallback setError action = mask \restore -> do+    outcome <- try (restore action)+    case outcome of+        Right () -> pure ()+        Left (err :: SomeException) ->+            TextForeign.withCString (Text.pack (displayException err)) setError+                `catch` \(_ :: SomeException) ->+                    ignoreCallbackExceptions $+                        withCString "duckdb-simple: Haskell callback failed" setError++-- | Contain exceptions in callbacks which have no error channel.+ignoreCallbackExceptions :: IO () -> IO ()+ignoreCallbackExceptions action = mask \restore ->+    restore action `catch` \(_ :: SomeException) -> pure ()++-- | Release object-owned callbacks through a static C entry point.+releaseCallbackResources :: Ptr () -> IO ()+releaseCallbackResources raw =+    mask_+        $ ignoreCallbackExceptions+        $ if raw == nullPtr+            then pure ()+            else do+                let stable = castPtrToStablePtr raw :: StablePtr (IO ())+                (deRefStablePtr stable >>= id) `finally` freeStablePtr stable++-- | Release callback state independently of the registered function.+releaseCallbackState :: Ptr () -> IO ()+releaseCallbackState raw =+    mask_+        $ ignoreCallbackExceptions+        $ if raw == nullPtr then pure () else freeStablePtr (castPtrToStablePtr raw)++foreign export ccall "duckdb_simple_release_callback_resources"+    releaseCallbackResources :: Ptr () -> IO ()++foreign import ccall "&duckdb_simple_release_callback_resources"+    callbackResourcesDestructor :: DuckDBDeleteCallback++foreign export ccall "duckdb_simple_release_callback_state"+    releaseCallbackState :: Ptr () -> IO ()++foreign import ccall "&duckdb_simple_release_callback_state"+    callbackStateDestructor :: DuckDBDeleteCallback

@@ -15,8 +15,8 @@ import qualified Data.Text as Text import qualified Data.Text.Foreign as TextForeign import Database.DuckDB.FFI-import Database.DuckDB.Simple.Internal (Connection, withClientContext)-import Foreign.C.String (CString, peekCString)+import Database.DuckDB.Simple.Internal (Connection, peekUtf8CString, throwRegistrationError, withClientContext)+import Foreign.C.String (CString) import Foreign.Marshal.Alloc (alloca) import Foreign.Ptr (nullPtr) import Foreign.Storable (poke)@@ -30,31 +30,37 @@  -- | Look up the backend type name of a named catalog. catalogTypeName :: Connection -> Text -> IO (Maybe Text)-catalogTypeName conn catalogName =-    withClientContext conn \ctx ->-        TextForeign.withCString catalogName \cName ->-            withMaybeCatalog ctx cName \catalog -> do-                namePtr <- c_duckdb_catalog_get_type_name catalog-                if namePtr == nullPtr-                    then pure Nothing-                    else Just . Text.pack <$> peekCString namePtr+catalogTypeName conn catalogName+    | Text.any (== '\0') catalogName = throwRegistrationError "catalog name contains NUL"+    | otherwise =+        withClientContext conn \ctx ->+            TextForeign.withCString catalogName \cName ->+                withMaybeCatalog ctx cName \catalog -> do+                    namePtr <- c_duckdb_catalog_get_type_name catalog+                    if namePtr == nullPtr+                        then pure Nothing+                        else Just <$> peekUtf8CString namePtr  -- | Look up a catalog entry by catalog, schema, name, and expected entry kind. lookupCatalogEntry :: Connection -> Text -> Text -> Text -> DuckDBCatalogEntryType -> IO (Maybe CatalogEntry)-lookupCatalogEntry conn catalogName schemaName entryName entryType =-    withClientContext conn \ctx ->-        TextForeign.withCString catalogName \cCatalog ->-            TextForeign.withCString schemaName \cSchema ->-                TextForeign.withCString entryName \cEntry ->-                    withMaybeCatalog ctx cCatalog \catalog ->-                        withMaybeCatalogEntry catalog ctx entryType cSchema cEntry \entry -> do-                            typ <- c_duckdb_catalog_entry_get_type entry-                            namePtr <- c_duckdb_catalog_entry_get_name entry-                            if namePtr == nullPtr-                                then pure Nothing-                                else do-                                    name <- Text.pack <$> peekCString namePtr-                                    pure (Just CatalogEntry{catalogEntryName = name, catalogEntryType = typ})+lookupCatalogEntry conn catalogName schemaName entryName entryType+    | any (Text.any (== '\0')) [catalogName, schemaName, entryName] = throwRegistrationError "catalog lookup name contains NUL"+    | entryType `notElem` [DuckDBCatalogEntryTypeTable, DuckDBCatalogEntryTypeView, DuckDBCatalogEntryTypeIndex, DuckDBCatalogEntryTypeSequence, DuckDBCatalogEntryTypeCollation, DuckDBCatalogEntryTypeType] =+        throwRegistrationError "unsupported catalog entry type"+    | otherwise =+        withClientContext conn \ctx ->+            TextForeign.withCString catalogName \cCatalog ->+                TextForeign.withCString schemaName \cSchema ->+                    TextForeign.withCString entryName \cEntry ->+                        withMaybeCatalog ctx cCatalog \catalog ->+                            withMaybeCatalogEntry catalog ctx entryType cSchema cEntry \entry -> do+                                typ <- c_duckdb_catalog_entry_get_type entry+                                namePtr <- c_duckdb_catalog_entry_get_name entry+                                if namePtr == nullPtr+                                    then pure Nothing+                                    else do+                                        name <- peekUtf8CString namePtr+                                        pure (Just CatalogEntry{catalogEntryName = name, catalogEntryType = typ})  destroyCatalog :: DuckDBCatalog -> IO () destroyCatalog catalog =@@ -65,11 +71,9 @@     alloca \ptr -> poke ptr entry >> c_duckdb_destroy_catalog_entry ptr  withMaybeCatalog :: DuckDBClientContext -> CString -> (DuckDBCatalog -> IO (Maybe a)) -> IO (Maybe a)-withMaybeCatalog ctx name action = do-    catalog <- c_duckdb_client_context_get_catalog ctx name-    if catalog == nullPtr-        then pure Nothing-        else bracket (pure catalog) destroyCatalog action+withMaybeCatalog ctx name action =+    bracket (c_duckdb_client_context_get_catalog ctx name) destroyCatalog \catalog ->+        if catalog == nullPtr then pure Nothing else action catalog  withMaybeCatalogEntry ::     DuckDBCatalog ->@@ -79,8 +83,6 @@     CString ->     (DuckDBCatalogEntry -> IO (Maybe a)) ->     IO (Maybe a)-withMaybeCatalogEntry catalog ctx entryType schemaName entryName action = do-    entry <- c_duckdb_catalog_get_entry catalog ctx entryType schemaName entryName-    if entry == nullPtr-        then pure Nothing-        else bracket (pure entry) destroyCatalogEntry action+withMaybeCatalogEntry catalog ctx entryType schemaName entryName action =+    bracket (c_duckdb_catalog_get_entry catalog ctx entryType schemaName entryName) destroyCatalogEntry \entry ->+        if entry == nullPtr then pure Nothing else action entry

@@ -16,8 +16,7 @@ import qualified Data.Text as Text import qualified Data.Text.Foreign as TextForeign import Database.DuckDB.FFI-import Database.DuckDB.Simple.Internal (Connection, destroyValue, withClientContext)-import Foreign.C.String (peekCString)+import Database.DuckDB.Simple.Internal (Connection, destroyValue, peekUtf8CString, throwRegistrationError, withClientContext) import Foreign.Marshal.Alloc (alloca) import Foreign.Ptr (castPtr, nullPtr) import Foreign.Storable (peek, poke)@@ -50,39 +49,41 @@                 if rc /= DuckDBSuccess                     then pure ConfigFlag{configFlagName = Text.pack (show idx), configFlagDescription = Text.pack ""}                     else do-                        name <- peek namePtr >>= peekCString-                        description <- peek descPtr >>= peekCString-                        pure ConfigFlag{configFlagName = Text.pack name, configFlagDescription = Text.pack description}+                        name <- peek namePtr >>= peekUtf8CString+                        description <- peek descPtr >>= peekUtf8CString+                        pure ConfigFlag{configFlagName = name, configFlagDescription = description}  -- | Read a configuration option from a live connection's client context. getConfigOption :: Connection -> Text -> IO (Maybe ConfigValue)-getConfigOption conn name =-    withClientContext conn \ctx ->-        TextForeign.withCString name \cName ->-            alloca \scopePtr -> do-                poke scopePtr DuckDBConfigOptionScopeInvalid-                value <- c_duckdb_client_context_get_config_option ctx cName scopePtr-                if value == nullPtr-                    then pure Nothing-                    else bracket-                        (pure value)+getConfigOption conn name+    | Text.any (== '\0') name = throwRegistrationError "config option name contains NUL"+    | otherwise =+        withClientContext conn \ctx ->+            TextForeign.withCString name \cName ->+                alloca \scopePtr -> do+                    poke scopePtr DuckDBConfigOptionScopeInvalid+                    bracket+                        (c_duckdb_client_context_get_config_option ctx cName scopePtr)                         destroyValue-                        \duckValue -> do-                            strPtr <- c_duckdb_get_varchar duckValue-                            rendered <--                                if strPtr == nullPtr-                                    then pure Text.empty-                                    else do-                                        txt <- Text.pack <$> peekCString strPtr-                                        c_duckdb_free (castPtr strPtr)-                                        pure txt-                            scope <- peek scopePtr-                            pure $-                                Just-                                    ConfigValue-                                        { configValueText = rendered-                                        , configValueScope =-                                            if scope == DuckDBConfigOptionScopeInvalid-                                                then Nothing-                                                else Just scope-                                        }+                        \duckValue ->+                            if duckValue == nullPtr+                                then pure Nothing+                                else do+                                    rendered <-+                                        bracket+                                            (c_duckdb_get_varchar duckValue)+                                            (c_duckdb_free . castPtr)+                                            \strPtr ->+                                                if strPtr == nullPtr+                                                    then pure Text.empty+                                                    else peekUtf8CString strPtr+                                    scope <- peek scopePtr+                                    pure $+                                        Just+                                            ConfigValue+                                                { configValueText = rendered+                                                , configValueScope =+                                                    if scope == DuckDBConfigOptionScopeInvalid+                                                        then Nothing+                                                        else Just scope+                                                }

@@ -14,19 +14,19 @@     registerCopyToFunction, ) where -import Control.Exception (SomeException, bracket, displayException, onException, try)+import Control.Exception (bracket) import Control.Monad (forM, when) import Data.Text (Text) import qualified Data.Text as Text import qualified Data.Text.Foreign as TextForeign import Database.DuckDB.FFI+import Database.DuckDB.Simple.Callback (runCallback, transferCallbackState, withCallbackResources) import Database.DuckDB.Simple.FromField (Field (..))-import Database.DuckDB.Simple.Internal (Connection, destroyLogicalType, mkDeleteCallback, releaseStablePtrData, throwRegistrationError, withConnectionHandle)+import Database.DuckDB.Simple.Internal (Connection, destroyLogicalType, peekUtf8CString, throwRegistrationError, withConnectionHandle) import Database.DuckDB.Simple.Materialize (materializeValue)-import Foreign.C.String (CString, peekCString) import Foreign.Marshal.Alloc (alloca)-import Foreign.Ptr (Ptr, freeHaskellFunPtr, nullPtr)-import Foreign.StablePtr (StablePtr, castPtrToStablePtr, castStablePtrToPtr, deRefStablePtr, freeStablePtr, newStablePtr)+import Foreign.Ptr (Ptr, nullPtr)+import Foreign.StablePtr (StablePtr, castPtrToStablePtr, deRefStablePtr) import Foreign.Storable (poke)  -- | Bind-phase metadata for a custom `COPY ... TO` function.@@ -58,7 +58,6 @@     , copyInitPtr :: !DuckDBCopyFunctionGlobalInitFun     , copySinkPtr :: !DuckDBCopyFunctionSinkFun     , copyFinalizePtr :: !DuckDBCopyFunctionFinalizeFun-    , copyStateDestroyPtr :: !DuckDBDeleteCallback     }  -- | Register a custom `COPY ... TO` implementation backed by Haskell callbacks.@@ -72,66 +71,47 @@     (CopyFinalizeInfo bindState globalState -> IO ()) ->     IO () registerCopyToFunction conn name bindFn initFn sinkFn finalizeFn = do-    stateDestroyCb <- mkDeleteCallback releaseStablePtrData-    bindPtr <- mkCopyBindFun (copyBindHandler stateDestroyCb bindFn)-    initPtr <- mkCopyGlobalInitFun (copyGlobalInitHandler stateDestroyCb initFn)-    sinkPtr <- mkCopySinkFun (copySinkHandler sinkFn)-    finalizePtr <- mkCopyFinalizeFun (copyFinalizeHandler finalizeFn)-    resources <--        newStablePtr-            CopyFunctionResources-                { copyBindPtr = bindPtr-                , copyInitPtr = initPtr-                , copySinkPtr = sinkPtr-                , copyFinalizePtr = finalizePtr-                , copyStateDestroyPtr = stateDestroyCb-                }-    destroyCb <- mkDeleteCallback releaseCopyResources-    let release =-            freeHaskellFunPtr bindPtr-                >> freeHaskellFunPtr initPtr-                >> freeHaskellFunPtr sinkPtr-                >> freeHaskellFunPtr finalizePtr-                >> freeHaskellFunPtr stateDestroyCb-                >> freeStablePtr resources-                >> freeHaskellFunPtr destroyCb-    bracket c_duckdb_create_copy_function destroyCopyFunction \copyFun ->-        (`onException` release) $ do-            TextForeign.withCString name \cName ->-                c_duckdb_copy_function_set_name copyFun cName-            c_duckdb_copy_function_set_bind copyFun bindPtr-            c_duckdb_copy_function_set_global_init copyFun initPtr-            c_duckdb_copy_function_set_sink copyFun sinkPtr-            c_duckdb_copy_function_set_finalize copyFun finalizePtr-            c_duckdb_copy_function_set_extra_info copyFun (castStablePtrToPtr resources) destroyCb-            withConnectionHandle conn \connPtr -> do-                rc <- c_duckdb_register_copy_function connPtr copyFun-                if rc == DuckDBSuccess-                    then pure ()-                    else throwRegistrationError "register copy function"+    when (Text.null name || Text.any (== '\0') name) $+        throwRegistrationError "invalid copy function name"+    bracket c_duckdb_create_copy_function destroyCopyFunction \copyFun -> do+        when (copyFun == nullPtr) $ throwRegistrationError "allocate copy function"+        withCallbackResources+            ( \allocate -> do+                copyBindPtr <- allocate (mkCopyBindFun (copyBindHandler bindFn))+                copyInitPtr <- allocate (mkCopyGlobalInitFun (copyGlobalInitHandler initFn))+                copySinkPtr <- allocate (mkCopySinkFun (copySinkHandler sinkFn))+                copyFinalizePtr <- allocate (mkCopyFinalizeFun (copyFinalizeHandler finalizeFn))+                pure CopyFunctionResources{copyBindPtr, copyInitPtr, copySinkPtr, copyFinalizePtr}+            )+            (c_duckdb_copy_function_set_extra_info copyFun)+            \CopyFunctionResources{copyBindPtr, copyInitPtr, copySinkPtr, copyFinalizePtr} -> do+                TextForeign.withCString name $ c_duckdb_copy_function_set_name copyFun+                c_duckdb_copy_function_set_bind copyFun copyBindPtr+                c_duckdb_copy_function_set_global_init copyFun copyInitPtr+                c_duckdb_copy_function_set_sink copyFun copySinkPtr+                c_duckdb_copy_function_set_finalize copyFun copyFinalizePtr+                withConnectionHandle conn \connPtr -> do+                    rc <- c_duckdb_register_copy_function connPtr copyFun+                    when (rc /= DuckDBSuccess) $ throwRegistrationError "register copy function"  copyBindHandler ::     forall bindState.-    DuckDBDeleteCallback ->     (CopyBindInfo -> IO bindState) ->     DuckDBCopyFunctionBindInfo ->     IO ()-copyBindHandler destroyCb bindFn info = do-    outcome <- try do+copyBindHandler bindFn info =+    runCallback (c_duckdb_copy_function_bind_set_error info) do         copyBindColumnTypes <- fetchColumnTypes info         bindState <- bindFn CopyBindInfo{copyBindColumnTypes}-        stable <- newStablePtr bindState-        c_duckdb_copy_function_bind_set_bind_data info (castStablePtrToPtr stable) destroyCb-    reportCopyError c_duckdb_copy_function_bind_set_error info outcome+        transferCallbackState (c_duckdb_copy_function_bind_set_bind_data info) bindState  copyGlobalInitHandler ::     forall bindState globalState.-    DuckDBDeleteCallback ->     (CopyInitInfo bindState -> IO globalState) ->     DuckDBCopyFunctionGlobalInitInfo ->     IO ()-copyGlobalInitHandler destroyCb initFn info = do-    outcome <- try do+copyGlobalInitHandler initFn info =+    runCallback (c_duckdb_copy_function_global_init_set_error info) do         rawBindState <- c_duckdb_copy_function_global_init_get_bind_data info         when (rawBindState == nullPtr) $             throwRegistrationError "missing copy bind state"@@ -140,11 +120,9 @@         filePath <-             if pathPtr == nullPtr                 then pure ""-                else peekCString pathPtr+                else Text.unpack <$> peekUtf8CString pathPtr         globalState <- initFn CopyInitInfo{copyInitBindState = bindState, copyInitFilePath = filePath}-        stable <- newStablePtr globalState-        c_duckdb_copy_function_global_init_set_global_state info (castStablePtrToPtr stable) destroyCb-    reportCopyError c_duckdb_copy_function_global_init_set_error info outcome+        transferCallbackState (c_duckdb_copy_function_global_init_set_global_state info) globalState  copySinkHandler ::     forall bindState globalState.@@ -152,41 +130,33 @@     DuckDBCopyFunctionSinkInfo ->     DuckDBDataChunk ->     IO ()-copySinkHandler sinkFn info chunk = do-    outcome <- try do+copySinkHandler sinkFn info chunk =+    runCallback (c_duckdb_copy_function_sink_set_error info) do         bindState <- readStablePtrState c_duckdb_copy_function_sink_get_bind_data info         globalState <- readStablePtrState c_duckdb_copy_function_sink_get_global_state info         rows <- materializeChunkRows chunk         sinkFn CopySinkInfo{copySinkBindState = bindState, copySinkGlobalState = globalState} rows-    reportCopyError c_duckdb_copy_function_sink_set_error info outcome  copyFinalizeHandler ::     forall bindState globalState.     (CopyFinalizeInfo bindState globalState -> IO ()) ->     DuckDBCopyFunctionFinalizeInfo ->     IO ()-copyFinalizeHandler finalizeFn info = do-    outcome <- try do+copyFinalizeHandler finalizeFn info =+    runCallback (c_duckdb_copy_function_finalize_set_error info) do         bindState <- readStablePtrState c_duckdb_copy_function_finalize_get_bind_data info         globalState <- readStablePtrState c_duckdb_copy_function_finalize_get_global_state info         finalizeFn CopyFinalizeInfo{copyFinalizeBindState = bindState, copyFinalizeGlobalState = globalState}-    reportCopyError c_duckdb_copy_function_finalize_set_error info outcome -reportCopyError :: (i -> CString -> IO ()) -> i -> Either SomeException () -> IO ()-reportCopyError _ _ (Right ()) = pure ()-reportCopyError setError info (Left err) =-    TextForeign.withCString (Text.pack (displayException err)) \cMsg ->-        setError info cMsg- fetchColumnTypes :: DuckDBCopyFunctionBindInfo -> IO [DuckDBType] fetchColumnTypes info = do     count <- c_duckdb_copy_function_bind_get_column_count info     let indices = [0 .. fromIntegral count - 1] :: [Int]     forM indices \idx -> do-        logical <- c_duckdb_copy_function_bind_get_column_type info (fromIntegral idx)-        dtype <- c_duckdb_get_type_id logical-        destroyLogicalType logical-        pure dtype+        bracket+            (c_duckdb_copy_function_bind_get_column_type info (fromIntegral idx))+            destroyLogicalType+            c_duckdb_get_type_id  readStablePtrState :: forall a i. (i -> IO (Ptr ())) -> i -> IO a readStablePtrState getter info = do@@ -211,27 +181,13 @@ makeColumnReader :: DuckDBDataChunk -> Int -> IO ColumnReader makeColumnReader chunk columnIndex = do     vector <- c_duckdb_data_chunk_get_vector chunk (fromIntegral columnIndex)-    logical <- c_duckdb_vector_get_column_type vector-    dtype <- c_duckdb_get_type_id logical-    destroyLogicalType logical+    dtype <- bracket (c_duckdb_vector_get_column_type vector) destroyLogicalType c_duckdb_get_type_id     dataPtr <- c_duckdb_vector_get_data vector     validity <- c_duckdb_vector_get_validity vector     let name = Text.pack ("column" <> show columnIndex)     pure \rowIdx -> do         fieldValue <- materializeValue dtype vector dataPtr validity (fromIntegral rowIdx)         pure Field{fieldName = name, fieldIndex = columnIndex, fieldValue}--releaseCopyResources :: Ptr () -> IO ()-releaseCopyResources rawPtr =-    when (rawPtr /= nullPtr) $ do-        let stablePtr = castPtrToStablePtr rawPtr :: StablePtr CopyFunctionResources-        CopyFunctionResources{copyBindPtr, copyInitPtr, copySinkPtr, copyFinalizePtr, copyStateDestroyPtr} <- deRefStablePtr stablePtr-        freeHaskellFunPtr copyBindPtr-        freeHaskellFunPtr copyInitPtr-        freeHaskellFunPtr copySinkPtr-        freeHaskellFunPtr copyFinalizePtr-        freeHaskellFunPtr copyStateDestroyPtr-        freeStablePtr stablePtr  destroyCopyFunction :: DuckDBCopyFunction -> IO () destroyCopyFunction copyFun =

@@ -0,0 +1,87 @@+{-# LANGUAGE BlockArguments #-}++{- |+Module      : Database.DuckDB.Simple.Deprecated.Streaming+Description : Optional native streaming through DuckDB's deprecated execution API.++Import this module qualified to request native streaming. DuckDB can still+materialize a result, and operators such as sorting can buffer data. Streaming+reduces result storage when the query supports it; it does not bound all query+memory. The default @Database.DuckDB.Simple@ API materializes native results.++Rows and Arrow batches use the same decoders and cleanup as the default API.+Cancellation also interrupts native chunk fetching. Prompt interruption requires+@-threaded@ and waits for native code and Haskell callbacks to return.++Serialize connection use for the entire cursor or fold. Another query on the+same connection can invalidate an active stream. Close or reset an abandoned+statement to release its result.+-}+module Database.DuckDB.Simple.Deprecated.Streaming+    {-# DEPRECATED "Uses DuckDB's deprecated streaming execution API. Prefer Database.DuckDB.Simple or Database.DuckDB.Simple.Arrow when materialization is acceptable." #-} (+    fold,+    fold_,+    foldNamed,+    nextRow,+    nextRowWith,+    foldArrow,+    foldArrow_,+) where++import Database.DuckDB.FFI (ArrowArray, ArrowSchema)+import Database.DuckDB.Simple (bind, bindNamed, withStatement)+import qualified Database.DuckDB.Simple.Arrow.Internal as Arrow+import Database.DuckDB.Simple.FromRow (FromRow (..), RowParser)+import Database.DuckDB.Simple.Internal (Connection, Query, ResultMode (StreamingResult), Statement)+import qualified Database.DuckDB.Simple.Result as Result+import Database.DuckDB.Simple.ToField (NamedParam)+import Database.DuckDB.Simple.ToRow (ToRow (..))+import Foreign.Ptr (Ptr)++-- | Fold a parameterized query, requesting native streaming.+fold :: (FromRow row, ToRow params) => Connection -> Query -> params -> a -> (a -> row -> IO a) -> IO a+fold conn sql params initial step =+    withStatement conn sql \stmt -> do+        bind stmt (toRow params)+        Result.foldStatementWith StreamingResult fromRow stmt initial step++-- | Fold a query without parameters, requesting native streaming.+fold_ :: (FromRow row) => Connection -> Query -> a -> (a -> row -> IO a) -> IO a+fold_ conn sql initial step =+    withStatement conn sql \stmt ->+        Result.foldStatementWith StreamingResult fromRow stmt initial step++-- | Fold a query with named parameters, requesting native streaming.+foldNamed :: (FromRow row) => Connection -> Query -> [NamedParam] -> a -> (a -> row -> IO a) -> IO a+foldNamed conn sql params initial step =+    withStatement conn sql \stmt -> do+        bindNamed stmt params+        Result.foldStatementWith StreamingResult fromRow stmt initial step++-- | Fetch the next row, requesting native streaming on the first fetch.+nextRow :: (FromRow r) => Statement -> IO (Maybe r)+nextRow = nextRowWith fromRow++{- | Fetch a row with a custom parser, requesting streaming on the first fetch.+  The active result retains its execution mode until reset. Switching between+  this module and the default @Database.DuckDB.Simple.nextRow@ does not restart+  or change an active result. EOF remains exhausted until an explicit reset.+-}+nextRowWith :: RowParser r -> Statement -> IO (Maybe r)+nextRowWith = Result.nextRowWith StreamingResult++{- | Fold Arrow batches, requesting native streaming.+  Each batch has a separate schema and array. The callback may read, release,+  or move them under the Arrow C Data Interface. The pointers themselves are+  valid during the callback only. See @Database.DuckDB.Simple.Arrow@ for the+  ownership rules.+  Arrow conversion uses the supported API; query execution uses the deprecated+  streaming entry point. The fold releases contents that the consumer has not+  released or moved. The consumer owns any moved contents.+-}+foldArrow :: (ToRow params) => Connection -> Query -> params -> a -> (a -> Ptr ArrowSchema -> Ptr ArrowArray -> IO a) -> IO a+foldArrow = Arrow.foldArrowWith StreamingResult++-- | Fold Arrow batches from a query without parameters.+foldArrow_ :: Connection -> Query -> a -> (a -> Ptr ArrowSchema -> Ptr ArrowArray -> IO a) -> IO a+foldArrow_ conn sql = foldArrow conn sql ()

@@ -14,49 +14,60 @@     fileHandleSync, ) where -import Control.Exception (bracket, throwIO)+import Control.Exception (bracket, mask_, throwIO) import qualified Data.ByteString as BS import Data.Int (Int64) import Data.Text (Text) import qualified Data.Text as Text+import qualified Data.Text.Foreign as TextForeign import Database.DuckDB.FFI-import Database.DuckDB.Simple.Internal (Connection, SQLError (..), withClientContext)-import Foreign.C.String (peekCString, withCString)+import Database.DuckDB.Simple.Internal (Connection, SQLError (..), peekUtf8CString, withClientContext) import Foreign.Marshal.Alloc (alloca, free, mallocBytes)-import Foreign.Ptr (castPtr, nullPtr)+import Foreign.Ptr (Ptr, castPtr, nullPtr) import Foreign.Storable (peek, poke)  -- | Open a file through DuckDB's file-system layer for the duration of an action. withFileHandle :: Connection -> FilePath -> [DuckDBFileFlag] -> (DuckDBFileHandle -> IO a) -> IO a-withFileHandle conn path flags action =-    withFileSystem conn \fs ->-        bracket-            c_duckdb_create_file_open_options-            destroyFileOpenOptions-            \opts -> do-                mapM_ (\flag -> expectState "set file-open flag" (c_duckdb_file_open_options_set_flag opts flag 1)) flags-                withCString path \cPath ->-                    alloca \filePtr -> do-                        rc <- c_duckdb_file_system_open fs cPath opts filePtr-                        if rc /= DuckDBSuccess-                            then throwFileSystemError fs path-                            else do-                                handle <- peek filePtr-                                bracket (pure handle) destroyFileHandle action+withFileHandle conn path flags action+    | '\0' `elem` path = throwIO (SQLError (Text.pack "duckdb-simple: file path contains NUL") Nothing Nothing)+    | otherwise =+        withFileSystem conn \fs ->+            bracket+                c_duckdb_create_file_open_options+                destroyFileOpenOptions+                \opts -> do+                    whenNull opts "allocate file-open options"+                    mapM_ (\flag -> expectState "set file-open flag" (c_duckdb_file_open_options_set_flag opts flag 1)) flags+                    TextForeign.withCString (Text.pack path) \cPath ->+                        bracket+                            ( alloca \filePtr -> do+                                poke filePtr nullPtr+                                rc <- c_duckdb_file_system_open fs cPath opts filePtr+                                if rc /= DuckDBSuccess+                                    then throwFileSystemError fs path+                                    else do+                                        handle <- peek filePtr+                                        whenNull handle "allocate file handle"+                                        pure handle+                            )+                            destroyFileHandle+                            action  -- | Read up to the requested number of bytes from a file handle. readFileHandleChunk :: DuckDBFileHandle -> Int64 -> IO BS.ByteString readFileHandleChunk handle requested     | requested <= 0 = pure BS.empty-    | otherwise = do-        raw <- mallocBytes (fromIntegral requested)-        bytesRead <- c_duckdb_file_handle_read handle raw requested-        if bytesRead < 0-            then free raw >> throwFileHandleError handle (Text.pack "read failed")-            else do-                bs <- BS.packCStringLen (castPtr raw, fromIntegral bytesRead)-                free raw-                pure bs+    | toInteger requested > toInteger (maxBound :: Int) =+        throwIO (SQLError (Text.pack "duckdb-simple: file read size exceeds Int range") Nothing Nothing)+    | otherwise =+        bracket (mallocBytes (fromIntegral requested)) free \raw -> do+            bytesRead <- c_duckdb_file_handle_read handle raw requested+            if bytesRead < 0+                then throwFileHandleError handle (Text.pack "read failed")+                else+                    if bytesRead > requested+                        then throwFileHandleError handle (Text.pack "read size exceeds buffer size")+                        else BS.packCStringLen (castPtr raw, fromIntegral bytesRead)  -- | Write an entire bytestring to a file handle. writeFileHandleBytes :: DuckDBFileHandle -> BS.ByteString -> IO Int64@@ -101,7 +112,7 @@         bracket             (c_duckdb_client_context_get_file_system ctx)             destroyFileSystem-            action+            (\fs -> whenNull fs "allocate file system" >> action fs)  destroyFileSystem :: DuckDBFileSystem -> IO () destroyFileSystem fs =@@ -116,12 +127,12 @@     alloca \ptr -> poke ptr handle >> c_duckdb_destroy_file_handle ptr  throwFileSystemError :: DuckDBFileSystem -> FilePath -> IO a-throwFileSystemError fs path = do+throwFileSystemError fs path = mask_ do     err <- c_duckdb_file_system_error_data fs     throwErrorData err (Text.concat [Text.pack "duckdb-simple: failed to open file ", Text.pack path])  throwFileHandleError :: DuckDBFileHandle -> Text -> IO a-throwFileHandleError handle fallback = do+throwFileHandleError handle fallback = mask_ do     err <- c_duckdb_file_handle_error_data handle     throwErrorData err fallback @@ -133,7 +144,7 @@         message <-             if msgPtr == nullPtr                 then pure fallback-                else Text.pack <$> peekCString msgPtr+                else peekUtf8CString msgPtr         throwIO             SQLError                 { sqlErrorMessage = message@@ -157,3 +168,8 @@                     , sqlErrorType = Nothing                     , sqlErrorQuery = Nothing                     }++-- | Reject a null file-system handle before use.+whenNull :: Ptr a -> String -> IO ()+whenNull ptr label =+    if ptr == nullPtr then expectState label (pure DuckDBError) else pure ()

@@ -66,7 +66,9 @@     UnionValue (..),  ) import Database.DuckDB.Simple.Ok+import Database.DuckDB.Simple.Time (Date, LocalTimestamp, UTCTimestamp, Unbounded (..)) import Database.DuckDB.Simple.Types (Null (..))+import GHC.Float (double2Float, float2Double) import GHC.Num.Integer (integerFromWordList) import Numeric.Natural (Natural) @@ -87,14 +89,14 @@     | FieldText Text     | FieldBool Bool     | FieldBlob BS.ByteString-    | FieldDate Day+    | FieldDate Date     | FieldTime TimeOfDay-    | FieldTimestamp LocalTime+    | FieldTimestamp LocalTimestamp     | FieldInterval IntervalValue     | FieldHugeInt Integer     | FieldUHugeInt Integer     | FieldDecimal DecimalValue-    | FieldTimestampTZ UTCTime+    | FieldTimestampTZ UTCTimestamp     | FieldTimeTZ TimeWithZone     | FieldBit BitString     | FieldBigNum BigNum@@ -219,30 +221,30 @@     }     deriving (Eq, Show) --- | Pattern synonym to make it easier to match on any integral type.-pattern FieldInt :: Int -> FieldValue+-- | Match signed values without narrowing to the machine word size.+pattern FieldInt :: Int64 -> FieldValue pattern FieldInt i <- (fieldValueToInt -> Just i)-    where-        FieldInt i = FieldInt64 (fromIntegral i)+  where+    FieldInt i = FieldInt64 i -fieldValueToInt :: FieldValue -> Maybe Int+fieldValueToInt :: FieldValue -> Maybe Int64 fieldValueToInt (FieldInt8 i) = Just (fromIntegral i) fieldValueToInt (FieldInt16 i) = Just (fromIntegral i) fieldValueToInt (FieldInt32 i) = Just (fromIntegral i)-fieldValueToInt (FieldInt64 i) = Just (fromIntegral i)+fieldValueToInt (FieldInt64 i) = Just i fieldValueToInt _ = Nothing --- | Pattern synonym to make it easier to match on any word size-pattern FieldWord :: Word -> FieldValue+-- | Match unsigned values without narrowing to the machine word size.+pattern FieldWord :: Word64 -> FieldValue pattern FieldWord i <- (fieldValueToWord -> Just i)-    where-        FieldWord i = FieldWord64 (fromIntegral i)+  where+    FieldWord i = FieldWord64 i -fieldValueToWord :: FieldValue -> Maybe Word+fieldValueToWord :: FieldValue -> Maybe Word64 fieldValueToWord (FieldWord8 i) = Just (fromIntegral i) fieldValueToWord (FieldWord16 i) = Just (fromIntegral i) fieldValueToWord (FieldWord32 i) = Just (fromIntegral i)-fieldValueToWord (FieldWord64 i) = Just (fromIntegral i)+fieldValueToWord (FieldWord64 i) = Just i fieldValueToWord _ = Nothing  -- | Metadata for a single column in a row.@@ -345,7 +347,7 @@ instance FromField Int8 where     fromField f@Field{fieldValue} =         case fieldValue of-            FieldInt i -> Ok (fromIntegral i)+            FieldInt i -> boundedIntegral f i             FieldHugeInt value -> boundedFromInteger f value             FieldUHugeInt value -> boundedFromInteger f value             FieldEnum value -> boundedFromInteger f (fromIntegral value)@@ -355,7 +357,7 @@ instance FromField Int64 where     fromField f@Field{fieldValue} =         case fieldValue of-            FieldInt i -> Ok (fromIntegral i)+            FieldInt i -> Ok i             FieldHugeInt value -> boundedFromInteger f value             FieldUHugeInt value -> boundedFromInteger f value             FieldEnum value -> Ok (fromIntegral value)@@ -431,7 +433,7 @@                 | i >= 0 -> Ok (fromIntegral i)                 | otherwise ->                     returnError ConversionFailed f "negative value cannot be converted to unsigned integer"-            FieldWord w -> Ok (fromIntegral w)+            FieldWord w -> Ok w             FieldHugeInt value                 | value >= 0 -> boundedFromInteger f value                 | otherwise ->@@ -448,7 +450,7 @@                 | i >= 0 -> boundedIntegral f i                 | otherwise ->                     returnError ConversionFailed f "negative value cannot be converted to unsigned integer"-            FieldWord w -> Ok (fromIntegral w)+            FieldWord w -> boundedFromInteger f (toInteger w)             FieldHugeInt value                 | value >= 0 -> boundedFromInteger f value                 | otherwise ->@@ -465,7 +467,7 @@                 | i >= 0 -> boundedIntegral f i                 | otherwise ->                     returnError ConversionFailed f "negative value cannot be converted to unsigned integer"-            FieldWord w -> Ok (fromIntegral w)+            FieldWord w -> boundedFromInteger f (toInteger w)             FieldHugeInt value                 | value >= 0 -> boundedFromInteger f value                 | otherwise ->@@ -482,7 +484,7 @@                 | i >= 0 -> boundedIntegral f i                 | otherwise ->                     returnError ConversionFailed f "negative value cannot be converted to unsigned integer"-            FieldWord w -> Ok (fromIntegral w)+            FieldWord w -> boundedFromInteger f (toInteger w)             FieldHugeInt value                 | value >= 0 -> boundedFromInteger f value                 | otherwise ->@@ -499,7 +501,7 @@                 | i >= 0 -> boundedFromInteger f (fromIntegral i)                 | otherwise ->                     returnError ConversionFailed f "negative value cannot be converted to unsigned integer"-            FieldWord w -> Ok w+            FieldWord w -> boundedFromInteger f (toInteger w)             FieldHugeInt value                 | value >= 0 -> boundedFromInteger f value                 | otherwise ->@@ -513,6 +515,7 @@     fromField f@Field{fieldValue} =         case fieldValue of             FieldDouble d -> Ok d+            FieldFloat value -> Ok (float2Double value)             FieldInt i -> Ok (fromIntegral i)             FieldDecimal DecimalValue{decimalInteger, decimalScale} ->                 Ok (realToFrac decimalInteger / 10 ^ decimalScale)@@ -523,13 +526,17 @@     fromField field =         case (fromField field :: Ok Double) of             Errors err -> Errors err-            Ok d -> Ok (realToFrac d)+            Ok d+                | not (isInfinite d || isNaN d) && isInfinite (double2Float d) ->+                    returnError ConversionFailed field "floating-point value out of bounds"+                | otherwise -> Ok (double2Float d)  instance FromField Text where     fromField f@Field{fieldValue} =         case fieldValue of             FieldText t -> Ok t             FieldInt i -> Ok (Text.pack (show i))+            FieldFloat value -> Ok (Text.pack (show value))             FieldDouble d -> Ok (Text.pack (show d))             FieldBool b -> Ok (if b then Text.pack "1" else Text.pack "0")             FieldNull -> returnError UnexpectedNull f ""@@ -640,8 +647,16 @@ instance FromField Day where     fromField f@Field{fieldValue} =         case fieldValue of+            FieldDate day -> finiteField f day+            FieldTimestamp timestamp -> finiteField f (localDay <$> timestamp)+            FieldNull -> returnError UnexpectedNull f ""+            _ -> returnError Incompatible f ""++instance FromField (Unbounded Day) where+    fromField f@Field{fieldValue} =+        case fieldValue of             FieldDate day -> Ok day-            FieldTimestamp LocalTime{localDay} -> Ok localDay+            FieldTimestamp timestamp -> Ok (localDay <$> timestamp)             FieldNull -> returnError UnexpectedNull f ""             _ -> returnError Incompatible f "" @@ -649,7 +664,7 @@     fromField f@Field{fieldValue} =         case fieldValue of             FieldTime tod -> Ok tod-            FieldTimestamp LocalTime{localTimeOfDay} -> Ok localTimeOfDay+            FieldTimestamp timestamp -> finiteField f (localTimeOfDay <$> timestamp)             FieldNull -> returnError UnexpectedNull f ""             _ -> returnError Incompatible f "" @@ -663,9 +678,20 @@ instance FromField LocalTime where     fromField f@Field{fieldValue} =         case fieldValue of+            FieldTimestamp ts -> finiteField f ts+            FieldDate day -> finiteField f ((\value -> LocalTime value midnight) <$> day)+            FieldTimestampTZ utcTime -> finiteField f (utcToLocalTime utc <$> utcTime)+            FieldNull -> returnError UnexpectedNull f ""+            _ -> returnError Incompatible f ""+      where+        midnight = TimeOfDay 0 0 0++instance FromField (Unbounded LocalTime) where+    fromField f@Field{fieldValue} =+        case fieldValue of             FieldTimestamp ts -> Ok ts-            FieldDate day -> Ok (LocalTime day midnight)-            FieldTimestampTZ utcTime -> Ok (utcToLocalTime utc utcTime)+            FieldDate day -> Ok ((\value -> LocalTime value midnight) <$> day)+            FieldTimestampTZ utcTime -> Ok (utcToLocalTime utc <$> utcTime)             FieldNull -> returnError UnexpectedNull f ""             _ -> returnError Incompatible f ""       where@@ -681,26 +707,37 @@ instance FromField UTCTime where     fromField f@Field{fieldValue} =         case fieldValue of-            FieldTimestamp ts -> Ok (localTimeToUTC utc ts)+            FieldTimestamp ts -> finiteField f (localTimeToUTC utc <$> ts)+            FieldTimestampTZ utcTime -> finiteField f utcTime+            FieldDate day -> finiteField f ((\value -> localTimeToUTC utc (LocalTime value midnight)) <$> day)+            FieldNull -> returnError UnexpectedNull f ""+            _ -> returnError Incompatible f ""+      where+        midnight = TimeOfDay 0 0 0++instance FromField (Unbounded UTCTime) where+    fromField f@Field{fieldValue} =+        case fieldValue of+            FieldTimestamp ts -> Ok (localTimeToUTC utc <$> ts)             FieldTimestampTZ utcTime -> Ok utcTime-            FieldDate day -> Ok (localTimeToUTC utc (LocalTime day midnight))+            FieldDate day -> Ok ((\value -> localTimeToUTC utc (LocalTime value midnight)) <$> day)             FieldNull -> returnError UnexpectedNull f ""             _ -> returnError Incompatible f ""       where         midnight = TimeOfDay 0 0 0 +-- | Reject infinity when the requested Haskell type holds only finite values.+finiteField :: (Typeable a) => Field -> Unbounded a -> Ok a+finiteField _ (Finite value) = Ok value+finiteField field _ = returnError ConversionFailed field "infinity requires an Unbounded date or timestamp"+ instance (FromField a) => FromField (Maybe a) where     fromField Field{fieldValue = FieldNull} = Ok Nothing     fromField field = Just <$> fromField field  -- | Helper for bounded integral conversions.-boundedIntegral :: forall a. (Integral a, Bounded a, Typeable a) => Field -> Int -> Ok a-boundedIntegral f@Field{} i-    | toInteger i < toInteger (minBound :: a) =-        returnError ConversionFailed f "integer value out of bounds"-    | toInteger i > toInteger (maxBound :: a) =-        returnError ConversionFailed f "integer value out of bounds"-    | otherwise = Ok (fromIntegral i)+boundedIntegral :: forall a. (Integral a, Bounded a, Typeable a) => Field -> Int64 -> Ok a+boundedIntegral f = boundedFromInteger f . toInteger  boundedFromInteger :: forall a. (Integral a, Bounded a, Typeable a) => Field -> Integer -> Ok a boundedFromInteger f@Field{} value

@@ -3,6 +3,7 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE UndecidableInstances #-} @@ -30,8 +31,6 @@ import Control.Exception (     SomeException,     bracket,-    displayException,-    onException,     throwIO,     try,  )@@ -43,6 +42,7 @@ import qualified Data.Text.Foreign as TextForeign import Data.Word (Word16, Word32, Word64, Word8) import Database.DuckDB.FFI+import Database.DuckDB.Simple.Callback (runCallback, transferCallbackState, withCallbackResources) import Database.DuckDB.Simple.FromField (     Field (..),     FromField (..),@@ -52,18 +52,17 @@     Query (..),     SQLError (..),     destroyLogicalType,-    mkDeleteCallback,-    releaseStablePtrData,     withConnectionHandle,     withQueryCString,+    withResult,  ) import Database.DuckDB.Simple.Materialize (materializeValue) import Database.DuckDB.Simple.Ok (Ok (..))-import Foreign.C.String (peekCString) import Foreign.Marshal.Alloc (alloca)-import Foreign.Ptr (Ptr, castPtr, freeHaskellFunPtr, nullPtr)-import Foreign.StablePtr (StablePtr, castPtrToStablePtr, castStablePtrToPtr, deRefStablePtr, freeStablePtr, newStablePtr)+import Foreign.Ptr (FunPtr, Ptr, castPtr, nullPtr)+import Foreign.StablePtr (StablePtr, castPtrToStablePtr, deRefStablePtr) import Foreign.Storable (poke, pokeElemOff)+import GHC.Float (float2Double)  data ScalarFunctionResources = ScalarFunctionResources     { scalarFunctionExecPtr :: !DuckDBScalarFunctionFun@@ -74,6 +73,7 @@ data ScalarType     = ScalarTypeBoolean     | ScalarTypeBigInt+    | ScalarTypeUBigInt     | ScalarTypeDouble     | ScalarTypeVarchar @@ -82,6 +82,7 @@     = ScalarNull     | ScalarBoolean !Bool     | ScalarInteger !Int64+    | ScalarUnsigned !Word64     | ScalarDouble !Double     | ScalarText !Text @@ -107,8 +108,8 @@     toScalarValue value = pure (ScalarInteger value)  instance FunctionResult Word where-    scalarReturnType _ = ScalarTypeBigInt-    toScalarValue value = pure (ScalarInteger (fromIntegral value))+    scalarReturnType _ = ScalarTypeUBigInt+    toScalarValue value = pure (ScalarUnsigned (fromIntegral value))  instance FunctionResult Word16 where     scalarReturnType _ = ScalarTypeBigInt@@ -119,8 +120,8 @@     toScalarValue value = pure (ScalarInteger (fromIntegral value))  instance FunctionResult Word64 where-    scalarReturnType _ = ScalarTypeBigInt-    toScalarValue value = pure (ScalarInteger (fromIntegral value))+    scalarReturnType _ = ScalarTypeUBigInt+    toScalarValue value = pure (ScalarUnsigned (fromIntegral value))  instance FunctionResult Double where     scalarReturnType _ = ScalarTypeDouble@@ -128,7 +129,7 @@  instance FunctionResult Float where     scalarReturnType _ = ScalarTypeDouble-    toScalarValue value = pure (ScalarDouble (realToFrac value))+    toScalarValue value = pure (ScalarDouble (float2Double value))  instance FunctionResult Bool where     scalarReturnType _ = ScalarTypeBoolean@@ -227,64 +228,44 @@  -- | Register a Haskell function under the supplied name. createFunction :: forall f. (Function f) => Connection -> Text -> f -> IO ()-createFunction conn name fn = do-    funPtr <- mkScalarFun (scalarFunctionHandler fn)-    resources <- newStablePtr ScalarFunctionResources{scalarFunctionExecPtr = funPtr, scalarFunctionInitPtr = Nothing}-    destroyCb <- mkDeleteCallback releaseFunctionResources-    let release = destroyRegistrationResources funPtr Nothing resources destroyCb-    bracket c_duckdb_create_scalar_function cleanupScalarFunction \scalarFun ->-        (`onException` release) $ do-            TextForeign.withCString name $ \cName ->-                c_duckdb_scalar_function_set_name scalarFun cName-            forM_ (argumentTypes (Proxy :: Proxy f)) \dtype ->-                withLogicalType dtype $ \logical ->-                    c_duckdb_scalar_function_add_parameter scalarFun logical-            withLogicalType (duckTypeForScalar (returnType (Proxy :: Proxy f))) $ \logical ->-                c_duckdb_scalar_function_set_return_type scalarFun logical-            when (isVolatile (Proxy :: Proxy f)) $-                c_duckdb_scalar_function_set_volatile scalarFun-            c_duckdb_scalar_function_set_function scalarFun funPtr-            c_duckdb_scalar_function_set_extra_info scalarFun (castStablePtrToPtr resources) destroyCb-            withConnectionHandle conn \connPtr -> do-                rc <- c_duckdb_register_scalar_function connPtr scalarFun-                if rc == DuckDBSuccess-                    then pure ()-                    else throwIO (functionInvocationError (Text.pack "duckdb-simple: registering function failed"))+createFunction conn name fn =+    registerScalarFunction conn name (Proxy :: Proxy f) \allocate -> do+        scalarFunctionExecPtr <- allocate (mkScalarFun (scalarFunctionHandler fn))+        pure ScalarFunctionResources{scalarFunctionExecPtr, scalarFunctionInitPtr = Nothing}  -- | Register a scalar function with per-worker thread-local state. createFunctionWithState :: forall s f. (Function f) => Connection -> Text -> IO s -> (s -> f) -> IO ()-createFunctionWithState conn name initState mkFn = do-    stateDestroyCb <- mkDeleteCallback releaseStablePtrData-    execPtr <- mkScalarFun (scalarFunctionHandlerWithState mkFn)-    initPtr <- mkScalarInitFun (scalarFunctionInitHandler stateDestroyCb initState)-    resources <--        newStablePtr-            ScalarFunctionResources-                { scalarFunctionExecPtr = execPtr-                , scalarFunctionInitPtr = Just initPtr-                }-    destroyCb <- mkDeleteCallback releaseFunctionResources-    let release = destroyRegistrationResources execPtr (Just initPtr) resources destroyCb >> freeHaskellFunPtr stateDestroyCb-    bracket c_duckdb_create_scalar_function cleanupScalarFunction \scalarFun ->-        (`onException` release) $ do-            TextForeign.withCString name $ \cName ->-                c_duckdb_scalar_function_set_name scalarFun cName-            forM_ (argumentTypes (Proxy :: Proxy f)) \dtype ->-                withLogicalType dtype $ \logical ->-                    c_duckdb_scalar_function_add_parameter scalarFun logical-            withLogicalType (duckTypeForScalar (returnType (Proxy :: Proxy f))) $ \logical ->-                c_duckdb_scalar_function_set_return_type scalarFun logical-            when (isVolatile (Proxy :: Proxy f)) $-                c_duckdb_scalar_function_set_volatile scalarFun-            c_duckdb_scalar_function_set_function scalarFun execPtr-            c_duckdb_scalar_function_set_init scalarFun initPtr-            c_duckdb_scalar_function_set_extra_info scalarFun (castStablePtrToPtr resources) destroyCb-            withConnectionHandle conn \connPtr -> do-                rc <- c_duckdb_register_scalar_function connPtr scalarFun-                if rc == DuckDBSuccess-                    then pure ()-                    else throwIO (functionInvocationError (Text.pack "duckdb-simple: registering function failed"))+createFunctionWithState conn name initState mkFn =+    registerScalarFunction conn name (Proxy :: Proxy f) \allocate -> do+        scalarFunctionExecPtr <- allocate (mkScalarFun (scalarFunctionHandlerWithState mkFn))+        initPtr <- allocate (mkScalarInitFun (scalarFunctionInitHandler initState))+        pure ScalarFunctionResources{scalarFunctionExecPtr, scalarFunctionInitPtr = Just initPtr} +-- | Configure a scalar function and transfer its callbacks to DuckDB.+registerScalarFunction :: (Function f) => Connection -> Text -> Proxy f -> ((forall a. IO (FunPtr a) -> IO (FunPtr a)) -> IO ScalarFunctionResources) -> IO ()+registerScalarFunction conn name proxy acquire = do+    when (Text.null name || Text.any (== '\0') name) $+        throwIO (functionInvocationError "duckdb-simple: invalid scalar function name")+    bracket c_duckdb_create_scalar_function cleanupScalarFunction \scalarFun -> do+        when (scalarFun == nullPtr) $+            throwIO (functionInvocationError "duckdb-simple: failed to allocate scalar function")+        withCallbackResources+            acquire+            (c_duckdb_scalar_function_set_extra_info scalarFun)+            \ScalarFunctionResources{scalarFunctionExecPtr, scalarFunctionInitPtr} -> do+                TextForeign.withCString name $ c_duckdb_scalar_function_set_name scalarFun+                forM_ (argumentTypes proxy) \dtype ->+                    withLogicalType dtype $ c_duckdb_scalar_function_add_parameter scalarFun+                withLogicalType (duckTypeForScalar (returnType proxy)) $ c_duckdb_scalar_function_set_return_type scalarFun+                when (isVolatile proxy) $ c_duckdb_scalar_function_set_volatile scalarFun+                c_duckdb_scalar_function_set_special_handling scalarFun+                c_duckdb_scalar_function_set_function scalarFun scalarFunctionExecPtr+                forM_ scalarFunctionInitPtr $ c_duckdb_scalar_function_set_init scalarFun+                withConnectionHandle conn \connPtr -> do+                    rc <- c_duckdb_register_scalar_function connPtr scalarFun+                    when (rc /= DuckDBSuccess) $+                        throwIO (functionInvocationError "duckdb-simple: registering function failed")+ -- | Drop a previously registered scalar function by issuing a DROP FUNCTION statement. deleteFunction :: Connection -> Text -> IO () deleteFunction conn name =@@ -299,19 +280,7 @@                                     , qualifyIdentifier name                                     ]                     withQueryCString dropQuery \sql ->-                        alloca \resPtr -> do-                            rc <- c_duckdb_query connPtr sql resPtr-                            if rc == DuckDBSuccess-                                then c_duckdb_destroy_result resPtr-                                else do-                                    errMsg <- fetchResultError resPtr-                                    c_duckdb_destroy_result resPtr-                                    throwIO-                                        SQLError-                                            { sqlErrorMessage = errMsg-                                            , sqlErrorType = Nothing-                                            , sqlErrorQuery = Just dropQuery-                                            }+                        withResult conn dropQuery (c_duckdb_query connPtr sql) (const (pure ()))         case outcome of             Right () -> pure ()             Left err@@ -327,35 +296,14 @@         poke ptr scalarFun         c_duckdb_destroy_scalar_function ptr -destroyRegistrationResources ::-    DuckDBScalarFunctionFun ->-    Maybe DuckDBScalarFunctionInitFun ->-    StablePtr ScalarFunctionResources ->-    DuckDBDeleteCallback ->-    IO ()-destroyRegistrationResources funPtr mInitPtr resources destroyCb = do-    freeHaskellFunPtr funPtr-    forM_ mInitPtr freeHaskellFunPtr-    freeStablePtr resources-    freeHaskellFunPtr destroyCb--releaseFunctionResources :: Ptr () -> IO ()-releaseFunctionResources rawPtr =-    when (rawPtr /= nullPtr) $ do-        let stablePtr = castPtrToStablePtr rawPtr :: StablePtr ScalarFunctionResources-        ScalarFunctionResources{scalarFunctionExecPtr, scalarFunctionInitPtr} <- deRefStablePtr stablePtr-        freeHaskellFunPtr scalarFunctionExecPtr-        forM_ scalarFunctionInitPtr freeHaskellFunPtr-        freeStablePtr stablePtr- withLogicalType :: DuckDBType -> (DuckDBLogicalType -> IO a) -> IO a withLogicalType dtype =     bracket         ( do             logical <- c_duckdb_create_logical_type dtype-            when (logical == nullPtr) $-                throwIO $-                    functionInvocationError (Text.pack "duckdb-simple: failed to allocate logical type")+            when (logical == nullPtr)+                $ throwIO+                $ functionInvocationError (Text.pack "duckdb-simple: failed to allocate logical type")             pure logical         )         destroyLogicalType@@ -364,73 +312,56 @@ duckTypeForScalar = \case     ScalarTypeBoolean -> DuckDBTypeBoolean     ScalarTypeBigInt -> DuckDBTypeBigInt+    ScalarTypeUBigInt -> DuckDBTypeUBigInt     ScalarTypeDouble -> DuckDBTypeDouble     ScalarTypeVarchar -> DuckDBTypeVarchar  scalarFunctionHandler :: forall f. (Function f) => f -> DuckDBFunctionInfo -> DuckDBDataChunk -> DuckDBVector -> IO ()-scalarFunctionHandler fn info chunk outVec = do-    result <--        try do-            rawColumnCount <- c_duckdb_data_chunk_get_column_count chunk-            let columnCount = fromIntegral rawColumnCount :: Int-                expected = length (argumentTypes (Proxy :: Proxy f))-            when (columnCount /= expected) $-                throwIO $-                    functionInvocationError $-                        Text.concat-                            [ Text.pack "duckdb-simple: function expected "-                            , Text.pack (show expected)-                            , Text.pack " arguments but received "-                            , Text.pack (show columnCount)-                            ]-            rawRowCount <- c_duckdb_data_chunk_get_size chunk-            let rowCount = fromIntegral rawRowCount :: Int-            readers <- mapM (makeColumnReader chunk) [0 .. expected - 1]-            rows <--                forM [0 .. rowCount - 1] \row ->-                    forM readers \reader ->-                        reader (fromIntegral row)-            results <- mapM (`applyFunction` fn) rows-            writeResults (returnType (Proxy :: Proxy f)) results outVec-            c_duckdb_data_chunk_set_size chunk (fromIntegral rowCount)-    case result of-        Left (err :: SomeException) -> do-            c_duckdb_data_chunk_set_size chunk 0-            let message = Text.pack (displayException err)-            TextForeign.withCString message $ \cMsg ->-                c_duckdb_scalar_function_set_error info cMsg-        Right () -> pure ()+scalarFunctionHandler fn info chunk outVec =+    runCallback (c_duckdb_scalar_function_set_error info) do+        rawColumnCount <- c_duckdb_data_chunk_get_column_count chunk+        let columnCount = fromIntegral rawColumnCount :: Int+            expected = length (argumentTypes (Proxy :: Proxy f))+        when (columnCount /= expected)+            $ throwIO+            $ functionInvocationError+            $ Text.concat+                [ Text.pack "duckdb-simple: function expected "+                , Text.pack (show expected)+                , Text.pack " arguments but received "+                , Text.pack (show columnCount)+                ]+        rawRowCount <- c_duckdb_data_chunk_get_size chunk+        let rowCount = fromIntegral rawRowCount :: Int+        readers <- mapM (makeColumnReader chunk) [0 .. expected - 1]+        rows <-+            forM [0 .. rowCount - 1] \row ->+                forM readers \reader ->+                    reader (fromIntegral row)+        results <- mapM (`applyFunction` fn) rows+        writeResults (returnType (Proxy :: Proxy f)) results outVec  scalarFunctionHandlerWithState :: forall s f. (Function f) => (s -> f) -> DuckDBFunctionInfo -> DuckDBDataChunk -> DuckDBVector -> IO ()-scalarFunctionHandlerWithState mkFn info chunk outVec = do-    statePtr <- c_duckdb_scalar_function_get_state info-    if statePtr == nullPtr-        then-            TextForeign.withCString (Text.pack "duckdb-simple: scalar function state was not initialised") $-                c_duckdb_scalar_function_set_error info-        else do-            state <- deRefStablePtr (castPtrToStablePtr statePtr :: StablePtr s)-            scalarFunctionHandler (mkFn state) info chunk outVec+scalarFunctionHandlerWithState mkFn info chunk outVec =+    runCallback (c_duckdb_scalar_function_set_error info) do+        statePtr <- c_duckdb_scalar_function_get_state info+        when (statePtr == nullPtr) $+            throwIO (functionInvocationError "duckdb-simple: scalar function state was not initialised")+        state <- deRefStablePtr (castPtrToStablePtr statePtr :: StablePtr s)+        scalarFunctionHandler (mkFn state) info chunk outVec -scalarFunctionInitHandler :: forall s. DuckDBDeleteCallback -> IO s -> DuckDBInitInfo -> IO ()-scalarFunctionInitHandler destroyCb initState info = do-    outcome <- try initState-    case outcome of-        Left (err :: SomeException) ->-            TextForeign.withCString (Text.pack (displayException err)) $-                c_duckdb_scalar_function_init_set_error info-        Right state -> do-            stable <- newStablePtr state-            c_duckdb_scalar_function_init_set_state info (castStablePtrToPtr stable) destroyCb+scalarFunctionInitHandler :: IO s -> DuckDBInitInfo -> IO ()+scalarFunctionInitHandler initState info =+    runCallback (c_duckdb_scalar_function_init_set_error info) do+        state <- initState+        transferCallbackState (c_duckdb_scalar_function_init_set_state info) state  type ColumnReader = DuckDBIdx -> IO Field  makeColumnReader :: DuckDBDataChunk -> Int -> IO ColumnReader makeColumnReader chunk columnIndex = do     vector <- c_duckdb_data_chunk_get_vector chunk (fromIntegral columnIndex)-    logical <- c_duckdb_vector_get_column_type vector-    dtype <- c_duckdb_get_type_id logical-    destroyLogicalType logical+    dtype <- bracket (c_duckdb_vector_get_column_type vector) destroyLogicalType c_duckdb_get_type_id     dataPtr <- c_duckdb_vector_get_data vector     validity <- c_duckdb_vector_get_validity vector     let name = Text.pack ("arg" <> show columnIndex)@@ -459,6 +390,9 @@             (ScalarTypeBigInt, ScalarInteger intval) -> do                 markValid validityPtr idx                 pokeElemOff (castPtr dataPtr :: Ptr Int64) idx intval+            (ScalarTypeUBigInt, ScalarUnsigned intval) -> do+                markValid validityPtr idx+                pokeElemOff (castPtr dataPtr :: Ptr Word64) idx intval             (ScalarTypeDouble, ScalarDouble dbl) -> do                 markValid validityPtr idx                 pokeElemOff (castPtr dataPtr :: Ptr Double) idx dbl@@ -467,9 +401,9 @@                 TextForeign.withCStringLen txt \(ptr, len) ->                     c_duckdb_vector_assign_string_element_len outVec (fromIntegral idx) ptr (fromIntegral len)             _ ->-                throwIO $-                    functionInvocationError $-                        Text.pack "duckdb-simple: result type mismatch when materialising scalar function output"+                throwIO+                    $ functionInvocationError+                    $ Text.pack "duckdb-simple: result type mismatch when materialising scalar function output"  markInvalid :: Ptr Word64 -> Int -> IO () markInvalid validity idx@@ -504,13 +438,6 @@         , sqlErrorType = Nothing         , sqlErrorQuery = Nothing         }--fetchResultError :: Ptr DuckDBResult -> IO Text-fetchResultError resPtr = do-    msgPtr <- c_duckdb_result_error resPtr-    if msgPtr == nullPtr-        then pure (Text.pack "duckdb-simple: DROP FUNCTION failed")-        else Text.pack <$> peekCString msgPtr  qualifyIdentifier :: Text -> Text qualifyIdentifier rawName =

@@ -129,6 +129,7 @@     UnionValue (..),  ) import Database.DuckDB.Simple.Ok (Ok (..))+import Database.DuckDB.Simple.Time (Unbounded (..)) import Database.DuckDB.Simple.ToField (DuckDBColumnType (..), ToField (..))  --------------------------------------------------------------------------------@@ -231,7 +232,7 @@     duckLogicalType _ = LogicalTypeScalar DuckDBTypeBlob  instance DuckValue Day where-    duckToField = FieldDate+    duckToField = FieldDate . Finite     duckLogicalType _ = LogicalTypeScalar DuckDBTypeDate  instance DuckValue TimeOfDay where@@ -239,10 +240,22 @@     duckLogicalType _ = LogicalTypeScalar DuckDBTypeTime  instance DuckValue LocalTime where-    duckToField = FieldTimestamp+    duckToField = FieldTimestamp . Finite     duckLogicalType _ = LogicalTypeScalar DuckDBTypeTimestamp  instance DuckValue UTCTime where+    duckToField = FieldTimestampTZ . Finite+    duckLogicalType _ = LogicalTypeScalar DuckDBTypeTimestampTz++instance DuckValue (Unbounded Day) where+    duckToField = FieldDate+    duckLogicalType _ = LogicalTypeScalar DuckDBTypeDate++instance DuckValue (Unbounded LocalTime) where+    duckToField = FieldTimestamp+    duckLogicalType _ = LogicalTypeScalar DuckDBTypeTimestamp++instance DuckValue (Unbounded UTCTime) where     duckToField = FieldTimestampTZ     duckLogicalType _ = LogicalTypeScalar DuckDBTypeTimestampTz @@ -567,10 +580,25 @@         | idx /= 0 = Left ("duckdb-simple: union tag mismatch (expected 0, got " <> show idx <> ")")         | otherwise =             case payload of-                FieldNull -> pure (M1 (gStructNull (Proxy :: Proxy (f p))))-                FieldStruct structVal -> M1 <$> gStructDecodeStruct (Proxy :: Proxy (f p)) structVal+                FieldNull -> M1 . fst <$> gStructDecodeList (Proxy :: Proxy (f p)) []+                FieldStruct structVal -> do+                    ordered <- orderStructFields (gStructTypes (Proxy :: Proxy (f p))) structVal+                    M1 <$> gStructDecodeStruct (Proxy :: Proxy (f p)) ordered                 other -> Left ("duckdb-simple: expected STRUCT payload for union member, got " <> show other) +-- | Match decoded fields to their selector names before product decoding.+orderStructFields :: [FieldComponent LogicalTypeRep] -> StructValue FieldValue -> Either String (StructValue FieldValue)+orderStructFields components sv = do+    let names = resolveNames (zip [0 ..] (map fcName components))+        fields = elems (structValueFields sv)+        byName = Map.fromList [(structFieldName field, field) | field <- fields]+    unless (length fields <= length names) $+        Left "duckdb-simple: extra fields when decoding struct"+    unless (length fields == length names && Map.size byName == length names) $+        Left "duckdb-simple: struct field names or count mismatch"+    ordered <- traverse (\name -> maybe (Left ("duckdb-simple: missing struct field " <> Text.unpack name)) Right (Map.lookup name byName)) names+    pure sv{structValueFields = listArray (0, length ordered - 1) ordered}+ -------------------------------------------------------------------------------- -- GStructDecode: inverse of GStruct for decoding @@ -578,13 +606,6 @@ class GStructDecode f where     gStructDecodeStruct :: Proxy (f p) -> StructValue FieldValue -> Either String (f p) -    {- | Construct a null/empty value for a struct type.-    This is only valid for U1 (empty structs) and their compositions.-    For selectors with actual values, this should never be called in practice-    as nullary constructors are represented as U1.-    -}-    gStructNull :: Proxy (f p) -> f p-     {- | Consume a prefix of fields from left to right while decoding, returning     the reconstructed value and any remaining fields.     -}@@ -595,7 +616,6 @@         if null (elems (structValueFields structVal))             then Right U1             else Left ("duckdb-simple: expected empty struct, but got " <> show (length (elems (structValueFields structVal))) <> " field(s)")-    gStructNull _ = U1     gStructDecodeList _ xs = Right (U1, xs)  instance (GStructDecode a, GStructDecode b) => GStructDecode (a :*: b) where@@ -606,7 +626,6 @@         unless (null rest') $             Left ("duckdb-simple: extra " <> show (length rest') <> " field(s) when decoding struct (too many fields provided)")         pure (leftVal :*: rightVal)-    gStructNull _ = gStructNull (Proxy :: Proxy (a p)) :*: gStructNull (Proxy :: Proxy (b p))     gStructDecodeList _ xs = do         (leftVal, rest) <- gStructDecodeList (Proxy :: Proxy (a p)) xs         (rightVal, rest') <- gStructDecodeList (Proxy :: Proxy (b p)) rest@@ -619,15 +638,6 @@             [] -> Left "duckdb-simple: missing struct field (expected 1, got 0)"             xs -> Left ("duckdb-simple: expected single field struct, but got " <> show (length xs) <> " fields") -    -- IMPOSSIBLE: This should never be called in practice because nullary constructors-    -- are represented as U1, not as selectors with actual field values. A selector (M1 S)-    -- represents a record field that must contain a value, so there's no sensible way to-    -- construct a "null" instance. This method is only needed to satisfy the GStructDecode-    -- typeclass constraint, but in the actual decoding path (gSumDecode), nullary constructors-    -- always take the FieldNull case which constructs U1 directly, never calling gStructNull-    -- on a selector. If this error is ever reached, it indicates a bug in the generic-    -- traversal logic.-    gStructNull _ = error "duckdb-simple: impossible - gStructNull called on selector"     gStructDecodeList _ [] = Left "duckdb-simple: missing struct field (expected field but list is empty)"     gStructDecodeList _ (fv : rest) = do         val <- duckFromField fv@@ -635,7 +645,6 @@  instance (GStructDecode f) => GStructDecode (M1 C c f) where     gStructDecodeStruct _ structVal = M1 <$> gStructDecodeStruct (Proxy :: Proxy (f p)) structVal-    gStructNull _ = M1 (gStructNull (Proxy :: Proxy (f p)))     gStructDecodeList _ values = do         (inner, rest) <- gStructDecodeList (Proxy :: Proxy (f p)) values         pure (M1 inner, rest)@@ -657,13 +666,27 @@  instance (GStruct f, GStructDecode f) => GFromField' 'False (M1 D meta (M1 C c f)) where     gFromField' _ = \case-        FieldNull -> pure (M1 (M1 (gStructNull (Proxy :: Proxy (f p)))))-        FieldStruct sv -> M1 . M1 <$> gStructDecodeStruct (Proxy :: Proxy (f p)) sv+        FieldNull -> M1 . M1 . fst <$> gStructDecodeList (Proxy :: Proxy (f p)) []+        FieldStruct sv -> do+            ordered <- orderStructFields (gStructTypes (Proxy :: Proxy (f p))) sv+            M1 . M1 <$> gStructDecodeStruct (Proxy :: Proxy (f p)) ordered         other -> Left ("duckdb-simple: expected STRUCT value for product type, got " <> show other)  instance (GSum f) => GFromField' 'True (M1 D meta f) where     gFromField' _ = \case-        FieldUnion uv -> M1 <$> gSumDecode (fromIntegral (unionValueIndex uv)) (unionValuePayload uv)+        FieldUnion uv -> do+            let actualMembers = elems (unionValueMembers uv)+                actualIndex = fromIntegral (unionValueIndex uv)+            unless (actualIndex < length actualMembers) $+                Left "duckdb-simple: union tag out of range"+            unless (unionMemberName (actualMembers !! actualIndex) == unionValueLabel uv) $+                Left "duckdb-simple: union tag and member name mismatch"+            let members = gSumMembers (Proxy :: Proxy (f p))+                indexed = zip [0 ..] members+                matching = [idx | (idx, member) <- indexed, unionMemberName member == unionValueLabel uv]+            case matching of+                [idx] -> M1 <$> gSumDecode idx (unionValuePayload uv)+                _ -> Left "duckdb-simple: unknown union member name"         other -> Left ("duckdb-simple: expected UNION value for sum type, got " <> show other)  instance GFromField' 'False (M1 D meta U1) where

@@ -1,6 +1,9 @@ {-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE MagicHash #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE StrictData #-}+{-# OPTIONS_GHC -Wno-deprecations #-}  {- | Module      : Database.DuckDB.Simple.Internal@@ -19,6 +22,7 @@     StatementState (..),     StatementStreamState (..),     StatementStream (..),+    ResultMode (..),     StatementStreamColumn (..),     StatementStreamChunk (..),     StatementStreamChunkVector (..),@@ -28,25 +32,36 @@     -- * Helpers     connectionClosedError,     statementClosedError,+    keepAlive,     withDatabaseHandle,     withConnectionHandle,     withStatementHandle,     withQueryCString,+    peekUtf8CString,+    withResult,+    runInterruptibleQuery,+    executePreparedResult,+    fetchResultChunk,+    destroyDataChunk,+    fetchResultError,+    throwResultError,+    mkExecuteError,     withClientContext,     destroyClientContext,     destroyValue,     destroyLogicalType,     throwRegistrationError,-    releaseStablePtrData,-    mkDeleteCallback, ) where -import Control.Exception (Exception, bracket, throwIO)+import Control.Concurrent (forkIO, newEmptyMVar, putMVar, readMVar, threadDelay, tryReadMVar)+import Control.Exception (Exception, SomeException, bracket, bracket_, mask, mask_, onException, throwIO, try, uninterruptibleMask_) import Control.Monad (when)+import qualified Data.ByteString as BS import Data.IORef (IORef, readIORef) import Data.String (IsString (..)) import Data.Text (Text) import qualified Data.Text as Text+import qualified Data.Text.Encoding as TextEncoding import qualified Data.Text.Foreign as TextForeign import Data.Word (Word64) import Database.DuckDB.FFI (@@ -54,24 +69,36 @@     DuckDBConnection,     DuckDBDataChunk,     DuckDBDatabase,-    DuckDBDeleteCallback,     DuckDBErrorType,     DuckDBLogicalType,     DuckDBPreparedStatement,     DuckDBResult,+    DuckDBState,     DuckDBType,     DuckDBValue,     DuckDBVector,     c_duckdb_connection_get_client_context,     c_duckdb_destroy_client_context,+    c_duckdb_destroy_data_chunk,     c_duckdb_destroy_logical_type,+    c_duckdb_destroy_result,     c_duckdb_destroy_value,+    c_duckdb_execute_prepared,+    c_duckdb_fetch_chunk,+    c_duckdb_interrupt,+    c_duckdb_result_error,+    c_duckdb_result_error_type,+    pattern DuckDBErrorInvalid,+    pattern DuckDBSuccess,  )+import Database.DuckDB.FFI.Deprecated (c_duckdb_execute_prepared_streaming) import Foreign.C.String (CString) import Foreign.Marshal.Alloc (alloca)+import Foreign.Marshal.Utils (fillBytes) import Foreign.Ptr (Ptr, nullPtr)-import Foreign.StablePtr (StablePtr, castPtrToStablePtr, freeStablePtr)-import Foreign.Storable (peek, poke)+import Foreign.Storable (peek, poke, sizeOf)+import GHC.Exts (keepAlive#)+import GHC.IO (IO (..))  -- | Represents a textual SQL query with UTF-8 encoding semantics. newtype Query = Query@@ -115,6 +142,7 @@ -- | Streaming execution state for prepared statements. data StatementStreamState     = StatementStreamIdle+    | StatementStreamExhausted     | StatementStreamActive !StatementStream  -- | Streaming cursor backing an active result set.@@ -122,8 +150,12 @@     { statementStreamResult :: Ptr DuckDBResult     , statementStreamColumns :: [StatementStreamColumn]     , statementStreamChunk :: Maybe StatementStreamChunk+    , statementStreamMode :: ResultMode     } +-- | Select native execution. DuckDB can materialize a streaming request.+data ResultMode = MaterializedResult | StreamingResult+ -- | Metadata describing a result column surfaced through streaming. data StatementStreamColumn = StatementStreamColumn     { statementStreamColumnIndex :: Int@@ -185,40 +217,157 @@  -- | Provide a UTF-8 encoded C string view of the query text. withQueryCString :: Query -> (CString -> IO a) -> IO a-withQueryCString (Query txt) = TextForeign.withCString txt+withQueryCString query@(Query txt) action+    | Text.any (== '\0') txt =+        throwIO (SQLError (Text.pack "duckdb-simple: SQL contains NUL") Nothing (Just query))+    | otherwise = TextForeign.withCString txt action +-- | Copy a NUL-terminated UTF-8 string from DuckDB.+peekUtf8CString :: CString -> IO Text+peekUtf8CString ptr = TextEncoding.decodeUtf8 <$> BS.packCString ptr++-- | Execute a query and destroy its result after success or failure.+withResult :: Connection -> Query -> (Ptr DuckDBResult -> IO DuckDBState) -> (Ptr DuckDBResult -> IO a) -> IO a+withResult conn queryText executeResult action =+    alloca $ \resPtr ->+        bracket_+            (fillBytes resPtr 0 (sizeOf (undefined :: DuckDBResult)))+            (c_duckdb_destroy_result resPtr)+            $ do+                rc <- runInterruptibleQuery conn (executeResult resPtr)+                when (rc /= DuckDBSuccess) $ do+                    (message, errorType) <- fetchResultError resPtr+                    throwIO (mkExecuteError queryText message errorType)+                action resPtr++-- | Execute a prepared statement with the requested native result mode.+executePreparedResult :: ResultMode -> DuckDBPreparedStatement -> Ptr DuckDBResult -> IO DuckDBState+executePreparedResult MaterializedResult = c_duckdb_execute_prepared+executePreparedResult StreamingResult = c_duckdb_execute_prepared_streaming++{- | Fetch an owned chunk. The caller must remain masked until it records ownership.+  Streaming fetches can execute SQL. Keep their output in a caller-owned slot+  so cancellation can destroy a chunk that the worker has already fetched.+-}+fetchResultChunk :: ResultMode -> Connection -> Ptr DuckDBResult -> IO DuckDBDataChunk+fetchResultChunk MaterializedResult conn result =+    withConnectionHandle conn $ \_ -> c_duckdb_fetch_chunk result+fetchResultChunk StreamingResult conn result =+    alloca $ \chunkPtr -> mask_ $ do+        poke chunkPtr nullPtr+        let fetch = do+                chunk <- c_duckdb_fetch_chunk result+                poke chunkPtr chunk+                pure DuckDBSuccess+        (runInterruptibleQuery conn fetch >> peek chunkPtr)+            `onException` c_duckdb_destroy_data_chunk chunkPtr++-- | Destroy an owned native data chunk. A null chunk is allowed.+destroyDataChunk :: DuckDBDataChunk -> IO ()+destroyDataChunk chunk = alloca $ \ptr -> poke ptr chunk >> c_duckdb_destroy_data_chunk ptr++{- | Run a native query while the caller can receive asynchronous exceptions.+  Prompt cancellation requires the threaded RTS. Cleanup interrupts DuckDB and+  waits for the worker before the caller can release native storage.+-}+runInterruptibleQuery :: Connection -> IO DuckDBState -> IO DuckDBState+runInterruptibleQuery conn action =+    withConnectionHandle conn $ \handle ->+        mask $ \restore -> do+            done <- newEmptyMVar+            _ <- forkIO $ do+                outcome <- try action :: IO (Either SomeException DuckDBState)+                putMVar done outcome+            let cancelAndWait = do+                    finished <- tryReadMVar done+                    case finished of+                        Just _ -> pure ()+                        Nothing -> do+                            -- DuckDB clears the interrupt flag at query entry.+                            -- Repeat the interrupt to cover cancellation before entry.+                            c_duckdb_interrupt handle+                            threadDelay interruptRetryDelayMicros+                            cancelAndWait+            -- A second exception must not release storage still used by the worker.+            -- Native code and Haskell callbacks must return before cleanup can finish.+            outcome <- restore (readMVar done) `onException` uninterruptibleMask_ cancelAndWait+            either throwIO pure outcome++{- | Retry cancellation at most 100 times per second while native work finishes.+DuckDB can clear an earlier interrupt at query entry. This interval avoids+busy-spinning and can add up to one interval to completion detection.+-}+interruptRetryDelayMicros :: Int+interruptRetryDelayMicros = 10 * 1000++-- | Copy a result error while its native result remains alive.+fetchResultError :: Ptr DuckDBResult -> IO (Text, Maybe DuckDBErrorType)+fetchResultError resultPtr = do+    msgPtr <- c_duckdb_result_error resultPtr+    message <-+        if msgPtr == nullPtr+            then pure (Text.pack "duckdb-simple: query failed")+            else peekUtf8CString msgPtr+    errorType <- c_duckdb_result_error_type resultPtr+    pure (message, if errorType == DuckDBErrorInvalid then Nothing else Just errorType)++-- | Attach the query and native error category to an execution error.+mkExecuteError :: Query -> Text -> Maybe DuckDBErrorType -> SQLError+mkExecuteError queryText message errorType =+    SQLError+        { sqlErrorMessage = message+        , sqlErrorType = errorType+        , sqlErrorQuery = Just queryText+        }++-- | Report a fetch failure before treating a null chunk as end of input.+throwResultError :: Query -> Ptr DuckDBResult -> IO ()+throwResultError queryText resPtr = do+    errorPtr <- c_duckdb_result_error resPtr+    when (errorPtr /= nullPtr) $ do+        (message, errorType) <- fetchResultError resPtr+        throwIO (mkExecuteError queryText message errorType)++-- | Keep the weak-finalizer owner alive until the native operation returns.+keepAlive :: a -> IO b -> IO b+keepAlive owner (IO action) = IO (\state -> keepAlive# owner state action)+ -- | Internal helper for safely accessing the underlying prepared statement. withStatementHandle :: Statement -> (DuckDBPreparedStatement -> IO a) -> IO a-withStatementHandle stmt@Statement{statementState} action = do-    state <- readIORef statementState-    case state of-        StatementClosed -> throwIO (statementClosedError stmt)-        StatementOpen{statementHandle} -> action statementHandle+withStatementHandle stmt@Statement{statementState, statementConnection} action =+    keepAlive stmt $+        withConnectionHandle statementConnection $ \_ -> do+            state <- readIORef statementState+            case state of+                StatementClosed -> throwIO (statementClosedError stmt)+                StatementOpen{statementHandle} -> action statementHandle  -- | Internal helper for safely accessing the underlying connection handle. withConnectionHandle :: Connection -> (DuckDBConnection -> IO a) -> IO a-withConnectionHandle Connection{connectionState} action = do-    state <- readIORef connectionState-    case state of-        ConnectionClosed -> throwIO connectionClosedError-        ConnectionOpen{connectionHandle} -> action connectionHandle+withConnectionHandle conn@Connection{connectionState} action =+    keepAlive conn $ do+        state <- readIORef connectionState+        case state of+            ConnectionClosed -> throwIO connectionClosedError+            ConnectionOpen{connectionHandle} -> action connectionHandle  -- | Internal helper for safely accessing the underlying database handle. withDatabaseHandle :: Connection -> (DuckDBDatabase -> IO a) -> IO a-withDatabaseHandle Connection{connectionState} action = do-    state <- readIORef connectionState-    case state of-        ConnectionClosed -> throwIO connectionClosedError-        ConnectionOpen{connectionDatabase} -> action connectionDatabase+withDatabaseHandle conn@Connection{connectionState} action =+    keepAlive conn $ do+        state <- readIORef connectionState+        case state of+            ConnectionClosed -> throwIO connectionClosedError+            ConnectionOpen{connectionDatabase} -> action connectionDatabase  -- | Acquire the client context for the connection, destroying it after the action. withClientContext :: Connection -> (DuckDBClientContext -> IO a) -> IO a withClientContext conn action =     withConnectionHandle conn $ \connPtr ->-        alloca $ \ctxPtr -> do-            c_duckdb_connection_get_client_context connPtr ctxPtr-            ctx <- peek ctxPtr-            bracket (pure ctx) destroyClientContext action+        bracket+            (alloca $ \ctxPtr -> c_duckdb_connection_get_client_context connPtr ctxPtr >> peek ctxPtr)+            destroyClientContext+            action  -- | Destroy a client context handle. destroyClientContext :: DuckDBClientContext -> IO ()@@ -244,13 +393,3 @@             , sqlErrorType = Nothing             , sqlErrorQuery = Nothing             }---- | Free a stable pointer stored behind a raw @Ptr ()@.-releaseStablePtrData :: Ptr () -> IO ()-releaseStablePtrData rawPtr =-    when (rawPtr /= nullPtr) $-        freeStablePtr (castPtrToStablePtr rawPtr :: StablePtr ())---- | Create a DuckDB delete callback from a Haskell function.-foreign import ccall "wrapper"-    mkDeleteCallback :: (Ptr () -> IO ()) -> IO DuckDBDeleteCallback

@@ -10,7 +10,8 @@     registerLogStorage, ) where -import Control.Exception (bracket, onException)+import Control.Exception (bracket, mask_)+import Control.Monad (when) import Data.Ratio ((%)) import Data.Text (Text) import qualified Data.Text as Text@@ -18,11 +19,11 @@ import Data.Time.Clock (UTCTime) import Data.Time.Clock.POSIX (posixSecondsToUTCTime) import Database.DuckDB.FFI-import Database.DuckDB.Simple.Internal (Connection, mkDeleteCallback, throwRegistrationError, withDatabaseHandle)-import Foreign.C.String (CString, peekCString)+import Database.DuckDB.Simple.Callback (ignoreCallbackExceptions, withCallbackResources)+import Database.DuckDB.Simple.Internal (Connection, peekUtf8CString, throwRegistrationError, withDatabaseHandle)+import Foreign.C.String (CString) import Foreign.Marshal.Alloc (alloca)-import Foreign.Ptr (Ptr, freeHaskellFunPtr, nullPtr)-import Foreign.StablePtr (StablePtr, castPtrToStablePtr, castStablePtrToPtr, deRefStablePtr, freeStablePtr, newStablePtr)+import Foreign.Ptr (Ptr, nullFunPtr, nullPtr) import Foreign.Storable (peek, poke)  -- | A single log event delivered through DuckDB's log-storage callback.@@ -37,21 +38,22 @@ -- | Register a custom log storage callback on the database behind a connection. registerLogStorage :: Connection -> Text -> (LogEntry -> IO ()) -> IO () registerLogStorage conn name callback = do-    writeCb <- mkWriteLogEntryCallback (logStorageHandler callback)-    callbackStable <- newStablePtr writeCb-    deleteCb <- mkDeleteCallback releaseWriteLogCallback-    let release = freeHaskellFunPtr writeCb >> freeStablePtr callbackStable >> freeHaskellFunPtr deleteCb-    bracket c_duckdb_create_log_storage destroyLogStorage \storage ->-        (`onException` release) $ do-            TextForeign.withCString name \cName ->-                c_duckdb_log_storage_set_name storage cName-            c_duckdb_log_storage_set_write_log_entry storage writeCb-            c_duckdb_log_storage_set_extra_data storage (castStablePtrToPtr callbackStable) deleteCb-            withDatabaseHandle conn \db -> do-                rc <- c_duckdb_register_log_storage db storage-                if rc == DuckDBSuccess-                    then pure ()-                    else throwRegistrationError "register log storage"+    when (Text.null name || Text.any (== '\0') name) $+        throwRegistrationError "invalid log storage name"+    bracket c_duckdb_create_log_storage destroyLogStorage \storage -> do+        when (storage == nullPtr) $ throwRegistrationError "allocate log storage"+        withCallbackResources+            (\allocate -> allocate (mkWriteLogEntryCallback (logStorageHandler callback)))+            (c_duckdb_log_storage_set_extra_data storage)+            \writeCb -> do+                TextForeign.withCString name $ c_duckdb_log_storage_set_name storage+                c_duckdb_log_storage_set_write_log_entry storage writeCb+                withDatabaseHandle conn \db -> mask_ do+                    rc <- c_duckdb_register_log_storage db storage+                    -- DuckDB consumes extra data on duplicate-name failure.+                    -- Clear the wrapper to prevent a second destruction.+                    c_duckdb_log_storage_set_extra_data storage nullPtr nullFunPtr+                    when (rc /= DuckDBSuccess) $ throwRegistrationError "register log storage"  logStorageHandler ::     (LogEntry -> IO ()) ->@@ -61,14 +63,15 @@     CString ->     CString ->     IO ()-logStorageHandler callback _ timestampPtr levelPtr logTypePtr messagePtr = do-    entry <- do-        logEntryTimestamp <- readTimestamp timestampPtr-        logEntryLevel <- readCStringText levelPtr-        logEntryType <- readCStringText logTypePtr-        logEntryMessage <- readCStringText messagePtr-        pure LogEntry{logEntryTimestamp, logEntryLevel, logEntryType, logEntryMessage}-    callback entry+logStorageHandler callback _ timestampPtr levelPtr logTypePtr messagePtr =+    ignoreCallbackExceptions do+        entry <- do+            logEntryTimestamp <- readTimestamp timestampPtr+            logEntryLevel <- readCStringText levelPtr+            logEntryType <- readCStringText logTypePtr+            logEntryMessage <- readCStringText messagePtr+            pure LogEntry{logEntryTimestamp, logEntryLevel, logEntryType, logEntryMessage}+        callback entry  readTimestamp :: Ptr DuckDBTimestamp -> IO (Maybe UTCTime) readTimestamp ptr@@ -80,17 +83,7 @@ readCStringText :: CString -> IO Text readCStringText ptr     | ptr == nullPtr = pure Text.empty-    | otherwise = Text.pack <$> peekCString ptr--releaseWriteLogCallback :: Ptr () -> IO ()-releaseWriteLogCallback rawPtr =-    if rawPtr == nullPtr-        then pure ()-        else do-            let stablePtr = castPtrToStablePtr rawPtr :: StablePtr DuckDBLoggerWriteLogEntryFun-            callback <- deRefStablePtr stablePtr-            freeHaskellFunPtr callback-            freeStablePtr stablePtr+    | otherwise = peekUtf8CString ptr  destroyLogStorage :: DuckDBLogStorage -> IO () destroyLogStorage storage =

@@ -23,12 +23,14 @@ import Control.Exception (bracket, throwIO) import Control.Monad (forM, when) import Data.Array (Array, elems, listArray)+import qualified Data.ByteString as BS import Data.Map.Strict (Map) import Data.Text (Text) import qualified Data.Text as Text+import qualified Data.Text.Encoding as TextEncoding import Data.Word (Word16, Word64, Word8) import Database.DuckDB.FFI-import Foreign.C.String (peekCString, withCString)+import Foreign.C.String (CString) import Foreign.Marshal.Alloc (alloca) import Foreign.Marshal.Array (withArray) import Foreign.Marshal.Utils (withMany)@@ -103,14 +105,12 @@             childCount <- word64ToInt (Text.pack "struct child count") childCountRaw             fields <-                 forM [0 .. childCount - 1] \idx -> do-                    namePtr <- c_duckdb_struct_type_child_name logical (fromIntegral idx)-                    when (namePtr == nullPtr) $-                        throwIO (userError "duckdb-simple: struct child name is null")-                    name <- Text.pack <$> peekCString namePtr-                    c_duckdb_free (castPtr namePtr)-                    childLogical <- c_duckdb_struct_type_child_type logical (fromIntegral idx)+                    name <- bracket (c_duckdb_struct_type_child_name logical (fromIntegral idx)) (c_duckdb_free . castPtr) $ \ptr -> do+                        when (ptr == nullPtr) $+                            throwIO (userError "duckdb-simple: struct child name is null")+                        TextEncoding.decodeUtf8 <$> BS.packCString ptr                     childRep <--                        bracket (pure childLogical) destroyLogicalType logicalTypeToRep+                        bracket (c_duckdb_struct_type_child_type logical (fromIntegral idx)) destroyLogicalType logicalTypeToRep                     pure StructField{structFieldName = name, structFieldValue = childRep}             pure $                 LogicalTypeStruct@@ -123,14 +123,12 @@             memberCount <- word64ToInt (Text.pack "union member count") memberCountRaw             members <-                 forM [0 .. memberCount - 1] \idx -> do-                    namePtr <- c_duckdb_union_type_member_name logical (fromIntegral idx)-                    when (namePtr == nullPtr) $-                        throwIO (userError "duckdb-simple: union member name is null")-                    name <- Text.pack <$> peekCString namePtr-                    c_duckdb_free (castPtr namePtr)-                    memberLogical <- c_duckdb_union_type_member_type logical (fromIntegral idx)+                    name <- bracket (c_duckdb_union_type_member_name logical (fromIntegral idx)) (c_duckdb_free . castPtr) $ \ptr -> do+                        when (ptr == nullPtr) $+                            throwIO (userError "duckdb-simple: union member name is null")+                        TextEncoding.decodeUtf8 <$> BS.packCString ptr                     memberRep <--                        bracket (pure memberLogical) destroyLogicalType logicalTypeToRep+                        bracket (c_duckdb_union_type_member_type logical (fromIntegral idx)) destroyLogicalType logicalTypeToRep                     pure UnionMemberType{unionMemberName = name, unionMemberType = memberRep}             pure $                 LogicalTypeUnion@@ -139,19 +137,15 @@                         else listArray (0, memberCount - 1) members                     )         DuckDBTypeList -> do-            childLogical <- c_duckdb_list_type_child_type logical-            childRep <- bracket (pure childLogical) destroyLogicalType logicalTypeToRep+            childRep <- bracket (c_duckdb_list_type_child_type logical) destroyLogicalType logicalTypeToRep             pure (LogicalTypeList childRep)         DuckDBTypeArray -> do-            childLogical <- c_duckdb_array_type_child_type logical-            childRep <- bracket (pure childLogical) destroyLogicalType logicalTypeToRep+            childRep <- bracket (c_duckdb_array_type_child_type logical) destroyLogicalType logicalTypeToRep             size <- c_duckdb_array_type_array_size logical             pure (LogicalTypeArray childRep size)         DuckDBTypeMap -> do-            keyLogical <- c_duckdb_map_type_key_type logical-            valueLogical <- c_duckdb_map_type_value_type logical-            keyRep <- bracket (pure keyLogical) destroyLogicalType logicalTypeToRep-            valueRep <- bracket (pure valueLogical) destroyLogicalType logicalTypeToRep+            keyRep <- bracket (c_duckdb_map_type_key_type logical) destroyLogicalType logicalTypeToRep+            valueRep <- bracket (c_duckdb_map_type_value_type logical) destroyLogicalType logicalTypeToRep             pure (LogicalTypeMap keyRep valueRep)         DuckDBTypeDecimal -> do             width <- c_duckdb_decimal_width logical@@ -162,11 +156,10 @@             let count = fromIntegral dictSize :: Int             entries <-                 forM [0 .. count - 1] \idx -> do-                    entryPtr <- c_duckdb_enum_dictionary_value logical (fromIntegral idx)-                    when (entryPtr == nullPtr) $-                        throwIO (userError "duckdb-simple: enum dictionary value is null")-                    entry <- Text.pack <$> peekCString entryPtr-                    c_duckdb_free (castPtr entryPtr)+                    entry <- bracket (c_duckdb_enum_dictionary_value logical (fromIntegral idx)) (c_duckdb_free . castPtr) $ \ptr -> do+                        when (ptr == nullPtr) $+                            throwIO (userError "duckdb-simple: enum dictionary value is null")+                        TextEncoding.decodeUtf8 <$> BS.packCString ptr                     pure entry             pure $                 LogicalTypeEnum@@ -179,64 +172,52 @@  -- | Materialize a DuckDB logical type handle from a @LogicalTypeRep@ tree. logicalTypeFromRep :: LogicalTypeRep -> IO DuckDBLogicalType-logicalTypeFromRep = \case-    LogicalTypeScalar dtype ->-        c_duckdb_create_logical_type dtype-    LogicalTypeDecimal width scale ->-        c_duckdb_create_decimal_type width scale-    LogicalTypeList elemRep ->-        bracket (logicalTypeFromRep elemRep) destroyLogicalType c_duckdb_create_list_type-    LogicalTypeArray elemRep size ->-        bracket (logicalTypeFromRep elemRep) destroyLogicalType $-            flip c_duckdb_create_array_type size-    LogicalTypeMap keyRep valueRep ->-        bracket (logicalTypeFromRep keyRep) destroyLogicalType \keyType ->-            bracket (logicalTypeFromRep valueRep) destroyLogicalType $-                c_duckdb_create_map_type keyType-    LogicalTypeStruct fieldArray -> do-        let fields = elems fieldArray-            count = length fields-        if count == 0-            then withMany withCString [] \namePtrs ->-                withArray namePtrs \nameArray ->-                    withArray [] \typeArray ->-                        c_duckdb_create_struct_type typeArray nameArray 0-            else do-                childTypes <- mapM (logicalTypeFromRep . structFieldValue) fields-                let names = map (Text.unpack . structFieldName) fields-                result <--                    withMany withCString names \namePtrs ->-                        withArray namePtrs \nameArray ->-                            withArray childTypes \typeArray ->-                                c_duckdb_create_struct_type typeArray nameArray (fromIntegral count)-                mapM_ destroyLogicalType childTypes-                pure result-    LogicalTypeUnion memberArray -> do-        let members = elems memberArray-            count = length members-        if count == 0-            then withMany withCString [] \namePtrs ->-                withArray namePtrs \nameArray ->-                    withArray [] \memberPtr ->-                        c_duckdb_create_union_type memberPtr nameArray 0-            else do-                memberTypes <- mapM (logicalTypeFromRep . unionMemberType) members-                let names = map (Text.unpack . unionMemberName) members-                result <--                    withMany withCString names \namePtrs ->-                        withArray namePtrs \nameArray ->-                            withArray memberTypes \memberPtr ->-                                c_duckdb_create_union_type memberPtr nameArray (fromIntegral count)-                mapM_ destroyLogicalType memberTypes-                pure result-    LogicalTypeEnum dictArray -> do-        let entries = elems dictArray-            count = length entries-        if count == 0-            then c_duckdb_create_enum_type nullPtr 0-            else withMany withCString (map Text.unpack entries) \namePtrs ->-                withArray namePtrs \nameArray ->-                    c_duckdb_create_enum_type nameArray (fromIntegral count)+logicalTypeFromRep rep = do+    logical <- create rep+    when (logical == nullPtr) $+        throwIO (userError "duckdb-simple: DuckDB logical type construction failed")+    pure logical+  where+    create = \case+        LogicalTypeScalar dtype -> c_duckdb_create_logical_type dtype+        LogicalTypeDecimal width scale -> do+            when (width < 1 || width > 38 || scale > width) $+                throwIO (userError "duckdb-simple: invalid DECIMAL width or scale")+            c_duckdb_create_decimal_type width scale+        LogicalTypeList elemRep ->+            bracket (logicalTypeFromRep elemRep) destroyLogicalType c_duckdb_create_list_type+        LogicalTypeArray elemRep size ->+            bracket (logicalTypeFromRep elemRep) destroyLogicalType $+                flip c_duckdb_create_array_type size+        LogicalTypeMap keyRep valueRep ->+            bracket (logicalTypeFromRep keyRep) destroyLogicalType \keyType ->+                bracket (logicalTypeFromRep valueRep) destroyLogicalType $+                    c_duckdb_create_map_type keyType+        LogicalTypeStruct fieldArray -> do+            let fields = elems fieldArray+            withMany (\field -> bracket (logicalTypeFromRep (structFieldValue field)) destroyLogicalType) fields \childTypes ->+                withMany withTypeName (map structFieldName fields) \names ->+                    withArray names \nameArray ->+                        withArray childTypes \typeArray ->+                            c_duckdb_create_struct_type typeArray nameArray (fromIntegral (length fields))+        LogicalTypeUnion memberArray -> do+            let members = elems memberArray+            withMany (\member -> bracket (logicalTypeFromRep (unionMemberType member)) destroyLogicalType) members \memberTypes ->+                withMany withTypeName (map unionMemberName members) \names ->+                    withArray names \nameArray ->+                        withArray memberTypes \typeArray ->+                            c_duckdb_create_union_type typeArray nameArray (fromIntegral (length members))+        LogicalTypeEnum dictArray ->+            withMany withTypeName (elems dictArray) \names ->+                withArray names \nameArray ->+                    c_duckdb_create_enum_type nameArray (fromIntegral (length names))++-- | Encode native type names as UTF-8 and reject embedded NUL.+withTypeName :: Text -> (CString -> IO a) -> IO a+withTypeName name action = do+    when (Text.any (== '\0') name) $+        throwIO (userError "duckdb-simple: logical type name contains NUL")+    BS.useAsCString (TextEncoding.encodeUtf8 name) action  word64ToInt :: Text -> Word64 -> IO Int word64ToInt label value =

@@ -16,11 +16,9 @@ import Data.Text (Text) import qualified Data.Text as Text import qualified Data.Text.Encoding as TextEncoding-import Data.Time.Calendar (Day, fromGregorian)-import Data.Time.Clock (UTCTime (..))+import Data.Time.Calendar (addDays, fromGregorian) import Data.Time.Clock.POSIX (posixSecondsToUTCTime) import Data.Time.LocalTime (-    LocalTime (..),     TimeOfDay (..),     minutesToTimeZone,     utc,@@ -46,30 +44,11 @@     UnionValue (..),     logicalTypeToRep,  )+import Database.DuckDB.Simple.Time (Date, LocalTimestamp, UTCTimestamp, Unbounded (..)) import Foreign.C.Types (CBool (..)) import Foreign.Marshal.Alloc (alloca) import Foreign.Ptr (Ptr, castPtr, nullPtr, plusPtr)-import Foreign.Storable (Storable (..), peekByteOff, peekElemOff, pokeByteOff)--data DuckDBListEntry = DuckDBListEntry-    { duckDBListEntryOffset :: !Word64-    , duckDBListEntryLength :: !Word64-    }-    deriving (Eq, Show)--instance Storable DuckDBListEntry where-    sizeOf _ = listEntryWordSize * 2-    alignment _ = alignment (0 :: Word64)-    peek ptr = do-        offset <- peekByteOff ptr 0-        len <- peekByteOff ptr listEntryWordSize-        pure DuckDBListEntry{duckDBListEntryOffset = offset, duckDBListEntryLength = len}-    poke ptr DuckDBListEntry{duckDBListEntryOffset, duckDBListEntryLength} = do-        pokeByteOff ptr 0 duckDBListEntryOffset-        pokeByteOff ptr listEntryWordSize duckDBListEntryLength--listEntryWordSize :: Int-listEntryWordSize = sizeOf (0 :: Word64)+import Foreign.Storable (Storable (..), peekElemOff)  chunkIsRowValid :: Ptr Word64 -> DuckDBIdx -> IO Bool chunkIsRowValid validity rowIdx@@ -453,11 +432,8 @@                     }  vectorElementType :: DuckDBVector -> IO DuckDBType-vectorElementType vec = do-    logical <- c_duckdb_vector_get_column_type vec-    dtype <- c_duckdb_get_type_id logical-    destroyLogicalType logical-    pure dtype+vectorElementType vec =+    bracket (c_duckdb_vector_get_column_type vec) destroyLogicalType c_duckdb_get_type_id  listEntryBounds :: Text -> DuckDBListEntry -> IO (Int, Int) listEntryBounds context DuckDBListEntry{duckDBListEntryOffset, duckDBListEntryLength} = do@@ -477,12 +453,10 @@             then pure (fromInteger actual)             else throwIO (userError ("duckdb-simple: " <> Text.unpack context <> " exceeds Int range")) -decodeDuckDBDate :: DuckDBDate -> IO Day-decodeDuckDBDate raw =-    alloca $ \ptr -> do-        c_duckdb_from_date raw ptr-        dateStruct <- peek ptr-        pure (dateStructToDay dateStruct)+-- | Decode dates with exact epoch arithmetic and preserve infinity.+decodeDuckDBDate :: DuckDBDate -> IO Date+decodeDuckDBDate (DuckDBDate days) =+    pure (decodeUnbounded (\value -> addDays (toInteger value) (fromGregorian 1970 1 1)) days)  decodeDuckDBTime :: DuckDBTime -> IO TimeOfDay decodeDuckDBTime raw =@@ -491,17 +465,21 @@         timeStruct <- peek ptr         pure (timeStructToTimeOfDay timeStruct) -decodeDuckDBTimestamp :: DuckDBTimestamp -> IO LocalTime-decodeDuckDBTimestamp raw =-    alloca $ \ptr -> do-        c_duckdb_from_timestamp raw ptr-        DuckDBTimestampStruct{duckDBTimestampStructDate = dateStruct, duckDBTimestampStructTime = timeStruct} <- peek ptr-        pure-            LocalTime-                { localDay = dateStructToDay dateStruct-                , localTimeOfDay = timeStructToTimeOfDay timeStruct-                }+decodeDuckDBTimestamp :: DuckDBTimestamp -> IO LocalTimestamp+decodeDuckDBTimestamp (DuckDBTimestamp micros) = decodeTimestampUnits 1000000 micros +-- | Interpret native infinity sentinels before converting a finite payload.+decodeUnbounded :: (Integral a, Bounded a) => (a -> b) -> a -> Unbounded b+decodeUnbounded decode value+    | value == maxBound = PosInfinity+    | value == negate maxBound = NegInfinity+    | otherwise = Finite (decode value)++-- | Decode timestamp units without overflowing an intermediate Int64.+decodeTimestampUnits :: Integer -> Int64 -> IO LocalTimestamp+decodeTimestampUnits units =+    pure . decodeUnbounded (utcToLocalTime utc . posixSecondsToUTCTime . fromRational . (% units) . toInteger)+ decodeDuckDBTimeNs :: DuckDBTimeNs -> TimeOfDay decodeDuckDBTimeNs (DuckDBTimeNs nanos) =     let (hours, remainderHours) = nanos `divMod` (60 * 60 * 1000000000)@@ -519,27 +497,27 @@     alloca $ \ptr -> do         c_duckdb_from_time_tz raw ptr         DuckDBTimeTzStruct{duckDBTimeTzStructTime = timeStruct, duckDBTimeTzStructOffset = offset} <- peek ptr+        when (offset `rem` 60 /= 0) $+            throwIO (userError "duckdb-simple: TIMETZ offset cannot be represented in whole minutes")         let timeOfDay = timeStructToTimeOfDay timeStruct             minutes = fromIntegral offset `div` 60             zone = minutesToTimeZone minutes         pure TimeWithZone{timeWithZoneTime = timeOfDay, timeWithZoneZone = zone} -decodeDuckDBTimestampSeconds :: DuckDBTimestampS -> IO LocalTime+decodeDuckDBTimestampSeconds :: DuckDBTimestampS -> IO LocalTimestamp decodeDuckDBTimestampSeconds (DuckDBTimestampS seconds) =-    decodeDuckDBTimestamp (DuckDBTimestamp (seconds * 1000000))+    decodeTimestampUnits 1 seconds -decodeDuckDBTimestampMilliseconds :: DuckDBTimestampMs -> IO LocalTime+decodeDuckDBTimestampMilliseconds :: DuckDBTimestampMs -> IO LocalTimestamp decodeDuckDBTimestampMilliseconds (DuckDBTimestampMs millis) =-    decodeDuckDBTimestamp (DuckDBTimestamp (millis * 1000))+    decodeTimestampUnits 1000 millis -decodeDuckDBTimestampNanoseconds :: DuckDBTimestampNs -> IO LocalTime-decodeDuckDBTimestampNanoseconds (DuckDBTimestampNs nanos) = do-    let utcTime = posixSecondsToUTCTime (fromRational (toInteger nanos % 1000000000))-    pure (utcToLocalTime utc utcTime)+decodeDuckDBTimestampNanoseconds :: DuckDBTimestampNs -> IO LocalTimestamp+decodeDuckDBTimestampNanoseconds (DuckDBTimestampNs nanos) = decodeTimestampUnits 1000000000 nanos -decodeDuckDBTimestampUTCTime :: DuckDBTimestamp -> IO UTCTime+decodeDuckDBTimestampUTCTime :: DuckDBTimestamp -> IO UTCTimestamp decodeDuckDBTimestampUTCTime (DuckDBTimestamp micros) =-    pure (posixSecondsToUTCTime (fromRational (toInteger micros % 1000000)))+    pure (decodeUnbounded (posixSecondsToUTCTime . fromRational . (% 1000000) . toInteger) micros)  intervalValueFromDuckDB :: DuckDBInterval -> IntervalValue intervalValueFromDuckDB DuckDBInterval{duckDBIntervalMonths, duckDBIntervalDays, duckDBIntervalMicros} =@@ -562,13 +540,6 @@     alloca $ \ptr -> do         poke ptr logicalType         c_duckdb_destroy_logical_type ptr--dateStructToDay :: DuckDBDateStruct -> Day-dateStructToDay DuckDBDateStruct{duckDBDateStructYear, duckDBDateStructMonth, duckDBDateStructDay} =-    fromGregorian-        (fromIntegral duckDBDateStructYear)-        (fromIntegral duckDBDateStructMonth)-        (fromIntegral duckDBDateStructDay)  timeStructToTimeOfDay :: DuckDBTimeStruct -> TimeOfDay timeStructToTimeOfDay DuckDBTimeStruct{duckDBTimeStructHour, duckDBTimeStructMinute, duckDBTimeStructSecond, duckDBTimeStructMicros} =

@@ -0,0 +1,240 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE TupleSections #-}++-- | Shared result decoding and cursor ownership for both execution modes.+module Database.DuckDB.Simple.Result (+    collectRows,+    foldStatementWith,+    nextRowWith,+    resetStatementStream,+    cleanupStatementStreamRef,+) where++import Control.Exception (bracket, evaluate, finally, mask, mask_, onException, throwIO)+import Control.Monad (forM, when, zipWithM)+import Data.IORef (IORef, atomicModifyIORef', readIORef, writeIORef)+import qualified Data.Text as Text+import Database.DuckDB.FFI+import Database.DuckDB.Simple.FromField (Field (..))+import Database.DuckDB.Simple.FromRow (RowParser, parseRow, rowErrorsToSqlError)+import Database.DuckDB.Simple.Internal+import Database.DuckDB.Simple.Materialize (materializeValue)+import Database.DuckDB.Simple.Ok (Ok (..))+import Foreign.Marshal.Alloc (free, malloc)+import Foreign.Marshal.Utils (fillBytes)+import Foreign.Ptr (Ptr, nullPtr)+import Foreign.Storable (sizeOf)++-- | Fold rows from one statement and release its result on every exit path.+foldStatementWith :: ResultMode -> RowParser row -> Statement -> a -> (a -> row -> IO a) -> IO a+foldStatementWith mode parser stmt initial step =+    let loop acc = do+            nextVal <- nextRowWith mode parser stmt+            case nextVal of+                Nothing -> pure acc+                Just row -> do+                    acc' <- step acc row+                    acc' `seq` loop acc'+     in loop initial `finally` resetStatementStream stmt++-- | Read one row. The first fetch selects execution until the cursor resets.+nextRowWith :: ResultMode -> RowParser r -> Statement -> IO (Maybe r)+nextRowWith mode parser stmt@Statement{statementStream} =+    withStatementHandle stmt \_ -> mask \restore -> do+        state <- readIORef statementStream+        case state of+            StatementStreamExhausted -> pure Nothing+            StatementStreamIdle -> do+                newStream <- startStatementStream mode stmt+                case newStream of+                    Nothing -> writeIORef statementStream StatementStreamExhausted >> pure Nothing+                    Just stream -> do+                        writeIORef statementStream (StatementStreamActive stream)+                        restore (consumeStream statementStream parser stmt stream)+                            `onException` exhaustStatementStream statementStream+            StatementStreamActive stream ->+                restore (consumeStream statementStream parser stmt stream)+                    `onException` exhaustStatementStream statementStream++-- | Release an active cursor and allow the statement to execute again.+resetStatementStream :: Statement -> IO ()+resetStatementStream Statement{statementStream} =+    cleanupStatementStreamRef statementStream++consumeStream :: IORef StatementStreamState -> RowParser r -> Statement -> StatementStream -> IO (Maybe r)+consumeStream streamRef parser stmt stream = mask \restore -> do+    loaded <- case statementStreamChunk stream of+        Nothing -> fetchChunk (statementConnection stmt) (statementQuery stmt) stream+        Just _ -> pure stream+    writeIORef streamRef (StatementStreamActive loaded)+    case statementStreamChunk loaded of+        Nothing -> exhaustStatementStream streamRef >> pure Nothing+        Just chunk -> do+            fields <- restore $ buildMaterializedRow (statementStreamColumns loaded) (statementStreamChunkVectors chunk) (statementStreamChunkIndex chunk)+            parsed <- restore (evaluate (parseRow parser fields))+            case parsed of+                Errors rowErr -> throwIO $ rowErrorsToSqlError (statementQuery stmt) rowErr+                Ok value -> do+                    let nextIndex = statementStreamChunkIndex chunk + 1+                    if nextIndex < statementStreamChunkSize chunk+                        then writeIORef streamRef (StatementStreamActive loaded{statementStreamChunk = Just chunk{statementStreamChunkIndex = nextIndex}})+                        else do+                            writeIORef streamRef (StatementStreamActive loaded{statementStreamChunk = Nothing})+                            finalizeChunk chunk+                    pure (Just value)++-- | Release the cursor and retain its exhausted state until an explicit reset.+exhaustStatementStream :: IORef StatementStreamState -> IO ()+exhaustStatementStream ref = mask_ do+    state <- atomicModifyIORef' ref (StatementStreamExhausted,)+    finalizeStreamState state++startStatementStream :: ResultMode -> Statement -> IO (Maybe StatementStream)+startStatementStream mode stmt =+    withStatementHandle stmt \handle -> do+        resultPtr <- malloc+        fillBytes resultPtr 0 (sizeOf (undefined :: DuckDBResult))+        let release = c_duckdb_destroy_result resultPtr `finally` free resultPtr+        flip onException release do+            rc <- runInterruptibleQuery (statementConnection stmt) (executePreparedResult mode handle resultPtr)+            when (rc /= DuckDBSuccess) do+                (errMsg, errType) <- fetchResultError resultPtr+                throwIO $ mkExecuteError (statementQuery stmt) errMsg errType+            resultType <- c_duckdb_result_return_type resultPtr+            if resultType /= DuckDBResultTypeQueryResult+                then release >> pure Nothing+                else do+                    columns <- collectResultColumns resultPtr+                    pure (Just (StatementStream resultPtr columns Nothing mode))++fetchChunk :: Connection -> Query -> StatementStream -> IO StatementStream+fetchChunk conn queryText stream@StatementStream{statementStreamResult} = do+    chunk <- fetchResultChunk (statementStreamMode stream) conn statementStreamResult+    if chunk == nullPtr+        then do+            throwResultError queryText statementStreamResult+            pure stream+        else do+            rawSize <- c_duckdb_data_chunk_get_size chunk+            let rowCount = fromIntegral rawSize :: Int+            if rowCount <= 0+                then do+                    destroyDataChunk chunk+                    fetchChunk conn queryText stream+                else do+                    vectors <-+                        prepareChunkVectors chunk (statementStreamColumns stream)+                            `onException` destroyDataChunk chunk+                    let chunkState =+                            StatementStreamChunk+                                { statementStreamChunkPtr = chunk+                                , statementStreamChunkSize = rowCount+                                , statementStreamChunkIndex = 0+                                , statementStreamChunkVectors = vectors+                                }+                    pure stream{statementStreamChunk = Just chunkState}++prepareChunkVectors :: DuckDBDataChunk -> [StatementStreamColumn] -> IO [StatementStreamChunkVector]+prepareChunkVectors chunk columns =+    forM columns \StatementStreamColumn{statementStreamColumnIndex} -> do+        vector <- c_duckdb_data_chunk_get_vector chunk (fromIntegral statementStreamColumnIndex)+        dataPtr <- c_duckdb_vector_get_data vector+        validity <- c_duckdb_vector_get_validity vector+        pure+            StatementStreamChunkVector+                { statementStreamChunkVectorHandle = vector+                , statementStreamChunkVectorData = dataPtr+                , statementStreamChunkVectorValidity = validity+                }++-- | Clear cursor ownership before releasing native resources.+cleanupStatementStreamRef :: IORef StatementStreamState -> IO ()+cleanupStatementStreamRef ref = mask_ do+    state <- atomicModifyIORef' ref (StatementStreamIdle,)+    finalizeStreamState state++finalizeStreamState :: StatementStreamState -> IO ()+finalizeStreamState = \case+    StatementStreamIdle -> pure ()+    StatementStreamExhausted -> pure ()+    StatementStreamActive stream -> finalizeStream stream++finalizeStream :: StatementStream -> IO ()+finalizeStream StatementStream{statementStreamResult, statementStreamChunk} = do+    maybe (pure ()) finalizeChunk statementStreamChunk+    c_duckdb_destroy_result statementStreamResult+    free statementStreamResult++finalizeChunk :: StatementStreamChunk -> IO ()+finalizeChunk StatementStreamChunk{statementStreamChunkPtr} =+    destroyDataChunk statementStreamChunkPtr++-- | Copy all rows from a materialized native result.+collectRows :: Query -> Ptr DuckDBResult -> IO [[Field]]+collectRows queryText resPtr = do+    columns <- collectResultColumns resPtr+    collectChunks columns []+  where+    collectChunks columns acc = do+        fetched <- bracket (c_duckdb_fetch_chunk resPtr) destroyDataChunk \chunk ->+            if chunk == nullPtr+                then throwResultError queryText resPtr >> pure Nothing+                else Just <$> decodeChunk columns chunk+        case fetched of+            Nothing -> pure (concat (reverse acc))+            Just rows -> do+                let acc' = maybe acc (: acc) rows+                collectChunks columns acc'++    decodeChunk columns chunk = do+        rawSize <- c_duckdb_data_chunk_get_size chunk+        let rowCount = fromIntegral rawSize :: Int+        if rowCount <= 0+            then pure Nothing+            else+                if null columns+                    then pure (Just (replicate rowCount []))+                    else do+                        vectors <- prepareChunkVectors chunk columns+                        rows <- mapM (buildMaterializedRow columns vectors) [0 .. rowCount - 1]+                        pure (Just rows)++collectResultColumns :: Ptr DuckDBResult -> IO [StatementStreamColumn]+collectResultColumns resPtr = do+    rawCount <- c_duckdb_column_count resPtr+    let cc = fromIntegral rawCount :: Int+    forM [0 .. cc - 1] \columnIndex -> do+        namePtr <- c_duckdb_column_name resPtr (fromIntegral columnIndex)+        name <-+            if namePtr == nullPtr+                then pure (Text.pack ("column" <> show columnIndex))+                else peekUtf8CString namePtr+        dtype <- c_duckdb_column_type resPtr (fromIntegral columnIndex)+        pure+            StatementStreamColumn+                { statementStreamColumnIndex = columnIndex+                , statementStreamColumnName = name+                , statementStreamColumnType = dtype+                }++buildMaterializedRow :: [StatementStreamColumn] -> [StatementStreamChunkVector] -> Int -> IO [Field]+buildMaterializedRow columns vectors rowIdx =+    zipWithM (buildMaterializedField rowIdx) columns vectors++buildMaterializedField :: Int -> StatementStreamColumn -> StatementStreamChunkVector -> IO Field+buildMaterializedField rowIdx column StatementStreamChunkVector{statementStreamChunkVectorHandle, statementStreamChunkVectorData, statementStreamChunkVectorValidity} = do+    value <-+        materializeValue+            (statementStreamColumnType column)+            statementStreamChunkVectorHandle+            statementStreamChunkVectorData+            statementStreamChunkVectorValidity+            rowIdx+    pure+        Field+            { fieldName = statementStreamColumnName column+            , fieldIndex = statementStreamColumnIndex column+            , fieldValue = value+            }

@@ -0,0 +1,27 @@+{-# LANGUAGE DeriveFunctor #-}++-- | Dates and timestamps that can represent DuckDB's infinities.+module Database.DuckDB.Simple.Time (+    Unbounded (..),+    Date,+    LocalTimestamp,+    UTCTimestamp,+) where++import Data.Time (Day, LocalTime, UTCTime)++-- | A finite value or either infinity, as in @postgresql-simple@.+data Unbounded a+    = NegInfinity+    | Finite !a+    | PosInfinity+    deriving (Eq, Ord, Show, Read, Functor)++-- | A DuckDB DATE, including infinity.+type Date = Unbounded Day++-- | A timestamp without a time zone, including infinity.+type LocalTimestamp = Unbounded LocalTime++-- | A timestamp with a time zone, including infinity.+type UTCTimestamp = Unbounded UTCTime

@@ -25,16 +25,16 @@ ) where  import Control.Exception (bracket, throwIO)-import Control.Monad (when, zipWithM)+import Control.Monad (when) import Data.Array (Array, elems)-import Data.Bits (complement, shiftL, shiftR, (.&.))+import Data.Bits (complement, shiftL, shiftR, (.&.), (.|.)) import qualified Data.ByteString as BS import Data.Int (Int16, Int32, Int64, Int8) import Data.Proxy (Proxy (..)) import Data.Text (Text) import qualified Data.Text as Text-import qualified Data.Text.Foreign as TextForeign-import Data.Time.Calendar (Day, toGregorian)+import qualified Data.Text.Encoding as TextEncoding+import Data.Time.Calendar (Day, diffDays, fromGregorian) import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds) import Data.Time.LocalTime (LocalTime (..), TimeOfDay (..), TimeZone (..), timeOfDayToTime, timeZoneMinutes, utc, utcToLocalTime) import qualified Data.UUID as UUID@@ -56,12 +56,14 @@     structValueTypeRep,     unionValueTypeRep,  )+import Database.DuckDB.Simple.Time (Date, LocalTimestamp, UTCTimestamp, Unbounded (..)) import Database.DuckDB.Simple.Types (Null (..)) import Foreign.C.String (peekCString)-import Foreign.C.Types (CDouble (..))+import Foreign.C.Types (CDouble (..), CFloat (..)) import Foreign.Marshal (fromBool) import Foreign.Marshal.Alloc (alloca) import Foreign.Marshal.Array (withArray)+import Foreign.Marshal.Utils (withMany) import Foreign.Ptr (Ptr, castPtr, nullPtr) import Foreign.Storable (poke) import Numeric.Natural (Natural)@@ -143,6 +145,9 @@ instance ToField TimeOfDay instance ToField LocalTime instance ToField UTCTime+instance ToField (Unbounded Day)+instance ToField (Unbounded LocalTime)+instance ToField (Unbounded UTCTime)  instance ToField BigNum where     toField big@(BigNum n) = valueBinding (show n) (bigNumDuckValue big)@@ -254,6 +259,15 @@ instance DuckDBColumnType UTCTime where     duckdbColumnTypeFor _ = "TIMESTAMPTZ" +instance DuckDBColumnType (Unbounded Day) where+    duckdbColumnTypeFor _ = "DATE"++instance DuckDBColumnType (Unbounded LocalTime) where+    duckdbColumnTypeFor _ = "TIMESTAMP"++instance DuckDBColumnType (Unbounded UTCTime) where+    duckdbColumnTypeFor _ = "TIMESTAMPTZ"+ instance DuckDBColumnType (StructValue FieldValue) where     duckdbColumnTypeFor _ = "STRUCT" @@ -303,11 +317,12 @@ doubleDuckValue = c_duckdb_create_double . CDouble  floatDuckValue :: Float -> IO DuckDBValue-floatDuckValue = c_duckdb_create_double . CDouble . realToFrac+floatDuckValue = c_duckdb_create_float . CFloat  textDuckValue :: Text -> IO DuckDBValue textDuckValue txt =-    TextForeign.withCString txt c_duckdb_create_varchar+    BS.useAsCStringLen (TextEncoding.encodeUtf8 txt) \(ptr, len) ->+        c_duckdb_create_varchar_length ptr (fromIntegral len)  stringDuckValue :: String -> IO DuckDBValue stringDuckValue = textDuckValue . Text.pack@@ -330,29 +345,15 @@         c_duckdb_create_uuid ptr  bitDuckValue :: BitString -> IO DuckDBValue-bitDuckValue (BitString padding bits) =-    let withPacked action =-            if BS.null bits-                then alloca \ptr -> do-                    poke-                        ptr-                        DuckDBBit-                            { duckDBBitData = nullPtr-                            , duckDBBitSize = 0-                            }-                    action ptr-                else-                    let payload = BS.pack ((fromIntegral padding :: Word8) : BS.unpack bits)-                     in BS.useAsCStringLen payload \(rawPtr, len) ->-                            alloca \ptr -> do-                                poke-                                    ptr-                                    DuckDBBit-                                        { duckDBBitData = castPtr rawPtr-                                        , duckDBBitSize = fromIntegral len-                                        }-                                action ptr-     in withPacked c_duckdb_create_bit+bitDuckValue (BitString padding bits) = do+    when (BS.null bits || padding > 7) $+        throwIO (userError "duckdb-simple: BIT requires nonempty data and padding from 0 to 7")+    let nativePadding = complement ((1 `shiftL` (8 - fromIntegral padding)) - 1) :: Word8+        payload = BS.cons padding (BS.cons (BS.head bits .|. nativePadding) (BS.tail bits))+    BS.useAsCStringLen payload \(rawPtr, len) ->+        alloca \ptr -> do+            poke ptr DuckDBBit{duckDBBitData = castPtr rawPtr, duckDBBitSize = fromIntegral len}+            c_duckdb_create_bit ptr  bigNumDuckValue :: BigNum -> IO DuckDBValue bigNumDuckValue (BigNum big) =@@ -402,24 +403,33 @@  utcTimeDuckValue :: UTCTime -> IO DuckDBValue utcTimeDuckValue utcTime =-    let local = utcToLocalTime utc utcTime-     in localTimeDuckValue local+    encodeLocalTime (utcToLocalTime utc utcTime) >>= c_duckdb_create_timestamp_tz +-- | Bind a date, including either infinity sentinel.+dateDuckValue :: Date -> IO DuckDBValue+dateDuckValue value =+    encodeUnbounded (fmap unDuckDBDate . encodeDay) value >>= c_duckdb_create_date . DuckDBDate++-- | Bind a timestamp without a time zone, including infinity.+localTimestampDuckValue :: LocalTimestamp -> IO DuckDBValue+localTimestampDuckValue value =+    encodeUnbounded (encodeTimestampUnits 1000000) value >>= c_duckdb_create_timestamp . DuckDBTimestamp++-- | Bind a timestamp with a time zone, including infinity.+utcTimestampDuckValue :: UTCTimestamp -> IO DuckDBValue+utcTimestampDuckValue value =+    encodeUnbounded (encodeTimestampUnits 1000000 . utcToLocalTime utc) value >>= c_duckdb_create_timestamp_tz . DuckDBTimestamp+ arrayDuckValue ::     forall a.     (DuckDBColumnType a, ToDuckValue a) =>     Array Int a ->     IO DuckDBValue arrayDuckValue arr =-    bracket (createElementLogicalType (Proxy :: Proxy a)) destroyLogicalType \elementType -> do-        let elemsList = elems arr-            count = length elemsList-        values <- mapM toDuckValue elemsList-        result <--            withArray values \ptr ->-                c_duckdb_create_array_value elementType ptr (fromIntegral count)-        mapM_ destroyValue values-        pure result+    bracket (createElementLogicalType (Proxy :: Proxy a)) destroyLogicalType \elementType ->+        withCreatedValues (map toDuckValue (elems arr)) \values ->+            withDuckValues values \ptr ->+                checkedValue (c_duckdb_create_array_value elementType ptr (fromIntegral (length values)))  structValueDuckValue :: StructValue FieldValue -> IO DuckDBValue structValueDuckValue StructValue{structValueFields, structValueTypes, structValueIndex = _} = do@@ -431,37 +441,39 @@         throwIO (userError "duckdb-simple: struct value/type arity mismatch")     when (typeNames /= valueNames) $         throwIO (userError "duckdb-simple: struct value/type field names mismatch")-    childValues <--        zipWithM-            ( \StructField{structFieldValue = typeRep} StructField{structFieldValue = fieldVal} ->-                fieldValueWithTypeDuckValue typeRep fieldVal-            )-            typeFields-            valueFields-    structLogical <- logicalTypeFromRep (LogicalTypeStruct structValueTypes)-    result <--        withDuckValues childValues $ c_duckdb_create_struct_value structLogical-    mapM_ destroyValue childValues-    destroyLogicalType structLogical-    pure result+    let actions =+            zipWith+                ( \StructField{structFieldValue = typeRep} StructField{structFieldValue = fieldVal} ->+                    fieldValueWithTypeDuckValue typeRep fieldVal+                )+                typeFields+                valueFields+    bracket (logicalTypeFromRep (LogicalTypeStruct structValueTypes)) destroyLogicalType \structLogical ->+        withCreatedValues actions \childValues ->+            withDuckValues childValues $ \ptr ->+                checkedValue (c_duckdb_create_struct_value structLogical ptr)  unionValueDuckValue :: UnionValue FieldValue -> IO DuckDBValue-unionValueDuckValue UnionValue{unionValueIndex, unionValuePayload, unionValueMembers} = do+unionValueDuckValue UnionValue{unionValueIndex, unionValueLabel, unionValuePayload, unionValueMembers} = do     let membersList = elems unionValueMembers         idx = fromIntegral unionValueIndex :: Int         memberCount = length membersList     when (idx < 0 || idx >= memberCount) $         throwIO (userError "duckdb-simple: union value tag out of range")-    let UnionMemberType{unionMemberType = memberType} = membersList !! idx-    payloadValue <- fieldValueWithTypeDuckValue memberType unionValuePayload-    unionLogical <- logicalTypeFromRep (LogicalTypeUnion unionValueMembers)-    result <- c_duckdb_create_union_value unionLogical (fromIntegral unionValueIndex) payloadValue-    destroyValue payloadValue-    destroyLogicalType unionLogical-    pure result+    let UnionMemberType{unionMemberName, unionMemberType = memberType} = membersList !! idx+    when (unionValueLabel /= unionMemberName) $+        throwIO (userError "duckdb-simple: union tag and member name mismatch")+    bracket (logicalTypeFromRep (LogicalTypeUnion unionValueMembers)) destroyLogicalType \unionLogical ->+        bracket (checkedValue (fieldValueWithTypeDuckValue memberType unionValuePayload)) destroyValue \payloadValue ->+            checkedValue (c_duckdb_create_union_value unionLogical (fromIntegral unionValueIndex) payloadValue)  fieldValueWithTypeDuckValue :: LogicalTypeRep -> FieldValue -> IO DuckDBValue-fieldValueWithTypeDuckValue _ FieldNull = nullDuckValue+fieldValueWithTypeDuckValue typeRep FieldNull =+    bracket (logicalTypeFromRep typeRep) destroyLogicalType \logical ->+        withCreatedValues [nullDuckValue] \values ->+            withDuckValues values \ptr ->+                bracket (checkedValue (c_duckdb_create_list_value logical ptr 1)) destroyValue \list ->+                    checkedValue (c_duckdb_get_list_child list 0) fieldValueWithTypeDuckValue rep value =     case rep of         LogicalTypeScalar dtype -> scalarFieldValueDuckValue dtype value@@ -473,15 +485,11 @@                 other -> typeMismatch "DECIMAL" other         LogicalTypeList elemRep ->             case value of-                FieldList elemsList -> do-                    childLogical <- logicalTypeFromRep elemRep-                    values <- mapM (fieldValueWithTypeDuckValue elemRep) elemsList-                    result <--                        withDuckValues values \ptr ->-                            c_duckdb_create_list_value childLogical ptr (fromIntegral (length elemsList))-                    mapM_ destroyValue values-                    destroyLogicalType childLogical-                    pure result+                FieldList elemsList ->+                    bracket (logicalTypeFromRep elemRep) destroyLogicalType \childLogical ->+                        withCreatedValues (map (fieldValueWithTypeDuckValue elemRep) elemsList) \values ->+                            withDuckValues values \ptr ->+                                checkedValue (c_duckdb_create_list_value childLogical ptr (fromIntegral (length values)))                 other -> typeMismatch "LIST" other         LogicalTypeArray elemRep size ->             case value of@@ -490,30 +498,20 @@                         actualCount = length elemsList                     when (fromIntegral actualCount /= size) $                         throwIO (userError "duckdb-simple: array length mismatch")-                    childLogical <- logicalTypeFromRep elemRep-                    values <- mapM (fieldValueWithTypeDuckValue elemRep) elemsList-                    result <--                        withDuckValues values \ptr ->-                            c_duckdb_create_array_value childLogical ptr (fromIntegral actualCount)-                    mapM_ destroyValue values-                    destroyLogicalType childLogical-                    pure result+                    bracket (logicalTypeFromRep elemRep) destroyLogicalType \childLogical ->+                        withCreatedValues (map (fieldValueWithTypeDuckValue elemRep) elemsList) \values ->+                            withDuckValues values \ptr ->+                                checkedValue (c_duckdb_create_array_value childLogical ptr (fromIntegral actualCount))                 other -> typeMismatch "ARRAY" other         LogicalTypeMap keyRep valueRep ->             case value of-                FieldMap pairs -> do-                    let count = length pairs-                    keyValues <- mapM (fieldValueWithTypeDuckValue keyRep . fst) pairs-                    valValues <- mapM (fieldValueWithTypeDuckValue valueRep . snd) pairs-                    mapLogical <- logicalTypeFromRep (LogicalTypeMap keyRep valueRep)-                    result <--                        withDuckValues keyValues \keyPtr ->-                            withDuckValues valValues \valPtr ->-                                c_duckdb_create_map_value mapLogical keyPtr valPtr (fromIntegral count)-                    mapM_ destroyValue keyValues-                    mapM_ destroyValue valValues-                    destroyLogicalType mapLogical-                    pure result+                FieldMap pairs ->+                    bracket (logicalTypeFromRep (LogicalTypeMap keyRep valueRep)) destroyLogicalType \mapLogical ->+                        withCreatedValues (map (fieldValueWithTypeDuckValue keyRep . fst) pairs) \keyValues ->+                            withCreatedValues (map (fieldValueWithTypeDuckValue valueRep . snd) pairs) \valValues ->+                                withDuckValues keyValues \keyPtr ->+                                    withDuckValues valValues \valPtr ->+                                        checkedValue (c_duckdb_create_map_value mapLogical keyPtr valPtr (fromIntegral (length pairs)))                 other -> typeMismatch "MAP" other         LogicalTypeStruct structRep ->             case value of@@ -550,11 +548,19 @@         (DuckDBTypeBlob, FieldBlob b) -> blobDuckValue b         (DuckDBTypeUUID, FieldUUID u) -> uuidDuckValue u         (DuckDBTypeBit, FieldBit bits) -> bitDuckValue bits-        (DuckDBTypeDate, FieldDate d) -> dayDuckValue d+        (DuckDBTypeDate, FieldDate d) -> dateDuckValue d         (DuckDBTypeTime, FieldTime t) -> timeOfDayDuckValue t+        (DuckDBTypeTimeNs, FieldTime t) ->+            timeOfDayUnits 1000000000 t >>= c_duckdb_create_time_ns . DuckDBTimeNs . fromInteger         (DuckDBTypeTimeTz, FieldTimeTZ tz) -> timeWithZoneDuckValue tz-        (DuckDBTypeTimestamp, FieldTimestamp ts) -> localTimeDuckValue ts-        (DuckDBTypeTimestampTz, FieldTimestampTZ ts) -> utcTimeDuckValue ts+        (DuckDBTypeTimestamp, FieldTimestamp ts) -> localTimestampDuckValue ts+        (DuckDBTypeTimestampS, FieldTimestamp ts) ->+            encodeUnbounded (encodeTimestampUnits 1) ts >>= c_duckdb_create_timestamp_s . DuckDBTimestampS+        (DuckDBTypeTimestampMs, FieldTimestamp ts) ->+            encodeUnbounded (encodeTimestampUnits 1000) ts >>= c_duckdb_create_timestamp_ms . DuckDBTimestampMs+        (DuckDBTypeTimestampNs, FieldTimestamp ts) ->+            encodeUnbounded (encodeTimestampUnits 1000000000) ts >>= c_duckdb_create_timestamp_ns . DuckDBTimestampNs+        (DuckDBTypeTimestampTz, FieldTimestampTZ ts) -> utcTimestampDuckValue ts         (DuckDBTypeInterval, FieldInterval iv) -> intervalDuckValue iv         (DuckDBTypeHugeInt, FieldHugeInt i) -> hugeIntDuckValue i         (DuckDBTypeUHugeInt, FieldUHugeInt i) -> uhugeIntDuckValue i@@ -575,10 +581,10 @@  enumDuckValue :: Array Int Text -> Word32 -> IO DuckDBValue enumDuckValue dict idx = do-    enumLogical <- logicalTypeFromRep (LogicalTypeEnum dict)-    result <- c_duckdb_create_enum_value enumLogical (fromIntegral idx)-    destroyLogicalType enumLogical-    pure result+    when (toInteger idx >= toInteger (length (elems dict))) $+        throwIO (userError "duckdb-simple: ENUM index out of range")+    bracket (logicalTypeFromRep (LogicalTypeEnum dict)) destroyLogicalType \enumLogical ->+        checkedValue (c_duckdb_create_enum_value enumLogical (fromIntegral idx))  hugeIntDuckValue :: Integer -> IO DuckDBValue hugeIntDuckValue value =@@ -596,6 +602,10 @@  decimalDuckValue :: DecimalValue -> IO DuckDBValue decimalDuckValue DecimalValue{decimalWidth, decimalScale, decimalInteger} = do+    when (decimalWidth < 1 || decimalWidth > 38 || decimalScale > decimalWidth) $+        throwIO (userError "duckdb-simple: invalid DECIMAL width or scale")+    when (abs decimalInteger >= 10 ^ decimalWidth) $+        throwIO (userError "duckdb-simple: DECIMAL value exceeds declared precision")     huge <- integerToHugeInt decimalInteger     alloca \ptr -> do         poke@@ -615,8 +625,10 @@  timeWithZoneDuckValue :: TimeWithZone -> IO DuckDBValue timeWithZoneDuckValue TimeWithZone{timeWithZoneTime, timeWithZoneZone} = do-    let totalMicros = diffTimeToPicoseconds (timeOfDayToTime timeWithZoneTime) `div` 1000000-        offsetSeconds = timeZoneMinutes timeWithZoneZone * 60+    totalMicros <- timeOfDayUnits 1000000 timeWithZoneTime+    let offsetSeconds = toInteger (timeZoneMinutes timeWithZoneZone) * 60+    when (abs offsetSeconds > 57599) $+        throwIO (userError "duckdb-simple: TIME WITH TIME ZONE offset out of range")     tzValue <- c_duckdb_create_time_tz (fromIntegral totalMicros) (fromIntegral offsetSeconds)     c_duckdb_create_time_tz_value tzValue @@ -642,6 +654,18 @@         upper = fromIntegral (value `shiftR` 64)     pure DuckDBUHugeInt{duckDBUHugeIntLower = lower, duckDBUHugeIntUpper = upper} +-- | Release every child handle when construction fails or completes.+withCreatedValues :: [IO DuckDBValue] -> ([DuckDBValue] -> IO a) -> IO a+withCreatedValues = withMany (\action -> bracket (checkedValue action) destroyValue)++-- | Reject failed native constructors before a handle is used.+checkedValue :: IO DuckDBValue -> IO DuckDBValue+checkedValue action = do+    value <- action+    when (value == nullPtr) $+        throwIO (userError "duckdb-simple: DuckDB value construction failed")+    pure value+ withDuckValues :: [DuckDBValue] -> (Ptr DuckDBValue -> IO a) -> IO a withDuckValues xs action = withArray xs action @@ -785,6 +809,15 @@ instance ToDuckValue UTCTime where     toDuckValue = utcTimeDuckValue +instance ToDuckValue (Unbounded Day) where+    toDuckValue = dateDuckValue++instance ToDuckValue (Unbounded LocalTime) where+    toDuckValue = localTimestampDuckValue++instance ToDuckValue (Unbounded UTCTime) where+    toDuckValue = utcTimestampDuckValue+ instance ToDuckValue (StructValue FieldValue) where     toDuckValue = structValueDuckValue @@ -795,56 +828,55 @@     toDuckValue Nothing = nullDuckValue     toDuckValue (Just value) = toDuckValue value +-- | Preserve infinity sentinels and validate finite values before narrowing.+encodeUnbounded :: (Integral b, Bounded b) => (a -> IO b) -> Unbounded a -> IO b+encodeUnbounded _ NegInfinity = pure (negate maxBound)+encodeUnbounded _ PosInfinity = pure maxBound+encodeUnbounded encode (Finite value) = encode value++-- | Encode finite dates as days from the Unix epoch. encodeDay :: Day -> IO DuckDBDate-encodeDay day =-    alloca \ptr -> do-        poke ptr (dayToDateStruct day)-        c_duckdb_to_date ptr+encodeDay day = do+    let days = diffDays day (fromGregorian 1970 1 1)+    checkFiniteRange "DATE" (minBound :: Int32) maxBound days+    pure (DuckDBDate (fromInteger days)) +-- | Encode a time of day at DuckDB microsecond precision. encodeTimeOfDay :: TimeOfDay -> IO DuckDBTime-encodeTimeOfDay tod =-    alloca \ptr -> do-        poke ptr (timeOfDayToStruct tod)-        c_duckdb_to_time ptr+encodeTimeOfDay tod = DuckDBTime . fromInteger <$> timeOfDayUnits 1000000 tod +-- | Encode finite timestamps without native calendar conversions. encodeLocalTime :: LocalTime -> IO DuckDBTimestamp-encodeLocalTime LocalTime{localDay, localTimeOfDay} =-    alloca \ptr -> do-        poke-            ptr-            DuckDBTimestampStruct-                { duckDBTimestampStructDate = dayToDateStruct localDay-                , duckDBTimestampStructTime = timeOfDayToStruct localTimeOfDay-                }-        c_duckdb_to_timestamp ptr+encodeLocalTime ts = DuckDBTimestamp <$> encodeTimestampUnits 1000000 ts -dayToDateStruct :: Day -> DuckDBDateStruct-dayToDateStruct day =-    let (year, month, dayOfMonth) = toGregorian day-     in DuckDBDateStruct-            { duckDBDateStructYear = fromIntegral year-            , duckDBDateStructMonth = fromIntegral month-            , duckDBDateStructDay = fromIntegral dayOfMonth-            }+-- | Encode finite timestamps in the requested number of units per second.+encodeTimestampUnits :: Integer -> LocalTime -> IO Int64+encodeTimestampUnits units LocalTime{localDay, localTimeOfDay} = do+    time <- timeOfDayUnits units localTimeOfDay+    let total = diffDays localDay (fromGregorian 1970 1 1) * 86400 * units + time+    checkFiniteRange "TIMESTAMP" (minBound :: Int64) maxBound total+    pure (fromInteger total) -timeOfDayToStruct :: TimeOfDay -> DuckDBTimeStruct-timeOfDayToStruct tod =-    let totalPicoseconds = diffTimeToPicoseconds (timeOfDayToTime tod)-        totalMicros = totalPicoseconds `div` 1000000-        (hours, remHour) = totalMicros `divMod` (60 * 60 * 1000000)-        (minutes, remMinute) = remHour `divMod` (60 * 1000000)-        (seconds, micros) = remMinute `divMod` 1000000-     in DuckDBTimeStruct-            { duckDBTimeStructHour = fromIntegral hours-            , duckDBTimeStructMinute = fromIntegral minutes-            , duckDBTimeStructSecond = fromIntegral seconds-            , duckDBTimeStructMicros = fromIntegral micros-            }+-- | Check storage limits and the two DuckDB infinity sentinels.+checkFiniteRange :: (Integral a) => String -> a -> a -> Integer -> IO ()+checkFiniteRange label lower upper value =+    when (value < toInteger lower || value >= toInteger upper || value == negate (toInteger upper)) $+        throwIO (userError ("duckdb-simple: " <> label <> " value out of finite range")) +-- | Validate time components and convert to the requested units per second.+timeOfDayUnits :: Integer -> TimeOfDay -> IO Integer+timeOfDayUnits units tod@(TimeOfDay hours minutes seconds) = do+    when (hours < 0 || hours > 24 || minutes < 0 || minutes > 59 || seconds < 0 || seconds >= 61 || (hours == 24 && (minutes /= 0 || seconds /= 0))) $+        throwIO (userError "duckdb-simple: invalid time of day")+    let value = diffTimeToPicoseconds (timeOfDayToTime tod) `div` (1000000000000 `div` units)+    when (value > 86400 * units) $+        throwIO (userError "duckdb-simple: time of day out of range")+    pure value+ bindDuckValue :: Statement -> DuckDBIdx -> IO DuckDBValue -> IO () bindDuckValue stmt idx makeValue =     withStatementHandle stmt \handle ->-        bracket makeValue destroyValue \value -> do+        bracket (checkedValue makeValue) destroyValue \value -> do             rc <- c_duckdb_bind_value handle idx value             when (rc /= DuckDBSuccess) $ do                 err <- fetchPrepareError handle

@@ -0,0 +1,241 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-deprecations #-}++-- | Integration tests for scoped Arrow batches and their native ownership.+module ArrowTests (arrowTests) where++import Control.Concurrent (forkIO, killThread, newEmptyMVar, putMVar, takeMVar)+import Control.Exception (AsyncException (ThreadKilled), IOException, SomeException, bracket, bracket_, fromException, mask_, throwIO, try)+import Control.Monad (forM, forM_, when)+import Data.Bits (testBit)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.Int (Int64)+import qualified Data.Text as Text+import Data.Word (Word8)+import Database.DuckDB.FFI+import Database.DuckDB.Simple+import qualified Database.DuckDB.Simple.Arrow as Arrow+import qualified Database.DuckDB.Simple.Deprecated.Streaming as Streaming+import Database.DuckDB.Simple.Internal (peekUtf8CString)+import Foreign.C.String (peekCString)+import Foreign.Marshal.Alloc (alloca)+import Foreign.Marshal.Utils (fillBytes)+import Foreign.Ptr (Ptr, castPtr, freeHaskellFunPtr, nullFunPtr, nullPtr)+import Foreign.Storable (peek, peekElemOff, poke, sizeOf)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit++-- | Exercise native results and count their Arrow release callbacks.+arrowTests :: TestTree+arrowTests =+    testGroup "Arrow batches" [arrowModeTests False, arrowModeTests True]++-- | Run the same ownership checks for both public Arrow interfaces.+arrowModeTests :: Bool -> TestTree+arrowModeTests streaming =+    testGroup+        (if streaming then "deprecated streaming" else "materialized")+        [ testCase "parameters, Unicode names, NULLs and multiple batches" $+            withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+                (batches, chunks) <-+                    foldArrow conn "SELECT CASE WHEN i % 3 = 0 THEN NULL ELSE i END AS \"íslenska_λ\" FROM range(?::BIGINT) t(i)" (Only (5000 :: Int64)) (0 :: Int, []) \(count, acc) schemaPtr arrayPtr -> do+                        schema <- peek schemaPtr+                        arrowSchemaChildCount schema @?= 1+                        child <- peekElemOff (arrowSchemaChildren schema) 0 >>= peek+                        peekUtf8CString (arrowSchemaName child) >>= (@?= "íslenska_λ")+                        peekCString (arrowSchemaFormat child) >>= (@?= "l")+                        values <- readInt64Batch arrayPtr+                        pure (count + 1, values : acc)+                assertBool "more than one native batch" (batches > 1)+                concat (reverse chunks) @?= [if i `rem` 3 == 0 then Nothing else Just i | i <- [0 .. 4999]]+        , testCase "empty results retain the initial accumulator" $+            withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+                result <- foldArrow_ conn "SELECT 1::BIGINT WHERE false" (17 :: Int) \_ _ _ -> assertFailure "unexpected batch" >> pure 0+                result @?= 17+        , testCase "result metadata follows bound parameter types" $+            withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+                values <- foldArrow conn "SELECT ? AS value" (Only (5000000000 :: Int64)) [] \acc _ array -> do+                    batch <- readInt64Batch array+                    pure (acc <> batch)+                values @?= [Just 5000000000]+        , testCase "success releases the schema and every batch" $+            withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+                withReleaseCounters \schemaReleases arrayReleases observe -> do+                    batches <- foldArrow_ conn "SELECT i FROM range(5000) t(i)" (0 :: Int) \count schema array -> do+                        observe schema array+                        pure (count + 1)+                    readIORef schemaReleases >>= (@?= batches)+                    readIORef arrayReleases >>= (@?= batches)+                    assertConnectionUsable conn+        , testCase "consumers can release each schema and batch" $+            withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+                withReleaseCounters \schemaReleases arrayReleases observe -> do+                    batches <- foldArrow_ conn "SELECT i FROM range(5000) t(i)" (0 :: Int) \count schema array -> do+                        observe schema array+                        releaseArrowArray array+                        releaseArrowSchema schema+                        pure (count + 1)+                    assertBool "more than one native batch" (batches > 1)+                    readIORef schemaReleases >>= (@?= batches)+                    readIORef arrayReleases >>= (@?= batches)+                    assertConnectionUsable conn+        , testCase "moved objects survive query and connection close" $+            withReleaseCounters \schemaReleases arrayReleases observe -> do+                alloca \savedSchema -> alloca \savedArray ->+                    bracket_+                        ( do+                            fillBytes savedSchema 0 (sizeOf (undefined :: ArrowSchema))+                            fillBytes savedArray 0 (sizeOf (undefined :: ArrowArray))+                        )+                        (releaseArrowArray savedArray >> releaseArrowSchema savedSchema)+                        do+                            withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+                                foldArrow_ conn "SELECT 42::BIGINT AS id" () \() schemaPtr arrayPtr -> mask_ do+                                    observe schemaPtr arrayPtr+                                    schema <- peek schemaPtr+                                    array <- peek arrayPtr+                                    poke savedSchema schema+                                    poke schemaPtr schema{arrowSchemaRelease = nullFunPtr}+                                    poke savedArray array+                                    poke arrayPtr array{arrowArrayRelease = nullFunPtr}+                            readIORef schemaReleases >>= (@?= 0)+                            readIORef arrayReleases >>= (@?= 0)+                            schema <- peek savedSchema+                            child <- peekElemOff (arrowSchemaChildren schema) 0 >>= peek+                            peekCString (arrowSchemaName child) >>= (@?= "id")+                            readInt64Batch savedArray >>= (@?= [Just 42])+                readIORef schemaReleases >>= (@?= 1)+                readIORef arrayReleases >>= (@?= 1)+        , testCase "exceptions during consumption release each object once" $+            forM_ [False, True] \consumeSchema ->+                withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+                    withReleaseCounters \schemaReleases arrayReleases observe -> do+                        result <- try $ foldArrow_ conn "SELECT 42::BIGINT" () \() schema array -> do+                            observe schema array+                            releaseArrowArray array+                            when consumeSchema (releaseArrowSchema schema)+                            throwIO (userError "consumer failed after release")+                        case result of+                            Left (err :: IOException) -> assertBool "consumer exception" ("consumer failed" `Text.isInfixOf` Text.pack (show err))+                            Right () -> assertFailure "expected consumer exception"+                        readIORef schemaReleases >>= (@?= 1)+                        readIORef arrayReleases >>= (@?= 1)+                        assertConnectionUsable conn+        , testCase "callback exceptions release the active batch and schema" $+            withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+                withReleaseCounters \schemaReleases arrayReleases observe -> do+                    result <- try $ foldArrow_ conn "SELECT i FROM range(5000) t(i)" () \_ schema array -> do+                        observe schema array+                        throwIO (userError "Arrow callback failed")+                    case result of+                        Left (err :: IOException) -> assertBool "original error" ("Arrow callback failed" `Text.isInfixOf` Text.pack (show err))+                        Right () -> assertFailure "expected callback exception"+                    readIORef schemaReleases >>= (@?= 1)+                    readIORef arrayReleases >>= (@?= 1)+                    assertConnectionUsable conn+        , testCase "cancellation releases the active batch and schema" $+            withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+                withReleaseCounters \schemaReleases arrayReleases observe -> do+                    entered <- newEmptyMVar+                    blocked <- newEmptyMVar+                    finished <- newEmptyMVar+                    worker <- forkIO do+                        result <- try $ foldArrow_ conn "SELECT i FROM range(5000) t(i)" () \_ schema array -> do+                            observe schema array+                            putMVar entered ()+                            takeMVar blocked+                        putMVar finished (result :: Either SomeException ())+                    takeMVar entered+                    killThread worker+                    result <- takeMVar finished+                    case result of+                        Left err -> fromException err @?= Just ThreadKilled+                        Right () -> assertFailure "expected cancellation"+                    readIORef schemaReleases >>= (@?= 1)+                    readIORef arrayReleases >>= (@?= 1)+                    assertConnectionUsable conn+        , testCase "SQL errors leave the connection usable" $+            withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+                result <- try $ foldArrow_ conn "SELECT error('Arrow query failed')" () \_ _ _ -> assertFailure "unexpected batch"+                case result of+                    Left (err :: SQLError) -> assertBool "native error" ("Arrow query failed" `Text.isInfixOf` sqlErrorMessage err)+                    Right () -> assertFailure "expected query error"+                assertConnectionUsable conn+        , testCase "closing the connection stops further batch callbacks" $+            withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+                withReleaseCounters \schemaReleases arrayReleases observe -> do+                    callbacks <- newIORef (0 :: Int)+                    result <- try $ foldArrow_ conn "SELECT i FROM range(5000) t(i)" () \_ schema array -> do+                        observe schema array+                        modifyIORef' callbacks (+ 1)+                        close conn+                    case result of+                        Left (err :: SQLError) -> sqlErrorMessage err @?= "duckdb-simple: connection is closed"+                        Right () -> assertFailure "expected closed connection error"+                    readIORef callbacks >>= (@?= 1)+                    readIORef schemaReleases >>= (@?= 1)+                    readIORef arrayReleases >>= (@?= 1)+        ]+  where+    foldArrow :: (ToRow q) => Connection -> Query -> q -> a -> (a -> Ptr ArrowSchema -> Ptr ArrowArray -> IO a) -> IO a+    foldArrow = if streaming then Streaming.foldArrow else Arrow.foldArrow++    foldArrow_ :: Connection -> Query -> a -> (a -> Ptr ArrowSchema -> Ptr ArrowArray -> IO a) -> IO a+    foldArrow_ = if streaming then Streaming.foldArrow_ else Arrow.foldArrow_++-- | Copy a BIGINT Arrow column, including its validity bitmap.+readInt64Batch :: Ptr ArrowArray -> IO [Maybe Int64]+readInt64Batch arrayPtr = do+    array <- peek arrayPtr+    child <- peekElemOff (arrowArrayChildren array) 0 >>= peek+    validity <- peekElemOff (arrowArrayBuffers child) 0+    values <- peekElemOff (arrowArrayBuffers child) 1+    forM [0 .. fromIntegral (arrowArrayLength array) - 1] \row -> do+        let index = fromIntegral (arrowArrayOffset child) + row+        valid <-+            if validity == nullPtr+                then pure True+                else do+                    byte <- peekElemOff (castPtr validity :: Ptr Word8) (index `div` 8)+                    pure (testBit byte (index `rem` 8))+        if valid then Just <$> peekElemOff (castPtr values) index else pure Nothing++-- | Wrap real release callbacks to observe cleanup without changing ownership.+withReleaseCounters :: (IORef Int -> IORef Int -> (Ptr ArrowSchema -> Ptr ArrowArray -> IO ()) -> IO a) -> IO a+withReleaseCounters action = do+    schemaReleases <- newIORef 0+    arrayReleases <- newIORef 0+    originalSchema <- newIORef Nothing+    originalArray <- newIORef Nothing+    bracket+        ( wrapArrowSchemaRelease \ptr -> do+            modifyIORef' schemaReleases (+ 1)+            callback <- readIORef originalSchema+            maybe (assertFailure "missing schema release") (\release -> mkArrowSchemaRelease release ptr) callback+        )+        freeHaskellFunPtr+        \schemaCallback ->+            bracket+                ( wrapArrowArrayRelease \ptr -> do+                    modifyIORef' arrayReleases (+ 1)+                    callback <- readIORef originalArray+                    maybe (assertFailure "missing array release") (\release -> mkArrowArrayRelease release ptr) callback+                )+                freeHaskellFunPtr+                \arrayCallback -> do+                    let observe schemaPtr arrayPtr = do+                            schema <- peek schemaPtr+                            assertBool "schema must not have been released" (arrowSchemaRelease schema /= nullFunPtr)+                            when (arrowSchemaRelease schema /= schemaCallback) do+                                modifyIORef' originalSchema (const (Just (arrowSchemaRelease schema)))+                                poke schemaPtr schema{arrowSchemaRelease = schemaCallback}+                            array <- peek arrayPtr+                            modifyIORef' originalArray (const (Just (arrowArrayRelease array)))+                            poke arrayPtr array{arrowArrayRelease = arrayCallback}+                    action schemaReleases arrayReleases observe++-- | Check that no failed Arrow operation leaves the connection busy.+assertConnectionUsable :: Connection -> Assertion+assertConnectionUsable conn = (query_ conn "SELECT 42" :: IO [Only Int64]) >>= (@?= [Only 42])

@@ -0,0 +1,198 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-deprecations #-}++-- | Cancellation during native execution must finish before handles are released.+module CancellationTests (cancellationTests) where++import Control.Concurrent (MVar, ThreadId, forkIOWithUnmask, newEmptyMVar, putMVar, readMVar, rtsSupportsBoundThreads, threadDelay, throwTo, tryPutMVar, tryReadMVar)+import Control.Exception (AsyncException (..), SomeException, bracket, fromException, mask_, try)+import Control.Monad (unless, void)+import Data.IORef (atomicWriteIORef, newIORef, readIORef)+import Data.Int (Int64)+import Database.DuckDB.FFI (c_duckdb_interrupt, c_duckdb_query)+import Database.DuckDB.Simple+import Database.DuckDB.Simple.Arrow (foldArrow_)+import qualified Database.DuckDB.Simple.Deprecated.Streaming as Streaming+import Database.DuckDB.Simple.Internal (withConnectionHandle, withQueryCString, withResult)+import GHC.Conc (BlockReason (..), ThreadStatus (..), threadStatus)+import System.Timeout (timeout)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit++-- | Exercise each native execution entry point and two cancellation races.+cancellationTests :: TestTree+cancellationTests =+    testGroup "native cancellation" $+        [ testCase name $ withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+            assertBool "cancellation requires the threaded runtime" rtsSupportsBoundThreads+            entered <- newEmptyMVar+            createFunction conn "query_started" (signal entered >> pure (1 :: Int64))+            withCaller conn (pure ()) (run conn longQuery) \caller done -> do+                await "native query did not start" (readMVar entered)+                (_, sent) <- startCaller (throwTo caller UserInterrupt)+                outcome <- await "native query did not cancel" (readMVar done)+                assertAsync [UserInterrupt] outcome+                await "interrupt sender did not finish" (readMVar sent) >>= assertSucceeded+            assertReusable conn+        | (name, run) <-+            [ ("query_", \conn sql -> void (query_ conn sql :: IO [Only Double]))+            , ("query with a prepared statement", \conn sql -> void (query conn sql () :: IO [Only Double]))+            , ("execute_", \conn sql -> void (execute_ conn sql))+            , ("execute with a prepared statement", \conn sql -> void (execute conn sql ()))+            , ("statement cursor", \conn sql -> withStatement conn sql \stmt -> void (nextRow stmt :: IO (Maybe (Only Double))))+            , ("fold_", \conn sql -> void (fold_ conn sql (0 :: Double) (\acc (Only value) -> pure (acc + value))))+            , ("Arrow fold", \conn sql -> foldArrow_ conn sql () (\() _ _ -> pure ()))+            ]+        ]+            <> [ testCase name $ withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+                    void (execute_ conn "SET streaming_buffer_size = '64KB'")+                    filtering <- newIORef False+                    entered <- newEmptyMVar+                    createFunction conn "discard_remaining" \(_ :: Int64) -> do+                        active <- readIORef filtering+                        if active then signal entered >> pure True else pure False+                    let sql = "SELECT i FROM range(1000000000000) t(i) WHERE NOT discard_remaining(i)"+                    withCaller conn (pure ()) (run conn sql (atomicWriteIORef filtering True)) \caller done -> do+                        await "fetch did not start after the first delivered batch" (readMVar entered)+                        (_, sent) <- startCaller (throwTo caller UserInterrupt)+                        await "streaming fetch did not cancel" (readMVar done) >>= assertAsync [UserInterrupt]+                        await "interrupt sender did not finish" (readMVar sent) >>= assertSucceeded+                    assertReusable conn+               | (name, run) <-+                    [ ("cancel a native row fetch", \conn sql delivered -> Streaming.fold_ conn sql () (\() (Only (_ :: Int64)) -> delivered))+                    , ("cancel a native Arrow fetch", \conn sql delivered -> Streaming.foldArrow_ conn sql () (\() _ _ -> delivered))+                    ,+                        ( "mixed cursor entry points retain interruptible fetching"+                        , \conn sql delivered -> withStatement conn sql \stmt -> do+                            Streaming.nextRow stmt >>= (@?= Just (Only (0 :: Int64)))+                            delivered+                            let consume =+                                    nextRow stmt >>= \case+                                        Nothing -> pure ()+                                        Just (Only (_ :: Int64)) -> consume+                            consume+                        )+                    ]+               ]+            <> [ testCase "cancellation before native entry is not cleared by startup" $+                    withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+                        queued <- newEmptyMVar+                        begin <- newEmptyMVar+                        createFunction conn "query_started" (pure 1 :: IO Int64)+                        let action = withConnectionHandle conn \handle ->+                                withQueryCString longQuery \sql ->+                                    withResult+                                        conn+                                        longQuery+                                        (\result -> signal queued >> readMVar begin >> c_duckdb_query handle sql result)+                                        (const (pure ()))+                        withCaller conn (signal begin) action \caller done -> do+                            await "query worker did not reach its entry gate" (readMVar queued)+                            (_, sent) <- startCaller (throwTo caller UserInterrupt)+                            await "first interrupt was not delivered" (readMVar sent) >>= assertSucceeded+                            awaitCleanup caller+                            signal begin+                            await "startup cleared the cancellation" (readMVar done) >>= assertAsync [UserInterrupt]+                        assertReusable conn+               , testCase "a second cancellation waits for a blocked callback to return" $+                    withConnectionWithConfig ":memory:" [("threads", "1")] \conn -> do+                        entered <- newEmptyMVar+                        release <- newEmptyMVar+                        createFunction conn "blocked_callback" (signal entered >> readMVar release >> pure (1 :: Int64))+                        withCaller conn (signal release) (void (query_ conn "SELECT blocked_callback()" :: IO [Only Int64])) \caller done -> do+                            await "callback did not start" (readMVar entered)+                            (_, firstSent) <- startCaller (throwTo caller UserInterrupt)+                            await "first interrupt was not delivered" (readMVar firstSent) >>= assertSucceeded+                            secondAttempted <- newEmptyMVar+                            (_, secondSent) <- startCaller (signal secondAttempted >> throwTo caller ThreadKilled)+                            await "second interrupt sender did not start" (readMVar secondAttempted)+                            premature <- timeout 50000 (readMVar secondSent)+                            case premature of+                                Nothing -> pure ()+                                Just _ -> assertFailure "second interrupt escaped cleanup"+                            tryReadMVar done >>= \case+                                Nothing -> pure ()+                                Just _ -> assertFailure "query returned while its callback was still running"+                            signal release+                            await "callback cancellation did not finish" (readMVar done) >>= assertAsync [UserInterrupt, ThreadKilled]+                            await "second interrupt sender did not finish" (readMVar secondSent) >>= assertSucceeded+                        assertReusable conn+               ]++-- | Run one volatile callback before a CPU query that cannot finish in the test window.+longQuery :: Query+longQuery =+    "WITH started AS MATERIALIZED (SELECT query_started() AS seed) \+    \SELECT sum(sin((a.i + b.j + started.seed)::DOUBLE)) \+    \FROM started, range(1000000) a(i), range(1000000) b(j)"++-- | Publish a thread's outcome while asynchronous exceptions are masked.+startCaller :: IO () -> IO (ThreadId, MVar (Either SomeException ()))+startCaller action = mask_ do+    done <- newEmptyMVar+    caller <- forkIOWithUnmask \unmask -> try (unmask action) >>= putMVar done+    pure (caller, done)++-- | Release gates and interrupt unfinished SQL before the connection can close.+withCaller :: Connection -> IO () -> IO () -> (ThreadId -> MVar (Either SomeException ()) -> IO a) -> IO a+withCaller conn release action use =+    bracket (startCaller action) cleanup (uncurry use)+  where+    cleanup (_, done) = do+        release+        await "emergency query cleanup did not finish" (stop done)+    stop done = do+        finished <- tryReadMVar done+        case finished of+            Just _ -> pure ()+            Nothing -> do+                withConnectionHandle conn c_duckdb_interrupt+                threadDelay 10000+                stop done++-- | Wait until the cancelled caller has reached its interrupt-and-join loop.+awaitCleanup :: ThreadId -> IO ()+awaitCleanup caller = awaitCondition "caller did not enter cancellation cleanup" do+    threadStatus caller >>= \case+        ThreadBlocked BlockedOnForeignCall -> pure False+        ThreadBlocked _ -> pure True+        ThreadFinished -> assertFailure "caller finished before the native worker" >> pure False+        ThreadDied -> assertFailure "caller died before publishing its result" >> pure False+        ThreadRunning -> pure False++-- | Bound waits in the test thread, independently of the cancelled caller.+await :: String -> IO a -> IO a+await message action = timeout 5000000 action >>= maybe (assertFailure message >> fail message) pure++-- | Wait for a thread state without relying on an arbitrary scheduling delay.+awaitCondition :: String -> IO Bool -> IO ()+awaitCondition message condition = await message loop+  where+    loop = do+        ready <- condition+        unless ready (threadDelay 1000 >> loop)++-- | Open a test gate once; cleanup can safely call this again.+signal :: MVar () -> IO ()+signal gate = void (tryPutMVar gate ())++-- | Require the requested asynchronous exception to reach the caller.+assertAsync :: [AsyncException] -> Either SomeException () -> Assertion+assertAsync expected = \case+    Left err -> case fromException err of+        Just actual -> assertBool ("unexpected asynchronous exception: " <> show actual) (actual `elem` expected)+        Nothing -> assertFailure ("expected asynchronous cancellation, got " <> show err)+    Right () -> assertFailure "query completed instead of reporting cancellation"++-- | Check that an interrupt sender completed normally.+assertSucceeded :: Either SomeException () -> Assertion+assertSucceeded = either (assertFailure . show) pure++-- | The interrupted native query must have released its connection state.+assertReusable :: Connection -> Assertion+assertReusable conn = do+    rows <- query_ conn "SELECT 42" :: IO [Only Int64]+    rows @?= [Only 42]

@@ -0,0 +1,153 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-deprecations #-}++-- | Regression tests for result metadata, cursor state, and handle lifetime.+module CoreRegressionTests (coreRegressionTests) where++import Control.Exception (SomeException, throwIO, try)+import Control.Monad (forM_, replicateM_, void)+import Data.IORef (readIORef)+import Data.Int (Int64)+import Data.List (isInfixOf)+import Data.Text (Text)+import Database.DuckDB.FFI.Deprecated (c_duckdb_result_is_streaming)+import Database.DuckDB.Simple+import Database.DuckDB.Simple.FromField (FieldValue)+import Database.DuckDB.Simple.Internal (Statement (statementStream), StatementStream (statementStreamResult), StatementStreamState (..))+import System.Mem (performMajorGC)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit++-- | Exercise public operations against an in-memory database.+coreRegressionTests :: TestTree+coreRegressionTests =+    testGroup+        "core regressions"+        [ testCase "stream metadata follows parameter types" $+            withConnection ":memory:" $ \conn -> do+                rows <- fold conn "SELECT ?" (Only (42 :: Int64)) [] collect+                rows @?= [Only (42 :: Int64)]+        , testCase "cursor uses the supported materialized execution API" $+            withConnection ":memory:" $ \conn ->+                withStatement conn "SELECT i FROM range(10000) t(i)" $ \stmt -> do+                    void (nextRow stmt :: IO (Maybe (Only Int64)))+                    state <- readIORef (statementStream stmt)+                    case state of+                        StatementStreamActive stream -> do+                            streaming <- c_duckdb_result_is_streaming (statementStreamResult stream)+                            assertBool "expected a materialized native result" (streaming == 0)+                        _ -> assertFailure "expected an active cursor"+        , testCase "stream metadata preserves 64-bit values" $+            withConnection ":memory:" $ \conn -> do+                rows <- fold conn "SELECT coalesce(?, 1) FROM range(3)" (Only (5000000000 :: Int64)) [] collect+                rows @?= replicate 3 (Only (5000000000 :: Int64))+        , testCase "stream numeric to Text conversion uses result metadata" $+            withConnection ":memory:" $ \conn -> do+                rows <- fold conn "SELECT coalesce(?, 'x')" (Only (100 :: Int64)) [] collect+                rows @?= [Only ("100" :: Text)]+        , testCase "stream metadata follows schema rebinding" $+            withConnection ":memory:" $ \conn -> do+                void $ execute_ conn "CREATE TABLE rebind_probe(x VARCHAR)"+                withStatement conn "SELECT x FROM rebind_probe" $ \stmt -> do+                    void $ execute_ conn "DROP TABLE rebind_probe"+                    void $ execute_ conn "CREATE TABLE rebind_probe(x BIGINT)"+                    void $ execute_ conn "INSERT INTO rebind_probe VALUES (5000000000)"+                    nextRow stmt >>= (@?= Just (Only (5000000000 :: Int64)))+        , testCase "EOF remains exhausted" $+            withConnection ":memory:" $ \conn ->+                withStatement conn "SELECT 42" $ \stmt -> do+                    nextRow stmt >>= (@?= Just (Only (42 :: Int64)))+                    replicateM_ 3 $ nextRow stmt >>= (@?= (Nothing :: Maybe (Only Int64)))+        , testCase "DML cursor executes only once" $+            withConnection ":memory:" $ \conn -> do+                void $ execute_ conn "CREATE TABLE once_probe(x INTEGER)"+                withStatement conn "INSERT INTO once_probe VALUES (1)" $ \stmt ->+                    replicateM_ 3 $ void (nextRow stmt :: IO (Maybe (Only Int64)))+                query_ conn "SELECT count(*) FROM once_probe" >>= (@?= [Only (1 :: Int64)])+        , testCase "rebinding resets an exhausted cursor" $+            withConnection ":memory:" $ \conn ->+                withStatement conn "SELECT ?::BIGINT" $ \stmt -> do+                    bind stmt [toField (1 :: Int64)]+                    nextRow stmt >>= (@?= Just (Only (1 :: Int64)))+                    nextRow stmt >>= (@?= (Nothing :: Maybe (Only Int64)))+                    bind stmt [toField (2 :: Int64)]+                    nextRow stmt >>= (@?= Just (Only (2 :: Int64)))+        , testCase "clear bindings invalidates an active cursor" $+            withConnection ":memory:" $ \conn ->+                withStatement conn "SELECT ?::BIGINT FROM range(3)" $ \stmt -> do+                    bind stmt [toField (7 :: Int64)]+                    void (nextRow stmt :: IO (Maybe (Only Int64)))+                    clearStatementBindings stmt+                    assertThrows (nextRow stmt :: IO (Maybe (Only Int64)))+        , testCase "fetch errors are not EOF" $+            withConnectionWithConfig ":memory:" [("threads", "1")] $ \conn ->+                assertThrows (fold_ conn "SELECT CASE WHEN i = 5000 THEN error('late failure') ELSE i END FROM range(10000) t(i)" (0 :: Int64) (\acc (Only n) -> pure (acc + n)))+        , testCase "composite cursors preserve values after chunk cleanup" $+            withConnectionWithConfig ":memory:" [("threads", "1")] $ \conn ->+                forM_+                    [ "SELECT {'n': i, 'text': 'λ' || i::VARCHAR} FROM range(5000) t(i)"+                    , "SELECT CASE WHEN i % 2 = 0 THEN union_value(n := i)::UNION(n BIGINT, text VARCHAR) ELSE union_value(text := i::VARCHAR)::UNION(n BIGINT, text VARCHAR) END FROM range(5000) t(i)"+                    , "SELECT CASE WHEN i % 3 = 0 THEN NULL ELSE {'n': i, 'list': [i, NULL], 'union': union_value(v := i)} END FROM range(5000) t(i)"+                    ]+                    $ \sql -> do+                        eager <- query_ conn sql :: IO [Only FieldValue]+                        streamed <- reverse <$> fold_ conn sql [] (\acc row -> pure (row : acc))+                        streamed @?= eager+        , testCase "composite cursor failures release the result" $+            withConnectionWithConfig ":memory:" [("threads", "1")] $ \conn -> do+                let sql = "SELECT {'n': i, 'values': [i, NULL]} FROM range(5000) t(i)"+                assertThrows (fold_ conn sql () (\() (Only (_ :: Int64)) -> pure ()))+                assertThrows $ fold_ conn sql (0 :: Int) $ \n (Only (_ :: FieldValue)) ->+                    if n == 2050 then throwIO (userError "composite fold failure") else pure (n + 1)+                query_ conn "SELECT 42" >>= (@?= [Only (42 :: Int64)])+        , testCase "closed connection rejects active cursor reads" $ do+            conn <- open ":memory:"+            stmt <- openStatement conn "SELECT i FROM range(3) t(i)"+            void (nextRow stmt :: IO (Maybe (Only Int64)))+            close conn+            assertThrows (nextRow stmt :: IO (Maybe (Only Int64)))+            closeStatement stmt+        , testCase "NUL in SQL is rejected" $+            withConnection ":memory:" $ \conn ->+                assertThrows (query_ conn "SELECT 1\0 + 2" :: IO [Only Int64])+        , testCase "NUL in database path is rejected" $+            assertThrows (withConnection ":memory:\0suffix" (const (pure ())))+        , testCase "NUL in configuration value is rejected" $+            assertThrows (withConnectionWithConfig ":memory:" [("threads", "1\0suffix")] (const (pure ())))+        , testCase "NUL in parameter name is rejected" $+            withConnection ":memory:" $ \conn ->+                withStatement conn "SELECT $value" $ \stmt ->+                    assertThrows (namedParameterIndex stmt "value\0suffix")+        , testCase "duplicate named bindings are rejected" $+            withConnection ":memory:" $ \conn ->+                withStatement conn "SELECT $a + $b" $ \stmt -> do+                    result <- try (bindNamed stmt ["a" := (1 :: Int64), "$a" := (2 :: Int64)])+                    case result of+                        Left (_ :: FormatError) -> pure ()+                        Right () -> assertFailure "expected duplicate binding error before execution"+        , testCase "transaction preserves the original exception if rollback fails" $+            withConnection ":memory:" $ \conn -> do+                result <- try $ withTransaction conn $ do+                    void $ execute_ conn "ROLLBACK"+                    throwIO (userError "original transaction failure")+                case result of+                    Left (err :: IOError) -> assertBool "original exception" ("original transaction failure" `isInfixOf` show err)+                    Right () -> assertFailure "expected transaction failure"+        , testCase "GC during the last connection use keeps native owners alive" $ do+            -- A bracket cleanup would retain conn and hide early finalization.+            conn <- open ":memory:"+            createFunction conn "gc_identity" (\(n :: Int64) -> performMajorGC >> pure n)+            rows <- query_ conn "SELECT gc_identity(i) FROM range(10) t(i)"+            rows @?= map Only ([0 .. 9] :: [Int64])+        ]+  where+    collect acc row = pure (acc <> [row])++-- | Require an exception without constraining its concrete type.+assertThrows :: IO a -> Assertion+assertThrows action = do+    result <- try (void action)+    case result of+        Left (_ :: SomeException) -> pure ()+        Right () -> assertFailure "expected an exception"

@@ -0,0 +1,304 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Regression tests for extension callbacks and native handle ownership.+module ExtensionRegressionTests (main, extensionRegressionTests) where++import Control.Exception (AsyncException (ThreadKilled), ErrorCall, Exception (..), SomeException, bracket, throwIO, try)+import Control.Monad (forM_, replicateM_, void)+import qualified Data.ByteString as BS+import Data.IORef (modifyIORef', newIORef, readIORef)+import Data.Int (Int64)+import Data.Proxy (Proxy (..))+import qualified Data.Text as Text+import Data.Word (Word64)+import Database.DuckDB.FFI+import Database.DuckDB.Simple+import qualified Database.DuckDB.Simple.Catalog as Catalog+import qualified Database.DuckDB.Simple.Config as Config+import qualified Database.DuckDB.Simple.Copy as Copy+import qualified Database.DuckDB.Simple.FileSystem as FileSystem+import qualified Database.DuckDB.Simple.Logging as Logging+import System.Directory (removeFile)+import System.IO (hClose, openBinaryTempFile)+import Test.Tasty (TestTree, defaultMain, testGroup)+import Test.Tasty.HUnit (Assertion, assertBool, assertFailure, testCase, (@?=))++-- | Run the extension regression tests as a standalone executable.+main :: IO ()+main = defaultMain extensionRegressionTests++-- | Check callback failures, value conversion, and resource cleanup.+extensionRegressionTests :: TestTree+extensionRegressionTests =+    testGroup+        "extension regressions"+        [ testCase "scalar closed connection cleanup" do+            conn <- open ":memory:"+            close conn+            replicateM_ 32 $ assertSqlError $ createFunction conn "closed_scalar" (1 :: Int64)+        , testCase "stateful closed connection cleanup" do+            conn <- open ":memory:"+            close conn+            replicateM_ 32 $ assertSqlError $ createFunctionWithState conn "closed_state" (pure ()) (\() -> (1 :: Int64))+        , testCase "COPY closed connection cleanup" do+            conn <- open ":memory:"+            close conn+            replicateM_ 32 $ assertSqlError $ registerCopy conn "closed_copy" Nothing+        , testCase "logging closed connection cleanup" do+            conn <- open ":memory:"+            close conn+            replicateM_ 32 $ assertSqlError $ Logging.registerLogStorage conn "closed_log" (\_ -> pure ())+        , testCase "scalar replacement registration cleanup" $+            withConnection ":memory:" \conn -> do+                createFunction conn "duplicate_scalar" (42 :: Int64)+                replicateM_ 32 $ createFunction conn "duplicate_scalar" (7 :: Int64)+                query_ conn "SELECT duplicate_scalar()" >>= (@?= [Only (7 :: Int64)])+        , testCase "stateful replacement registration cleanup" $+            withConnection ":memory:" \conn -> do+                createFunctionWithState conn "duplicate_state" (pure ()) (\() -> (42 :: Int64))+                replicateM_ 32 $ createFunctionWithState conn "duplicate_state" (pure ()) (\() -> (7 :: Int64))+                query_ conn "SELECT duplicate_state()" >>= (@?= [Only (7 :: Int64)])+        , testCase "scalar invalid signature registration cleanup" $+            withConnection ":memory:" \conn ->+                replicateM_ 32 $ assertSqlError $ createFunction conn "invalid_scalar" InvalidSignature+        , testCase "stateful invalid signature registration cleanup" $+            withConnection ":memory:" \conn ->+                replicateM_ 32 $ assertSqlError $ createFunctionWithState conn "invalid_state" (pure ()) (\() -> InvalidSignature)+        , testCase "logging duplicate registration cleanup" $+            withConnection ":memory:" \conn -> do+                Logging.registerLogStorage conn "duplicate_log" (\_ -> pure ())+                replicateM_ 32 $ assertSqlError $ Logging.registerLogStorage conn "duplicate_log" (\_ -> pure ())+                query_ conn "SELECT 42" >>= (@?= [Only (42 :: Int64)])+        , testCase "empty registration names fail safely" $+            withConnection ":memory:" \conn -> replicateM_ 32 do+                assertSqlError $ createFunction conn "" (1 :: Int64)+                assertSqlError $ createFunctionWithState conn "" (pure ()) (\() -> (1 :: Int64))+                assertSqlError $ registerCopy conn "" Nothing+                assertSqlError $ Logging.registerLogStorage conn "" (\_ -> pure ())+        , testCase "NUL registration names fail safely" $+            withConnection ":memory:" \conn -> do+                assertSqlError $ createFunction conn "nul\0scalar" (1 :: Int64)+                assertSqlError $ createFunctionWithState conn "nul\0state" (pure ()) (\() -> (1 :: Int64))+                assertSqlError $ registerCopy conn "nul\0copy" Nothing+                assertSqlError $ Logging.registerLogStorage conn "nul\0log" (\_ -> pure ())+        , testCase "partial scalar setup cleanup" $+            withConnection ":memory:" \conn -> replicateM_ 32 do+                outcome <- try (createFunction conn "broken_signature" BrokenSignature) :: IO (Either ErrorCall ())+                case outcome of+                    Left _ -> pure ()+                    Right () -> assertFailure "expected signature exception"+        , testCase "scalar exceptions become SQL errors" $+            withConnection ":memory:" \conn -> do+                createFunction conn "throw_scalar" (throwIO (userError "scalar callback failure") :: IO Int64)+                assertSqlErrorContaining "scalar callback failure" (query_ conn "SELECT throw_scalar()" :: IO [Only Int64])+                query_ conn "SELECT 42" >>= (@?= [Only (42 :: Int64)])+        , testCase "exception rendering cannot escape scalar callback" $+            withConnection ":memory:" \conn -> do+                createFunction conn "throw_bad_display" (throwIO BrokenDisplay :: IO Int64)+                assertSqlErrorContaining "Haskell callback failed" (query_ conn "SELECT throw_bad_display()" :: IO [Only Int64])+        , testCase "scalar async exceptions become SQL errors" $+            withConnection ":memory:" \conn -> do+                createFunction conn "throw_async_scalar" (throwIO ThreadKilled :: IO Int64)+                assertSqlErrorContaining "thread killed" (query_ conn "SELECT throw_async_scalar()" :: IO [Only Int64])+        , testCase "scalar initializer exceptions become SQL errors" $+            withConnection ":memory:" \conn -> do+                createFunctionWithState conn "throw_init" (throwIO (userError "state init failure") :: IO ()) (\() -> (1 :: Int64))+                assertSqlErrorContaining "state init failure" (query_ conn "SELECT throw_init()" :: IO [Only Int64])+        , testCase "scalar initializer async exceptions become SQL errors" $+            withConnection ":memory:" \conn -> do+                createFunctionWithState conn "throw_async_init" (throwIO ThreadKilled :: IO ()) (\() -> (1 :: Int64))+                assertSqlErrorContaining "thread killed" (query_ conn "SELECT throw_async_init()" :: IO [Only Int64])+        , testCase "stateful scalar exceptions become SQL errors" $+            withConnection ":memory:" \conn -> do+                createFunctionWithState conn "throw_state" (pure ()) (\() -> throwIO (userError "state execution failure") :: IO Int64)+                assertSqlErrorContaining "state execution failure" (query_ conn "SELECT throw_state()" :: IO [Only Int64])+        , testCase "Float scalar results preserve NaN infinity and negative zero" $+            withConnection ":memory:" \conn ->+                forM_+                    [ (0 / 0, isNaN)+                    , (1 / 0, \value -> isInfinite value && value > 0)+                    , (-1 / 0, \value -> isInfinite value && value < 0)+                    , (negate 0, isNegativeZero)+                    ]+                    \(value, matches) -> do+                        createFunction conn "float_value" (value :: Float)+                        createFunctionWithState conn "float_state" (pure ()) (\() -> value)+                        [(plain, stateful)] <- query_ conn "SELECT float_value(), float_state()" :: IO [(Double, Double)]+                        assertBool "plain callback preserves the floating-point value" (matches plain)+                        assertBool "stateful callback preserves the floating-point value" (matches stateful)+        , testCase "Word64 scalar preserves unsigned maximum" $+            withConnection ":memory:" \conn -> do+                createFunction conn "word64_identity" (id :: Word64 -> Word64)+                query_ conn "SELECT word64_identity(18446744073709551615::UBIGINT)" >>= (@?= [Only (maxBound :: Word64)])+                query_ conn "SELECT typeof(word64_identity(0::UBIGINT))" >>= (@?= [Only ("UBIGINT" :: Text.Text)])+        , testCase "Word scalar preserves unsigned maximum" $+            withConnection ":memory:" \conn -> do+                createFunction conn "word_maximum" (maxBound :: Word)+                query_ conn "SELECT word_maximum()" >>= (@?= [Only (maxBound :: Word)])+        , testCase "signed scalar preserves limits" $+            withConnection ":memory:" \conn -> do+                createFunction conn "int64_identity" (id :: Int64 -> Int64)+                query_ conn "SELECT int64_identity(x) FROM (VALUES ((-9223372036854775808)::BIGINT), (9223372036854775807::BIGINT)) t(x)"+                    >>= (@?= [Only (minBound :: Int64), Only maxBound])+        , testCase "scalar strings preserve UTF8 and embedded NUL" $+            withConnection ":memory:" \conn -> do+                let payload = "Matti \955\0 suffix" :: Text.Text+                createFunction conn "text_payload" payload+                createFunction conn "text_identity" (id :: Text.Text -> Text.Text)+                query_ conn "SELECT text_identity(text_payload())" >>= (@?= [Only payload])+        , testCase "scalar NULL arguments reach Maybe functions" $+            withConnection ":memory:" \conn -> do+                createFunction conn "null_default" (\(x :: Maybe Int64) -> maybe 42 id x)+                query_ conn "SELECT null_default(NULL::BIGINT)" >>= (@?= [Only (42 :: Int64)])+        , testCase "scalar nullable strings across chunks" $+            withConnection ":memory:" \conn -> do+                createFunction conn "optional_text" (\(x :: Int64) -> if even x then Just ("\955\0" <> Text.replicate 32 "x") else Nothing)+                query_ conn "SELECT count(*), count(optional_text(i)), min(length(optional_text(i))) FROM range(5000) t(i)"+                    >>= (@?= [(5000 :: Int64, 2500 :: Int64, Just (34 :: Int64))])+        , testCase "scalar DECIMAL inputs use supported numeric casts" $+            withConnection ":memory:" \conn -> do+                createFunction conn "double_identity" (id :: Double -> Double)+                query_ conn "SELECT double_identity(123.25::DECIMAL(10,2))" >>= (@?= [Only (123.25 :: Double)])+        , testCase "COPY exceptions become SQL errors in every phase" $+            withConnection ":memory:" \conn ->+                mapM_+                    ( \phase -> do+                        let name = "copy_fail_" <> Text.pack (show phase)+                        registerCopy conn name (Just phase)+                        assertSqlErrorContaining (Text.pack ("copy phase " <> show phase)) $+                            execute_ conn (Query ("COPY (SELECT 1) TO '/tmp/duckdb-simple-regression-copy' (FORMAT " <> name <> ")"))+                    )+                    [0 .. 3]+        , testCase "COPY state is valid across repeated queries" $+            withConnection ":memory:" \conn -> do+                registerCopy conn "copy_repeat" Nothing+                replicateM_ 32 $ void $ execute_ conn "COPY (SELECT * FROM range(5000)) TO '/tmp/duckdb-simple-regression-copy' (FORMAT copy_repeat)"+        , testCase "logging exceptions stay inside callback" $+            withConnection ":memory:" \conn -> do+                calls <- newIORef (0 :: Int)+                Logging.registerLogStorage conn "throwing_log" \_ -> do+                    modifyIORef' calls (+ 1)+                    throwIO (userError "log callback failure")+                void $ execute_ conn "SET logging_storage = 'throwing_log'"+                void $ execute_ conn "SET enable_logging = true"+                void $ execute_ conn "SELECT write_log('callback regression', level := 'INFO', scope := 'connection')"+                readIORef calls >>= \n -> assertBool "log callback ran" (n > 0)+                query_ conn "SELECT 42" >>= (@?= [Only (42 :: Int64)])+        , testCase "logging messages preserve UTF8" $+            withConnection ":memory:" \conn -> do+                messages <- newIORef []+                Logging.registerLogStorage conn "unicode_log" \entry ->+                    modifyIORef' messages (Logging.logEntryMessage entry :)+                void $ execute_ conn "SET logging_storage = 'unicode_log'"+                void $ execute_ conn "SET enable_logging = true"+                void $ execute_ conn "SELECT write_log('\955', level := 'INFO', scope := 'connection')"+                readIORef messages >>= \entries -> assertBool "Unicode log message" ("\955" `elem` entries)+        , testCase "filesystem action exceptions release handles" $+            withTemporaryFile \path ->+                withConnection ":memory:" \conn -> do+                    outcome <- try $ FileSystem.withFileHandle conn path [DuckDBFileFlagWrite] \_ -> throwIO (userError "file action failure")+                    case outcome :: Either SomeException () of+                        Left _ -> pure ()+                        Right () -> assertFailure "expected file action exception"+                    FileSystem.withFileHandle conn path [DuckDBFileFlagWrite] \handle -> do+                        FileSystem.writeFileHandleBytes handle (BS.pack [0, 255]) >>= (@?= 2)+                        FileSystem.fileHandleSync handle+                    FileSystem.withFileHandle conn path [DuckDBFileFlagRead] \handle -> do+                        FileSystem.readFileHandleChunk handle 0 >>= (@?= BS.empty)+                        FileSystem.readFileHandleChunk handle 8 >>= (@?= BS.pack [0, 255])+        , testCase "filesystem errors remain controlled" $+            withConnection ":memory:" \conn -> do+                assertSqlError $ FileSystem.withFileHandle conn "/tmp/duckdb-review-missing-dir/no-file" [DuckDBFileFlagRead] (\_ -> pure ())+                withTemporaryFile \path ->+                    assertSqlError $ FileSystem.withFileHandle conn (path <> "\0suffix") [DuckDBFileFlagRead] (\_ -> pure ())+        , testCase "catalog unsupported kinds fail safely" $+            withConnection ":memory:" \conn ->+                withTransaction conn $+                    mapM_+                        (\kind -> assertSqlError $ Catalog.lookupCatalogEntry conn "memory" "main" "probe" kind)+                        [DuckDBCatalogEntryTypeInvalid, DuckDBCatalogEntryTypeSchema, DuckDBCatalogEntryTypePreparedStatement, DuckDBCatalogEntryTypeDatabase, DuckDBCatalogEntryType 99]+        , testCase "catalog and config NUL names fail safely" $+            withConnection ":memory:" \conn -> do+                assertSqlError $ Config.getConfigOption conn "threads\0suffix"+                withTransaction conn do+                    assertSqlError $ Catalog.catalogTypeName conn "memory\0suffix"+                    assertSqlError $ Catalog.lookupCatalogEntry conn "memory" "main\0suffix" "probe" DuckDBCatalogEntryTypeTable+        , testCase "catalog names preserve UTF8" $+            withConnection ":memory:" \conn -> do+                void $ execute_ conn "CREATE TABLE \"\955\" (i INT)"+                withTransaction conn do+                    entry <- Catalog.lookupCatalogEntry conn "memory" "main" "\955" DuckDBCatalogEntryTypeTable+                    fmap Catalog.catalogEntryName entry @?= Just "\955"+        , testCase "catalog and config missing values remain safe" $+            withConnection ":memory:" \conn -> do+                Config.getConfigOption conn "no_such_config_option" >>= (@?= Nothing)+                withTransaction conn do+                    Catalog.catalogTypeName conn "no_such_catalog" >>= (@?= Nothing)+                    Catalog.lookupCatalogEntry conn "memory" "main" "no_such_table" DuckDBCatalogEntryTypeTable >>= (@?= Nothing)+        ]++-- | Raise a second exception when the callback error is rendered.+data BrokenDisplay = BrokenDisplay+    deriving (Show)++instance Exception BrokenDisplay where+    displayException _ = error "broken error rendering"++-- | Force a setup exception after scalar callbacks have been acquired.+data BrokenSignature = BrokenSignature++instance Function BrokenSignature where+    argumentTypes _ = error "broken scalar signature"+    returnType _ = error "unused return type"+    isVolatile _ = False+    applyFunction _ _ = error "unused function"++-- | Supply an argument type which DuckDB rejects at registration.+data InvalidSignature = InvalidSignature++instance Function InvalidSignature where+    argumentTypes _ = [DuckDBTypeInvalid]+    returnType _ = returnType (Proxy :: Proxy Int64)+    isVolatile _ = False+    applyFunction _ _ = error "unregistered function"++-- | Register a COPY callback and optionally fail one phase.+registerCopy :: Connection -> Text.Text -> Maybe Int -> IO ()+registerCopy conn name failingPhase =+    Copy.registerCopyToFunction+        conn+        name+        (\_ -> failPhase 0)+        (\info -> (Copy.copyInitBindState info @?= ()) >> failPhase 1)+        (\info _ -> (Copy.copySinkGlobalState info @?= ()) >> failPhase 2)+        (\info -> (Copy.copyFinalizeGlobalState info @?= ()) >> failPhase 3)+  where+    failPhase phase =+        if failingPhase == Just phase+            then throwIO (userError ("copy phase " <> show phase))+            else pure ()++-- | Require a controlled SQL exception.+assertSqlError :: IO a -> Assertion+assertSqlError = assertSqlErrorContaining Text.empty++-- | Require a SQL exception which includes the expected message.+assertSqlErrorContaining :: Text.Text -> IO a -> Assertion+assertSqlErrorContaining expected action = do+    outcome <- try (void action)+    case outcome of+        Left (err :: SQLError) ->+            assertBool ("expected SQL error containing " <> show expected <> ", got " <> show err) $+                expected `Text.isInfixOf` sqlErrorMessage err+        Right () -> assertFailure "expected SQL error"++-- | Remove a temporary file after the action.+withTemporaryFile :: (FilePath -> IO a) -> IO a+withTemporaryFile action =+    bracket+        (do (path, handle) <- openBinaryTempFile "/tmp" "duckdb-extension-regression"; hClose handle; pure path)+        removeFile+        action

@@ -128,8 +128,8 @@     , SomeRoundTrip "Word64" (Proxy :: Proxy Word64) (arbitrary :: Gen Word64) shrink     , SomeRoundTrip "Float" (Proxy :: Proxy Float) genFiniteFloat shrink     , SomeRoundTrip "Double" (Proxy :: Proxy Double) genFiniteDouble shrink-    , SomeRoundTrip "String" (Proxy :: Proxy String) genStringNoNul shrinkStringNoNul-    , SomeRoundTrip "Text" (Proxy :: Proxy Text.Text) genTextNoNul shrinkTextNoNul+    , SomeRoundTrip "String" (Proxy :: Proxy String) genString shrinkString+    , SomeRoundTrip "Text" (Proxy :: Proxy Text.Text) genText shrinkText     , SomeRoundTrip "ByteString" (Proxy :: Proxy BS.ByteString) genByteString shrinkByteString     , SomeRoundTrip "BitString" (Proxy :: Proxy BitString) genBitString shrinkBitString     , SomeRoundTrip "UUID" (Proxy :: Proxy UUID) genUUID shrinkNone@@ -168,25 +168,24 @@     (arbitrary :: Gen Double)         `suchThat` \x -> not (isNaN x || isInfinite x) -genStringNoNul :: Gen String-genStringNoNul =+genString :: Gen String+genString =     sized \n -> do         len <- chooseInt (0, max 0 (min n 32))-        vectorOf len genCharNoNul+        vectorOf len genChar -shrinkStringNoNul :: String -> [String]-shrinkStringNoNul =-    filter (all (/= '\0')) . shrink+shrinkString :: String -> [String]+shrinkString = shrink -genCharNoNul :: Gen Char-genCharNoNul =-    (arbitrary :: Gen Char) `suchThat` (/= '\0')+genChar :: Gen Char+genChar =+    frequency [(1, pure '\0'), (9, arbitrary)] -genTextNoNul :: Gen Text.Text-genTextNoNul = Text.pack <$> genStringNoNul+genText :: Gen Text.Text+genText = Text.pack <$> genString -shrinkTextNoNul :: Text.Text -> [Text.Text]-shrinkTextNoNul txt = Text.pack <$> shrinkStringNoNul (Text.unpack txt)+shrinkText :: Text.Text -> [Text.Text]+shrinkText txt = Text.pack <$> shrinkString (Text.unpack txt)  genByteString :: Gen BS.ByteString genByteString =@@ -392,7 +391,7 @@     SimpleBool -> FieldBool <$> (arbitrary :: Gen Bool)     SimpleInt32 -> FieldInt32 <$> (arbitrary :: Gen Int32)     SimpleDouble -> FieldDouble <$> genFiniteDouble-    SimpleText -> FieldText <$> genTextNoNul+    SimpleText -> FieldText <$> genText     SimpleBlob -> FieldBlob <$> genByteString  timeOfDayToMicroseconds :: TimeOfDay -> Integer

@@ -12,9 +12,12 @@ -- | Tasty-based test suite for duckdb-simple. module Main (main) where +import ArrowTests (arrowTests)+import CancellationTests (cancellationTests) import Control.Applicative ((<|>)) import Control.Exception (ErrorCall, Exception, SomeException, displayException, fromException, try) import Control.Monad (forM_, replicateM_, when)+import CoreRegressionTests (coreRegressionTests) import Data.Array (Array, elems, listArray) import qualified Data.ByteString as BS import Data.IORef (atomicModifyIORef', newIORef, readIORef)@@ -76,14 +79,19 @@     UnionValue (..),  ) import Database.DuckDB.Simple.Ok (Ok (..))+import Database.DuckDB.Simple.Time (Unbounded (..))+import ExtensionRegressionTests (extensionRegressionTests) import GHC.Generics (Generic) import Numeric.Natural (Natural) import Properties (roundTripTests)+import StreamingTests (nativeStreamingTests) import System.Directory (doesFileExist, removeFile) import Test.Tasty (TestTree, defaultMain, testGroup) import Test.Tasty.ExpectedFailure (expectFailBecause) import Test.Tasty.HUnit import Test.Tasty.QuickCheck (testProperty, (===))+import TimeTests (timeTests)+import ValueRegressionTests (valueRegressionTests)  data Person = Person     { personId :: Int@@ -196,6 +204,13 @@     testGroup         "duckdb-simple"         [ connectionTests+        , coreRegressionTests+        , arrowTests+        , cancellationTests+        , nativeStreamingTests+        , extensionRegressionTests+        , valueRegressionTests+        , timeTests         , withConnectionTests         , statementTests         , roundTripTests@@ -311,9 +326,9 @@         [ testCase "returns the action result" $ do             result <- withConnection ":memory:" \_ -> pure (21 :: Int)             assertEqual "action result" 21 result-        , testCase "propagates exceptions from the action" $-            assertThrowsErrorCall $-                withConnection ":memory:" (\_ -> error "boom" :: IO ())+        , testCase "propagates exceptions from the action"+            $ assertThrowsErrorCall+            $ withConnection ":memory:" (\_ -> error "boom" :: IO ())         ]  statementTests :: TestTree@@ -585,8 +600,8 @@     , successCase "UBIGINT" (quotedValue uBigValue) "UBIGINT" (ExpectEquals (FieldWord64 uBigValue))     , successCase "FLOAT" (quoted "3.25") "FLOAT" (ExpectEquals (FieldFloat floatValue))     , successCase "DOUBLE" (quoted "2.5") "DOUBLE" (ExpectEquals (FieldDouble doubleValue))-    , successCase "TIMESTAMP" (quoted "2024-01-02 03:04:05.123456") "TIMESTAMP" (ExpectEquals (FieldTimestamp timestampLocalTimeMicros))-    , successCase "DATE" (quoted "2024-10-12") "DATE" (ExpectEquals (FieldDate dateValue))+    , successCase "TIMESTAMP" (quoted "2024-01-02 03:04:05.123456") "TIMESTAMP" (ExpectEquals (FieldTimestamp (Finite timestampLocalTimeMicros)))+    , successCase "DATE" (quoted "2024-10-12") "DATE" (ExpectEquals (FieldDate (Finite dateValue)))     , successCase "TIME" (quoted "14:30:15.123456") "TIME" (ExpectEquals (FieldTime timeValue))     , successCase "INTERVAL" (quoted "1 day") "INTERVAL" (ExpectEquals (FieldInterval intervalValue))     , successCase "HUGEINT" (quotedValue hugeIntLiteral) "HUGEINT" (ExpectEquals (FieldHugeInt hugeIntLiteral))@@ -594,14 +609,14 @@     , successCase "VARCHAR" (quoted varcharText) "VARCHAR" (ExpectEquals (FieldText varcharText))     , successCase "BLOB" (quoted "duckdb") "BLOB" (ExpectEquals (FieldBlob blobValue))     , successCase "DECIMAL" (quoted "12345.6789") "DECIMAL(18,4)" (ExpectEquals (FieldDecimal decimalValue))-    , successCase "TIMESTAMP_S" (quoted "2024-01-02 03:04:05") "TIMESTAMP_S" (ExpectEquals (FieldTimestamp timestampLocalTimeSeconds))-    , successCase "TIMESTAMP_MS" (quoted "2024-01-02 03:04:05.123") "TIMESTAMP_MS" (ExpectEquals (FieldTimestamp timestampLocalTimeMillis))-    , successCase "TIMESTAMP_NS" (quoted "2024-01-02 03:04:05.123456789") "TIMESTAMP_NS" (ExpectEquals (FieldTimestamp timestampLocalTimeNanos))+    , successCase "TIMESTAMP_S" (quoted "2024-01-02 03:04:05") "TIMESTAMP_S" (ExpectEquals (FieldTimestamp (Finite timestampLocalTimeSeconds)))+    , successCase "TIMESTAMP_MS" (quoted "2024-01-02 03:04:05.123") "TIMESTAMP_MS" (ExpectEquals (FieldTimestamp (Finite timestampLocalTimeMillis)))+    , successCase "TIMESTAMP_NS" (quoted "2024-01-02 03:04:05.123456789") "TIMESTAMP_NS" (ExpectEquals (FieldTimestamp (Finite timestampLocalTimeNanos)))     , successCase "ENUM" (quoted "beta") "ENUM('alpha','beta')" (ExpectEquals (FieldEnum 1))     , successCase "LIST" (quoted "[1,2,3]") "INTEGER[]" (ExpectEquals (FieldList listElements))     , successDirect "MAP" "MAP(['a','b'], [1,2])" (expectMapEntries mapPairs)     , successCase "TIMETZ" (quoted "03:04:05+02:30") "TIME WITH TIME ZONE" (ExpectEquals (FieldTimeTZ timeWithZoneValue))-    , successCase "TIMESTAMPTZ" (quoted "2024-01-02 03:04:05+02:30") "TIMESTAMP WITH TIME ZONE" (ExpectEquals (FieldTimestampTZ timestampTzUtc))+    , successCase "TIMESTAMPTZ" (quoted "2024-01-02 03:04:05+02:30") "TIMESTAMP WITH TIME ZONE" (ExpectEquals (FieldTimestampTZ (Finite timestampTzUtc)))     , successCase "BIGNUM" (quoted bigNumLiteralText) "BIGNUM" (ExpectEquals (FieldBigNum (BigNum bigNumLiteral)))     , successCase "UUID" (quoted $ UUID.toText uuid) "UUID" (ExpectEquals (FieldUUID uuid))     , successCase "BIT" (quoted $ Text.pack $ show bits) "BIT" (ExpectEquals (FieldBit bits))

@@ -0,0 +1,162 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-deprecations #-}++-- | Check the explicit native streaming API and its cursor ownership.+module StreamingTests (nativeStreamingTests) where++import Control.Exception (IOException, throwIO, try)+import Control.Monad (forM_, replicateM_, void, when)+import Data.IORef (atomicWriteIORef, newIORef, readIORef, writeIORef)+import Data.Int (Int64)+import Data.List (isInfixOf)+import Data.Text (Text)+import qualified Data.Text as Text+import Database.DuckDB.FFI (c_duckdb_vector_size)+import Database.DuckDB.FFI.Deprecated (c_duckdb_result_is_streaming)+import Database.DuckDB.Simple+import qualified Database.DuckDB.Simple.Deprecated.Streaming as Streaming+import Database.DuckDB.Simple.FromField (FieldValue)+import Database.DuckDB.Simple.Internal (Statement (statementStream), StatementStream (statementStreamResult), StatementStreamState (..))+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit++-- | Exercise native streaming, resets, metadata, and cleanup across chunks.+nativeStreamingTests :: TestTree+nativeStreamingTests =+    testGroup+        "explicit native streaming"+        [ testCase "nextRow starts a native streaming result" $ withDb \conn ->+            withStatement conn "SELECT i FROM range(10000) t(i)" \stmt -> do+                Streaming.nextRow stmt >>= (@?= Just (Only (0 :: Int64)))+                state <- readIORef (statementStream stmt)+                case state of+                    StatementStreamActive stream -> do+                        flag <- c_duckdb_result_is_streaming (statementStreamResult stream)+                        assertBool "expected a native streaming result" (flag /= 0)+                    _ -> assertFailure "expected an active cursor"+        , testCase "default execution remains materialized when cursor entry points change" $ withDb \conn ->+            withStatement conn "SELECT i FROM range(10000) t(i)" \stmt -> do+                nextRow stmt >>= (@?= Just (Only (0 :: Int64)))+                Streaming.nextRow stmt >>= (@?= Just (Only (1 :: Int64)))+                state <- readIORef (statementStream stmt)+                case state of+                    StatementStreamActive stream -> do+                        flag <- c_duckdb_result_is_streaming (statementStreamResult stream)+                        flag @?= 0+                    _ -> assertFailure "expected an active cursor"+        , testCase "nested values survive chunk cleanup" $ withDb \conn ->+            forM_+                [ "SELECT CASE WHEN i % 7 = 0 THEN NULL ELSE {'n': i, 'text': 'λ' || i::VARCHAR, 'list': [i, NULL], 'union': union_value(value := i)} END FROM range(5000) t(i)"+                , "SELECT [{'n': i, 'text': 'λ' || i::VARCHAR}, NULL] FROM range(5000) t(i)"+                , "SELECT map([i], [{'values': [i, NULL], 'union': union_value(value := i)}]) FROM range(5000) t(i)"+                , "SELECT CASE WHEN i % 2 = 0 THEN union_value(n := i)::UNION(n BIGINT, text VARCHAR) ELSE union_value(text := i::VARCHAR)::UNION(n BIGINT, text VARCHAR) END FROM range(5000) t(i)"+                ]+                \sql -> do+                    expected <- query_ conn sql :: IO [Only FieldValue]+                    actual <- reverse <$> Streaming.fold_ conn sql [] (\acc row -> pure (row : acc))+                    actual @?= expected+        , testCase "fold and foldNamed use executed parameter types" $ withDb \conn -> do+            positional <- Streaming.fold conn "SELECT coalesce(?, 'x') FROM range(3)" (Only (5000000000 :: Int64)) [] (\acc row -> pure (row : acc))+            positional @?= replicate 3 (Only (5000000000 :: Int64))+            named <- Streaming.foldNamed conn "SELECT $value FROM range(3)" ["value" := ("λ" :: Text)] [] (\acc row -> pure (row : acc))+            named @?= replicate 3 (Only ("λ" :: Text))+        , testCase "rebinding an active cursor replaces its metadata" $ withDb \conn ->+            withStatement conn "SELECT ? FROM range(3)" \stmt -> do+                bind stmt [toField (5000000000 :: Int64)]+                Streaming.nextRow stmt >>= (@?= Just (Only (5000000000 :: Int64)))+                bind stmt [toField ("λ" :: Text)]+                Streaming.nextRow stmt >>= (@?= Just (Only ("λ" :: Text)))+        , testCase "schema changes replace prepare-time metadata" $ withDb \conn -> do+            void $ execute_ conn "CREATE TABLE stream_rebind(x VARCHAR)"+            withStatement conn "SELECT x FROM stream_rebind" \stmt -> do+                void $ execute_ conn "DROP TABLE stream_rebind"+                void $ execute_ conn "CREATE TABLE stream_rebind(x BIGINT)"+                void $ execute_ conn "INSERT INTO stream_rebind VALUES (5000000000)"+                Streaming.nextRow stmt >>= (@?= Just (Only (5000000000 :: Int64)))+        , testCase "EOF persists until an explicit binding reset" $ withDb \conn ->+            withStatement conn "SELECT ?::BIGINT" \stmt -> do+                bind stmt [toField (1 :: Int64)]+                Streaming.nextRow stmt >>= (@?= Just (Only (1 :: Int64)))+                replicateM_ 3 $ Streaming.nextRow stmt >>= (@?= (Nothing :: Maybe (Only Int64)))+                bind stmt [toField (2 :: Int64)]+                Streaming.nextRow stmt >>= (@?= Just (Only (2 :: Int64)))+                Streaming.nextRow stmt >>= (@?= (Nothing :: Maybe (Only Int64)))+        , testCase "clearing bindings discards an active result" $ withDb \conn ->+            withStatement conn "SELECT ?::BIGINT FROM range(3)" \stmt -> do+                bind stmt [toField (7 :: Int64)]+                Streaming.nextRow stmt >>= (@?= Just (Only (7 :: Int64)))+                clearStatementBindings stmt+                void $ expectSqlError (Streaming.nextRow stmt :: IO (Maybe (Only Int64)))+                bind stmt [toField (8 :: Int64)]+                Streaming.nextRow stmt >>= (@?= Just (Only (8 :: Int64)))+        , testCase "a DML cursor does not execute again after EOF" $ withDb \conn -> do+            void $ execute_ conn "CREATE TABLE stream_once(x BIGINT)"+            withStatement conn "INSERT INTO stream_once VALUES (1)" \stmt ->+                replicateM_ 3 $ Streaming.nextRow stmt >>= (@?= (Nothing :: Maybe (Only Int64)))+            query_ conn "SELECT count(*) FROM stream_once" >>= (@?= [Only (1 :: Int64)])+        , testCase "nextRowWith runs the supplied parser" $ withDb \conn ->+            withStatement conn "SELECT 20, 22" \stmt ->+                Streaming.nextRowWith ((+) <$> field <*> field) stmt >>= (@?= Just (42 :: Int64))+        , testCase "a fetch failure follows an already delivered batch" $ withDb \conn -> do+            void $ execute_ conn "SET streaming_buffer_size = '64KB'"+            failNext <- newIORef False+            delivered <- newIORef (0 :: Int)+            batchSize <- fromIntegral <$> c_duckdb_vector_size+            createFunction conn "late_stream_error" \(value :: Int64) -> do+                shouldFail <- readIORef failNext+                when shouldFail (throwIO (userError "late stream error"))+                pure value+            err <- expectSqlError $ Streaming.fold_ conn "SELECT late_stream_error(i) FROM range(1000000) t(i)" (0 :: Int) \count (Only (_ :: Int64)) -> do+                let next = count + 1+                writeIORef delivered next+                when (next == batchSize) (atomicWriteIORef failNext True)+                pure next+            count <- readIORef delivered+            assertBool "the first batch must precede the error" (count >= batchSize)+            assertBool "fetch error must retain its message" ("late stream error" `Text.isInfixOf` sqlErrorMessage err)+            assertReusable conn+        , testCase "Arrow reports a fetch failure after a delivered batch" $ withDb \conn -> do+            void $ execute_ conn "SET streaming_buffer_size = '64KB'"+            failNext <- newIORef False+            createFunction conn "late_arrow_error" \(value :: Int64) -> do+                shouldFail <- readIORef failNext+                when shouldFail (throwIO (userError "late Arrow error"))+                pure value+            err <- expectSqlError $ Streaming.foldArrow_ conn "SELECT late_arrow_error(i) FROM range(1000000) t(i)" () \() _ _ ->+                atomicWriteIORef failNext True+            readIORef failNext >>= assertBool "a batch reached the callback"+            assertBool "fetch error must retain its message" ("late Arrow error" `Text.isInfixOf` sqlErrorMessage err)+            assertReusable conn+        , testCase "row decode failure exhausts and releases the cursor" $ withDb \conn -> do+            withStatement conn "SELECT {'n': i} FROM range(5000) t(i)" \stmt -> do+                void $ expectSqlError (Streaming.nextRow stmt :: IO (Maybe (Only Int64)))+                Streaming.nextRow stmt >>= (@?= (Nothing :: Maybe (Only Int64)))+            assertReusable conn+        , testCase "fold callback failure releases the active chunk" $ withDb \conn -> do+            outcome <- try $ Streaming.fold_ conn "SELECT {'n': i, 'list': [i, NULL]} FROM range(5000) t(i)" (0 :: Int) \count (Only (_ :: FieldValue)) ->+                if count == 2050 then throwIO (userError "stream step failure") else pure (count + 1)+            case outcome of+                Left (err :: IOException) -> assertBool "original callback error" ("stream step failure" `isInfixOf` show err)+                Right _ -> assertFailure "expected the callback to fail"+            assertReusable conn+        ]++-- | Run each case with one native worker so row order is stable.+withDb :: (Connection -> IO a) -> IO a+withDb = withConnectionWithConfig ":memory:" [("threads", "1")]++-- | Require a SQL or field conversion error and retain its diagnostic fields.+expectSqlError :: IO a -> IO SQLError+expectSqlError action = do+    outcome <- try action+    case outcome of+        Left err -> pure err+        Right _ -> assertFailure "expected SQLError" >> fail "expected SQLError"++-- | Verify that cleanup leaves the connection ready for another query.+assertReusable :: Connection -> Assertion+assertReusable conn = do+    rows <- query_ conn "SELECT 42" :: IO [Only Int64]+    rows @?= [Only 42]

@@ -0,0 +1,105 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingVia #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Round trips for finite dates and native temporal infinities.+module TimeTests (timeTests) where++import Control.Exception (try)+import Control.Monad (forM_)+import Data.Array (listArray)+import qualified Data.Text as Text+import Data.Time (Day, LocalTime (..), TimeOfDay (..), UTCTime, fromGregorian, localTimeToUTC, utc)+import Database.DuckDB.Simple+import Database.DuckDB.Simple.FromField (Field (..), FieldValue (..), returnError)+import Database.DuckDB.Simple.Generic (ViaDuckDB (..))+import Database.DuckDB.Simple.Time+import GHC.Generics (Generic)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit++-- | Record fields exercise the generic scalar and collection instances.+data TemporalRecord = TemporalRecord+    { dateValue :: Date+    , localValue :: LocalTimestamp+    , utcValue :: UTCTimestamp+    , optionalValue :: Maybe Date+    , dates :: [Date]+    }+    deriving stock (Eq, Show, Generic)+    deriving (DuckDBColumnType, ToField, FromField) via (ViaDuckDB TemporalRecord)++-- | Each constructor retains its temporal member type.+data TemporalSum = ADate Date | ALocal LocalTimestamp | AUtc UTCTimestamp | ADateNull (Maybe Date)+    deriving stock (Eq, Show, Generic)+    deriving (DuckDBColumnType, ToField, FromField) via (ViaDuckDB TemporalSum)++-- | A custom decoder must receive infinity before any finite conversion.+newtype InfinitySign = InfinitySign Int+    deriving (Eq, Show)++instance FromField InfinitySign where+    fromField f@Field{fieldValue} = case fieldValue of+        FieldDate NegInfinity -> pure (InfinitySign (-1))+        FieldDate (Finite _) -> pure (InfinitySign 0)+        FieldDate PosInfinity -> pure (InfinitySign 1)+        _ -> returnError Incompatible f "expected DATE"++-- | Exercise native binding, eager queries, cursors, and generic composites.+timeTests :: TestTree+timeTests =+    testGroup+        "temporal infinity"+        [ testCase "dates bind and decode with their SQL type" $ withConnection ":memory:" $ \conn ->+            forM_ [NegInfinity, Finite day, PosInfinity] $ \value -> do+                (query conn "SELECT ?, typeof(?)" (value, value) :: IO [(Date, String)]) >>= (@?= [(value, "DATE")])+                (query conn "SELECT ?::VARCHAR" (Only value) :: IO [Only String]) >>= (@?= [Only (dateText value)])+        , testCase "local timestamps bind and decode with their SQL type" $ withConnection ":memory:" $ \conn ->+            forM_ [NegInfinity, Finite local, PosInfinity] $ \value ->+                (query conn "SELECT ?, typeof(?)" (value, value) :: IO [(LocalTimestamp, String)]) >>= (@?= [(value, "TIMESTAMP")])+        , testCase "UTC timestamps preserve instants and infinities outside UTC" $ withConnection ":memory:" $ \conn -> do+            _ <- execute_ conn "SET TimeZone='Pacific/Auckland'"+            forM_ [NegInfinity, Finite instant, PosInfinity] $ \value ->+                (query conn "SELECT ?, typeof(?)" (value, value) :: IO [(UTCTimestamp, String)]) >>= (@?= [(value, "TIMESTAMP WITH TIME ZONE")])+        , testCase "infinity and NULL remain distinct" $ withConnection ":memory:" $ \conn -> do+            (query conn "SELECT ?::DATE, ?::TIMESTAMP, ?::TIMESTAMPTZ" (Nothing :: Maybe Date, Nothing :: Maybe LocalTimestamp, Nothing :: Maybe UTCTimestamp) :: IO [(Maybe Date, Maybe LocalTimestamp, Maybe UTCTimestamp)]) >>= (@?= [(Nothing, Nothing, Nothing)])+            (query_ conn "SELECT 'infinity'::DATE" :: IO [Only (Maybe Date)]) >>= (@?= [Only (Just PosInfinity)])+        , testCase "ordinary finite targets reject infinity with a conversion error" $ withConnection ":memory:" $ \conn -> do+            assertConversionError (query_ conn "SELECT 'infinity'::DATE" :: IO [Only Day])+            assertConversionError (query_ conn "SELECT '-infinity'::TIMESTAMP" :: IO [Only LocalTime])+            assertConversionError (query_ conn "SELECT 'infinity'::TIMESTAMPTZ" :: IO [Only UTCTime])+            assertConversionError (query_ conn "SELECT 'infinity'::TIMESTAMP" :: IO [Only TimeOfDay])+            (query_ conn "SELECT 42" :: IO [Only Int]) >>= (@?= [Only 42])+        , testCase "custom FromField receives native infinity" $ withConnection ":memory:" $ \conn ->+            (query_ conn "SELECT value FROM (VALUES ('-infinity'::DATE), (DATE '2000-01-02'), ('infinity'::DATE)) t(value) ORDER BY value" :: IO [Only InfinitySign]) >>= (@?= map (Only . InfinitySign) [-1, 0, 1])+        , testCase "cursors preserve infinities" $ withConnection ":memory:" $ \conn -> do+            values <- fold_ conn "SELECT value FROM (VALUES ('-infinity'::DATE), (DATE '2000-01-02'), ('infinity'::DATE)) t(value) ORDER BY value" [] (\acc (Only value) -> pure (value : acc))+            reverse values @?= [NegInfinity, Finite day, PosInfinity]+        , testCase "date arrays preserve infinities" $ withConnection ":memory:" $ \conn -> do+            let values = listArray (0 :: Int, 2) [NegInfinity, Finite day, PosInfinity]+            query conn "SELECT ?" (Only values) >>= (@?= [Only values])+        , testCase "generic records retain finite values infinities and NULL" $ withConnection ":memory:" $ \conn -> do+            let values = TemporalRecord NegInfinity (Finite local) PosInfinity Nothing [PosInfinity, Finite day, NegInfinity]+            query conn "SELECT ?" (Only values) >>= (@?= [Only values])+        , testCase "generic union payloads retain their types" $ withConnection ":memory:" $ \conn ->+            forM_ [ADate NegInfinity, ADate (Finite day), ALocal PosInfinity, AUtc NegInfinity, ADateNull Nothing] $ \value ->+                query conn "SELECT ?" (Only value) >>= (@?= [Only value])+        ]+  where+    day = fromGregorian 2000 1 2+    local = LocalTime day (TimeOfDay 3 4 5)+    instant = localTimeToUTC utc local+    dateText NegInfinity = "-infinity"+    dateText PosInfinity = "infinity"+    dateText (Finite _) = "2000-01-02"++-- | Require a field conversion error, rather than an unrelated exception.+assertConversionError :: IO a -> Assertion+assertConversionError action = do+    result <- try action+    case result of+        Left SQLError{sqlErrorMessage = message} ->+            assertBool "error must identify infinity" ("infinity" `Text.isInfixOf` message)+        Right _ -> assertFailure "expected a conversion error"

@@ -0,0 +1,295 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingVia #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Regression tests for exact values and controlled conversion failures.+module ValueRegressionTests (main, valueRegressionTests) where++import Control.Exception (IOException, bracket, try)+import Control.Monad (forM_)+import Data.Array (Array, listArray)+import qualified Data.ByteString as BS+import Data.Int (Int32, Int64, Int8)+import qualified Data.Map.Strict as Map+import Data.Ratio ((%))+import Data.String (fromString)+import Data.Text (Text)+import Data.Time.Calendar (Day, addDays, fromGregorian)+import Data.Time.Clock (UTCTime (..))+import Data.Time.Clock.POSIX (posixSecondsToUTCTime)+import Data.Time.LocalTime (LocalTime (..), TimeOfDay (..), minutesToTimeZone, utc, utcToLocalTime)+import Data.Word (Word16, Word32, Word8)+import Database.DuckDB.FFI+import Database.DuckDB.Simple+import Database.DuckDB.Simple.FromField (BitString (..), DecimalValue (..), FieldValue (..), TimeWithZone (..), bsFromBool)+import Database.DuckDB.Simple.Generic (ViaDuckDB (..), genericFromFieldValue, genericToStructValue)+import Database.DuckDB.Simple.Internal (withConnectionHandle)+import Database.DuckDB.Simple.LogicalRep+import Database.DuckDB.Simple.Time (Unbounded (..))+import Database.DuckDB.Simple.ToField (ToDuckValue (..))+import Foreign.C.String (withCString)+import Foreign.Marshal.Alloc (alloca)+import Foreign.Ptr (nullPtr)+import Foreign.Storable (peek, poke)+import GHC.Generics (Generic)+import Test.Tasty (TestTree, defaultMain, testGroup)+import Test.Tasty.HUnit++-- | Native timestamps used to test the full storage range.+data NativeTimestamp = Seconds Int64 | Milliseconds Int64 | Microseconds Int64 | Nanoseconds Int64+    deriving (Show)++instance DuckDBColumnType NativeTimestamp where+    duckdbColumnTypeFor _ = "TIMESTAMP"++instance ToDuckValue NativeTimestamp where+    toDuckValue (Seconds value) = c_duckdb_create_timestamp_s (DuckDBTimestampS value)+    toDuckValue (Milliseconds value) = c_duckdb_create_timestamp_ms (DuckDBTimestampMs value)+    toDuckValue (Microseconds value) = c_duckdb_create_timestamp (DuckDBTimestamp value)+    toDuckValue (Nanoseconds value) = c_duckdb_create_timestamp_ns (DuckDBTimestampNs value)++instance ToField NativeTimestamp++-- | Record with identical field types to detect positional decoding.+data NamedRecord = NamedRecord {firstValue :: Int64, secondValue :: Int64}+    deriving (Eq, Show, Generic)++-- | Sum with a payload to test NULL member handling.+data NullableSum = EmptyMember | DataMember Int64+    deriving (Eq, Show, Generic)++-- | Generic nullary constructors retain their existing UNION schema.+data Colour = Red | Blue+    deriving stock (Eq, Show, Generic)+    deriving (DuckDBColumnType, ToField, FromField) via (ViaDuckDB Colour)++-- | Run this module without the main integration suite.+main :: IO ()+main = defaultMain valueRegressionTests++-- | Focused value regressions that also run against the baseline library.+valueRegressionTests :: TestTree+valueRegressionTests =+    testGroup "value regressions" $+        [ testCase "REAL decodes to Float and Double" $ withConnection ":memory:" \conn -> do+            (query_ conn "SELECT 1.25::REAL" :: IO [Only Float]) >>= (@?= [Only 1.25])+            (query_ conn "SELECT 1.25::REAL" :: IO [Only Double]) >>= (@?= [Only 1.25])+        , testCase "Float parameters still decode as Text" $ withConnection ":memory:" \conn ->+            (query conn "SELECT ?" (Only (1.25 :: Float)) :: IO [Only Text]) >>= (@?= [Only "1.25"])+        , testCase "generic nullary UNION constructors round trip through DuckDB" $ withConnection ":memory:" \conn ->+            forM_ [Red, Blue] \colour ->+                (query conn "SELECT ?" (Only colour) :: IO [Only Colour]) >>= (@?= [Only colour])+        , testCase "UNION NULL payload retains its member type" $ withConnection ":memory:" \conn -> do+            let members = listArray (0, 1) [UnionMemberType "number" (LogicalTypeScalar DuckDBTypeBigInt), UnionMemberType "text" (LogicalTypeScalar DuckDBTypeVarchar)]+                original = UnionValue 0 "number" FieldNull members+            (query conn "SELECT ?" (Only original) :: IO [Only (UnionValue FieldValue)]) >>= (@?= [Only original])+        , testCase "Float parameter retains FLOAT type" $ withConnection ":memory:" \conn -> do+            (query conn "SELECT typeof(?), ?" (1.25 :: Float, 1.25 :: Float) :: IO [(Text, Float)]) >>= (@?= [("FLOAT", 1.25)])+        , testCase "Float special values survive decoding" $ withConnection ":memory:" \conn -> do+            [Only value] <- query_ conn "SELECT 'NaN'::REAL" :: IO [Only Float]+            assertBool "expected NaN" (isNaN value)+            [Only value'] <- query_ conn "SELECT 'Infinity'::REAL" :: IO [Only Double]+            assertBool "expected positive infinity" (isInfinite value' && value' > 0)+        , testCase "finite Double overflow to Float fails" $ withConnection ":memory:" \conn ->+            assertConversionError (query_ conn "SELECT 1e300::DOUBLE" :: IO [Only Float])+        , testCase "embedded NUL and Unicode Text round trip" $ withConnection ":memory:" \conn -> do+            let value = "before\0after íslenska λ 😀" :: Text+            (query conn "SELECT ?" (Only value) :: IO [Only Text]) >>= (@?= [Only value])+        , testCase "Int8 rejects both overflow directions" $ withConnection ":memory:" \conn -> do+            forM_ ["SELECT 128::BIGINT", "SELECT -129::SMALLINT"] \sql ->+                assertConversionError (query_ conn sql :: IO [Only Int8])+            (query_ conn "SELECT -128::BIGINT UNION ALL SELECT 127" :: IO [Only Int8]) >>= (@?= [Only (-128), Only 127])+        , testCase "unsigned narrowing rejects overflow" $ withConnection ":memory:" \conn -> do+            assertConversionError (query_ conn "SELECT 256::USMALLINT" :: IO [Only Word8])+            assertConversionError (query_ conn "SELECT 65536::UINTEGER" :: IO [Only Word16])+            assertConversionError (query_ conn "SELECT 4294967296::UBIGINT" :: IO [Only Word32])+        , testCase "finite timestamp extrema preserve all units" $ withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+            forM_ [(Seconds, 1, "TIMESTAMP_S"), (Milliseconds, 1000, "TIMESTAMP_MS"), (Microseconds, 1000000, "TIMESTAMP"), (Nanoseconds, 1000000000, "TIMESTAMP_NS")] \(constructor, units, dtype) ->+                forM_ [minBound, negate (maxBound :: Int64) + 1, -1, 0, maxBound - 1] \value -> do+                    let expected = utcToLocalTime utc (posixSecondsToUTCTime (fromRational (toInteger value % units)))+                    appendTimestamp conn dtype (toDuckValue (constructor value))+                    (query_ conn "SELECT value FROM native_timestamp" :: IO [Only LocalTime]) >>= (@?= [Only expected])+                    [Only original] <- query_ conn "SELECT {'value': value} FROM native_timestamp" :: IO [Only (StructValue FieldValue)]+                    -- DuckDB needs an explicit parameter type for extreme S/MS values.+                    query conn (fromString ("SELECT ?::STRUCT(value " <> dtype <> ")")) (Only original) >>= (@?= [Only original])+                    _ <- execute_ conn "DROP TABLE native_timestamp"+                    pure ()+        , testCase "composite timestamps reject infinity and storage overflow" $ withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+            forM_ [(DuckDBTypeTimestampS, 1), (DuckDBTypeTimestampMs, 1000), (DuckDBTypeTimestamp, 1000000), (DuckDBTypeTimestampNs, 1000000000)] \(dtype, units) ->+                forM_ [toInteger (minBound :: Int64) - 1, negate (toInteger (maxBound :: Int64)), toInteger (maxBound :: Int64), toInteger (maxBound :: Int64) + 1] \value -> do+                    let timestamp = utcToLocalTime utc (posixSecondsToUTCTime (fromRational (value % units)))+                        struct = singleField (LogicalTypeScalar dtype) (FieldTimestamp (Finite timestamp))+                    assertIOError (query conn "SELECT ?" (Only struct) :: IO [Only FieldValue])+        , testCase "composite timestamps floor fractional units before the epoch" $ withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+            forM_ [(DuckDBTypeTimestampS, 1), (DuckDBTypeTimestampMs, 1000), (DuckDBTypeTimestamp, 1000000), (DuckDBTypeTimestampNs, 1000000000)] \(dtype, units) -> do+                let timestamp = utcToLocalTime utc (posixSecondsToUTCTime (fromRational ((-1) % (2 * units))))+                    expected = utcToLocalTime utc (posixSecondsToUTCTime (fromRational ((-1) % units)))+                    struct = singleField (LogicalTypeScalar dtype) (FieldTimestamp (Finite timestamp))+                (query conn "SELECT (?).value" (Only struct) :: IO [Only LocalTime]) >>= (@?= [Only expected])+        , testCase "composite TIME_NS rejects invalid clock components" $ withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+            forM_ [TimeOfDay (-1) 0 0, TimeOfDay 25 0 0, TimeOfDay 0 60 0, TimeOfDay 0 0 (-1), TimeOfDay 24 0 0.000000001, TimeOfDay 23 59 60.000000001] \value ->+                assertIOError (query conn "SELECT ?" (Only (singleField (LogicalTypeScalar DuckDBTypeTimeNs) (FieldTime value))) :: IO [Only FieldValue])+        , testCase "microsecond timestamps retain finite extrema" $ withConnection ":memory:" \conn ->+            forM_ [minBound, negate (maxBound :: Int64) + 1, -1, 0, maxBound - 1] \value -> do+                let expected = utcToLocalTime utc (posixSecondsToUTCTime (fromRational (toInteger value % 1000000)))+                _ <- execute_ conn "CREATE TABLE microsecond_timestamp(value TIMESTAMP)"+                _ <- execute conn "INSERT INTO microsecond_timestamp VALUES (?)" (Only expected)+                (query_ conn "SELECT value FROM microsecond_timestamp" :: IO [Only LocalTime]) >>= (@?= [Only expected])+                _ <- execute_ conn "DROP TABLE microsecond_timestamp"+                pure ()+        , testCase "out-of-range dates and timestamps fail without abort" $ withConnection ":memory:" \conn -> do+            let hugeDay = fromGregorian 10000000 1 1+                local = LocalTime hugeDay (TimeOfDay 0 0 0)+            assertIOError (query conn "SELECT ?" (Only hugeDay) :: IO [Only Day])+            assertIOError (query conn "SELECT ?" (Only local) :: IO [Only LocalTime])+            assertIOError (query conn "SELECT ?" (Only (UTCTime hugeDay 0)) :: IO [Only UTCTime])+        , testCase "date finite storage extrema round trip" $ withConnection ":memory:" \conn ->+            forM_ [minBound, negate (maxBound :: Int32) + 1, -1, 0, maxBound - 1] \value -> do+                let day = addDays (toInteger value) (fromGregorian 1970 1 1)+                (query conn "SELECT ?" (Only day) :: IO [Only Day]) >>= (@?= [Only day])+        , testCase "input infinity sentinels are rejected" $ withConnection ":memory:" \conn -> do+            forM_ [toInteger (maxBound :: Int32), negate (toInteger (maxBound :: Int32))] \days ->+                assertIOError (query conn "SELECT ?" (Only (addDays days (fromGregorian 1970 1 1))) :: IO [Only Day])+            forM_ [toInteger (maxBound :: Int64), negate (toInteger (maxBound :: Int64))] \micros -> do+                let value = utcToLocalTime utc (posixSecondsToUTCTime (fromRational (micros % 1000000)))+                assertIOError (query conn "SELECT ?" (Only value) :: IO [Only LocalTime])+        , testCase "UTC binding is TIMESTAMPTZ under a non-UTC zone" $ withConnection ":memory:" \conn -> do+            _ <- execute_ conn "SET TimeZone = 'Pacific/Auckland'"+            let value = UTCTime (fromGregorian 2000 1 2) 12345+            (query conn "SELECT typeof(?), ?" (value, value) :: IO [(Text, UTCTime)]) >>= (@?= [("TIMESTAMP WITH TIME ZONE", value)])+            _ <- execute_ conn "CREATE TABLE utc_binding(value TIMESTAMPTZ)"+            _ <- execute conn "INSERT INTO utc_binding VALUES (?)" (Only value)+            (query_ conn "SELECT value FROM utc_binding" :: IO [Only UTCTime]) >>= (@?= [Only value])+        , testCase "DECIMAL(38) retains exact integer precision" $ withConnection ":memory:" \conn -> do+            let decimal = DecimalValue 38 9 12345678901234567890123456789012345678+                struct = singleField (LogicalTypeDecimal 38 9) (FieldDecimal decimal)+            [Only result] <- query conn "SELECT ?" (Only struct) :: IO [Only (StructValue FieldValue)]+            structValueFields result @?= structValueFields struct+        , testCase "invalid DECIMAL width scale and magnitude are rejected" $ withConnection ":memory:" \conn ->+            forM_ [DecimalValue 0 0 0, DecimalValue 39 0 0, DecimalValue 2 3 1, DecimalValue 2 0 100, DecimalValue 2 0 (-100)] \decimal ->+                assertIOError (query conn "SELECT ?" (Only (singleField (LogicalTypeDecimal (decimalWidth decimal) (decimalScale decimal)) (FieldDecimal decimal))) :: IO [Only FieldValue])+        , testCase "nested conversion failure leaves the connection usable" $ withConnection ":memory:" \conn -> do+            let days = listArray (0, 1) [fromGregorian 2000 1 1, fromGregorian 10000000 1 1] :: Array Int Day+            assertIOError (query conn "SELECT ?" (Only days) :: IO [Only FieldValue])+            (query_ conn "SELECT 42" :: IO [Only Int64]) >>= (@?= [Only 42])+        , testCase "generic NULL product and payload return Left" $ do+            assertLeft (genericFromFieldValue FieldNull :: Either String NamedRecord)+            let member = UnionMemberType "DataMember" (LogicalTypeScalar DuckDBTypeBigInt)+                union = UnionValue 0 "DataMember" FieldNull (listArray (0, 0) [member])+            assertLeft (genericFromFieldValue (FieldUnion union) :: Either String NullableSum)+        , testCase "generic records decode by name" $ withConnection ":memory:" \conn -> do+            [Only struct] <- query_ conn "SELECT {'secondValue': 2::BIGINT, 'firstValue': 1::BIGINT}" :: IO [Only (StructValue FieldValue)]+            genericFromFieldValue (FieldStruct struct) @?= Right (NamedRecord 1 2)+        , testCase "generic sums decode by member name" $ withConnection ":memory:" \conn -> do+            [Only union] <- query_ conn "SELECT union_value(DataMember := {'field1': 42::BIGINT})" :: IO [Only (UnionValue FieldValue)]+            genericFromFieldValue (FieldUnion union) @?= Right (DataMember 42)+        , testCase "union binding rejects inconsistent member name" $ withConnection ":memory:" \conn -> do+            let member = UnionMemberType "right" (LogicalTypeScalar DuckDBTypeBigInt)+                union = UnionValue 0 "wrong" (FieldInt64 42) (listArray (0, 0) [member])+            assertIOError (query conn "SELECT ?" (Only union) :: IO [Only FieldValue])+        , testCase "generic records reject wrong names" $ do+            case genericToStructValue (NamedRecord 1 2) of+                Nothing -> assertFailure "missing generic struct"+                Just struct -> do+                    let fields = listArray (0, 1) [StructField "wrong" (FieldInt64 1), StructField "secondValue" (FieldInt64 2)]+                    assertLeft (genericFromFieldValue (FieldStruct struct{structValueFields = fields}) :: Either String NamedRecord)+        , testCase "logical type Unicode names round trip" $ do+            let logical = LogicalTypeStruct (listArray (0, 0) [StructField "íslenska_λ_😀" (LogicalTypeScalar DuckDBTypeBigInt)])+            bracket (logicalTypeFromRep logical) destroyLogicalType \handle ->+                logicalTypeToRep handle >>= (@?= logical)+        , testCase "logical type names reject embedded NUL" $ do+            let logical = LogicalTypeStruct (listArray (0, 0) [StructField "before\0after" (LogicalTypeScalar DuckDBTypeBigInt)])+            assertIOError (bracket (logicalTypeFromRep logical) destroyLogicalType (const (pure ())))+        , testCase "invalid native composite constructor returns controlled error" $ withConnection ":memory:" \conn -> do+            let struct = singleField (LogicalTypeMap (LogicalTypeScalar DuckDBTypeBigInt) (LogicalTypeScalar DuckDBTypeBigInt)) (FieldMap [(FieldNull, FieldInt64 1)])+            assertIOError (query conn "SELECT ?" (Only struct) :: IO [Only FieldValue])+        , testCase "BIT padding preserves SQL bit_count" $ withConnection ":memory:" \conn -> do+            let bits = bsFromBool [True, False, True]+            (query conn "SELECT bit_count(?), ?" (bits, bits) :: IO [(Int64, BitString)]) >>= (@?= [(2, bits)])+        , testCase "unsupported BIT input fails before native use" $ withConnection ":memory:" \conn ->+            forM_ [BitString 0 BS.empty, BitString 8 (BS.singleton 1)] \bits ->+                assertIOError (query conn "SELECT ?" (Only bits) :: IO [Only BitString])+        , testCase "TIMETZ second offsets fail without rounding" $ withConnection ":memory:" \conn ->+            assertIOError (query_ conn "SELECT '12:00:00+01:23:45'::TIMETZ" :: IO [Only TimeWithZone])+        , testCase "ENUM payload index must be in its dictionary" $ withConnection ":memory:" \conn ->+            assertIOError (query conn "SELECT ?" (Only (singleField (LogicalTypeEnum (listArray (0, 1) ["a", "b"])) (FieldEnum 2))) :: IO [Only FieldValue])+        , testCase "TIMETZ input offset overflow fails before native use" $ withConnection ":memory:" \conn -> do+            let value = TimeWithZone (TimeOfDay 12 0 0) (minutesToTimeZone maxBound)+                struct = singleField (LogicalTypeScalar DuckDBTypeTimeTz) (FieldTimeTZ value)+            assertIOError (query conn "SELECT ?" (Only struct) :: IO [Only FieldValue])+        , testCase "invalid TIME input returns controlled error" $ withConnection ":memory:" \conn ->+            assertIOError (query conn "SELECT ?" (Only (TimeOfDay 1000000000 0 0)) :: IO [Only TimeOfDay])+        ]+            <> [ testCase ("composite temporal round trip: " <> dtype) $ withConnectionWithConfig ":memory:" [("threads", "1")] \conn ->+                    forM_ ("NULL" : map (\value -> "'" <> value <> "'") values) \value ->+                        assertTemporalRoundTrip conn dtype (value <> "::" <> dtype)+               | (dtype, values) <-+                    [(dtype, ["-infinity", "infinity", "1969-12-31 23:59:59.123456789", "2000-01-01 12:34:56.123456789"]) | dtype <- ["TIMESTAMP", "TIMESTAMP_S", "TIMESTAMP_MS", "TIMESTAMP_NS", "TIMESTAMPTZ"]]+                        <> [("DATE", ["-infinity", "infinity", "1969-12-31", "2000-01-01"])]+                        <> [("TIME_NS", ["00:00:00", "12:34:56.123456789", "23:59:59.999999999", "24:00:00"])]+               ]+            <> [ testCase (dtype <> " " <> value <> " rejects a finite result type") $ withConnection ":memory:" \conn ->+                    assertConversionError (query_ conn (fromString ("SELECT '" <> value <> "'::" <> dtype)) :: IO [Only LocalTime])+               | dtype <- ["DATE", "TIMESTAMP", "TIMESTAMP_S", "TIMESTAMP_MS", "TIMESTAMP_NS", "TIMESTAMPTZ"]+               , value <- ["infinity", "-infinity"]+               ]++-- | Rebind native temporal values in structs, unions, and nested collections.+assertTemporalRoundTrip :: Connection -> String -> String -> Assertion+assertTemporalRoundTrip conn dtype expression = do+    let sql =+            "WITH temporal AS (SELECT "+                <> expression+                <> " AS value) "+                <> "SELECT {'scalar': value, 'list': [value, NULL], 'array': [value, NULL]::"+                <> dtype+                <> "[2], 'map': map([1], [value])}, union_value(value := value) FROM temporal"+    [original] <- query_ conn (fromString sql) :: IO [(StructValue FieldValue, UnionValue FieldValue)]+    query conn "SELECT ?, ?" original >>= (@?= [original])++-- | Store raw timestamp units without the prepared-parameter type conversion.+appendTimestamp :: Connection -> String -> IO DuckDBValue -> IO ()+appendTimestamp conn dtype createValue = do+    _ <- execute_ conn (fromString ("CREATE TABLE native_timestamp(value " <> dtype <> ")"))+    withConnectionHandle conn \handle ->+        alloca \appenderPtr -> do+            poke appenderPtr nullPtr+            withCString "native_timestamp" \table ->+                c_duckdb_appender_create handle nullPtr table appenderPtr >>= (@?= DuckDBSuccess)+            bracket (peek appenderPtr) (const (c_duckdb_appender_destroy appenderPtr >> pure ())) \appender -> do+                bracket createValue (\value -> alloca \ptr -> poke ptr value >> c_duckdb_destroy_value ptr) \value ->+                    c_duckdb_append_value appender value >>= (@?= DuckDBSuccess)+                c_duckdb_appender_end_row appender >>= (@?= DuckDBSuccess)+                c_duckdb_appender_flush appender >>= (@?= DuckDBSuccess)++-- | Build a single-field struct for nested binding tests.+singleField :: LogicalTypeRep -> FieldValue -> StructValue FieldValue+singleField logical value =+    StructValue+        (listArray (0, 0) [StructField "value" value])+        (listArray (0, 0) [StructField "value" logical])+        (Map.singleton "value" 0)++-- | Assert a controlled input or materialization error.+assertIOError :: IO a -> Assertion+assertIOError action = do+    result <- try action+    case result of+        Left (_ :: IOException) -> pure ()+        Right _ -> assertFailure "expected IOException"++-- | Assert a controlled FromField conversion error.+assertConversionError :: IO a -> Assertion+assertConversionError action = do+    result <- try action+    case result of+        Left (_ :: SQLError) -> pure ()+        Right _ -> assertFailure "expected ResultError"++-- | Assert rejection without forcing a partial generic value.+assertLeft :: Either String a -> Assertion+assertLeft (Left _) = pure ()+assertLeft (Right _) = assertFailure "expected Left"