kiroku-store-migrations 0.3.0.0 → 0.3.1.0
raw patch · 6 files changed
+514/−20 lines, 6 filesdep +timePVP ok
version bump matches the API change (PVP)
Dependencies added: time
API changes (from Hackage documentation)
Files
- CHANGELOG.md +8/−1
- README.md +9/−5
- kiroku-store-migrations.cabal +2/−1
- migrations/0009.sql +35/−0
- migrations/manifest +1/−0
- test/Main.hs +459/−13
CHANGELOG.md view
@@ -1,6 +1,13 @@ # Changelog -## Unreleased+## 0.3.1.0 — 2026-08-13++### New Features++* Added forward migration `0009`, which publishes the frozen, structurally read-only+ `kiroku.subscription_checkpoints_v1` relation for exact durable subscription-member+ checkpoints. Database readers can receive schema usage and view selection without access to+ Kiroku's private checkpoint table; no role or grant is created automatically. ## 0.3.0.0 — 2026-07-14
README.md view
@@ -2,8 +2,9 @@ `kiroku-store-migrations` owns Kiroku's PostgreSQL schema as one native `pg-migrate` component named `kiroku`. The component embeds an ordered manifest-and seven immutable SQL payloads, so applications can compose it with other-libraries without copying Kiroku SQL.+and nine SQL payloads, so applications can compose it with other libraries+without copying Kiroku SQL. The first seven payloads are immutable historical+Codd bytes; `0008` and `0009` are native-only forward migrations. ## Public API @@ -101,9 +102,12 @@ It proves manifest order, legacy SHA-256 parity, fresh apply, strict verify, idempotent rerun, concurrent locking, current Codd V5 import, legacy `codd_schema` import, partial-row rejection, import audit records, and-source-ledger preservation. `cabal test kiroku-store:kiroku-store-test` consumes-the same native plan through `kiroku-test-support` and proves the complete store-behavior, including append and read scenarios.+source-ledger preservation. The relation contract suite also proves the frozen+`kiroku.subscription_checkpoints_v1` catalog, non-null value semantics,+owner-rights privilege isolation, structural read-only behavior, downstream+view survival, and indexed query plan. `cabal test kiroku-store:kiroku-store-test`+consumes the same native plan through `kiroku-test-support` and proves the+complete store behavior, including append and read scenarios. ## Recovery
kiroku-store-migrations.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: kiroku-store-migrations-version: 0.3.0.0+version: 0.3.1.0 synopsis: Schema migrations for kiroku-store description: Native pg-migrate component, Codd history mapping, and migration executable@@ -92,3 +92,4 @@ , pg-migrate-test-support ^>=1.1.0.0 , temporary >=1.3 && <1.4 , text >=2.0 && <2.2+ , time >=1.12 && <1.15
+ migrations/0009.sql view
@@ -0,0 +1,35 @@+-- Publish the frozen, owner-rights, structurally read-only v1 checkpoint relation.+CREATE VIEW kiroku.subscription_checkpoints_v1+ (subscription_name,+ consumer_group_member,+ checkpoint_position,+ checkpoint_updated_at)+WITH (security_invoker = false)+AS+WITH checkpoint_rows AS NOT MATERIALIZED (+ SELECT subscription_name,+ consumer_group_member,+ last_seen AS checkpoint_position,+ updated_at AS checkpoint_updated_at+ FROM kiroku.subscriptions+)+SELECT subscription_name,+ consumer_group_member,+ checkpoint_position,+ checkpoint_updated_at+FROM checkpoint_rows;++COMMENT ON VIEW kiroku.subscription_checkpoints_v1 IS+ 'Stable read-only v1 relation of exact persisted subscription-member checkpoints; its columns and order are frozen, and its rows are unordered unless the caller supplies ORDER BY.';++COMMENT ON COLUMN kiroku.subscription_checkpoints_v1.subscription_name IS+ 'Persisted subscription name; member zero does not distinguish a non-group subscription from member zero of a consumer group.';++COMMENT ON COLUMN kiroku.subscription_checkpoints_v1.consumer_group_member IS+ 'Persisted consumer-group member key; member zero carries no topology classification.';++COMMENT ON COLUMN kiroku.subscription_checkpoints_v1.checkpoint_position IS+ 'Exact persisted global position for this subscription member; an explicit reset may move it backward or forward.';++COMMENT ON COLUMN kiroku.subscription_checkpoints_v1.checkpoint_updated_at IS+ 'Time of the latest checkpoint-row upsert; it does not imply position advancement or worker liveness.';
migrations/manifest view
@@ -6,3 +6,4 @@ 0006-stream-name-length-check.sql 0007-stream-truncate-before.sql 0008-schema-management-comment.sql+0009.sql
test/Main.hs view
@@ -9,12 +9,14 @@ import Data.ByteString qualified as ByteString import Data.Either (isLeft) import Data.Foldable (toList)-import Data.Int (Int64)+import Data.Int (Int32, Int64) import Data.List (sort) import Data.List.NonEmpty (NonEmpty (..)) import Data.Text (Text) import Data.Text qualified as Text import Data.Text.IO qualified as Text.IO+import Data.Time (UTCTime (..), fromGregorian, secondsToDiffTime)+import Data.Unique (hashUnique, newUnique) import Database.PostgreSQL.Migrate import Database.PostgreSQL.Migrate.History.Codd import Database.PostgreSQL.Migrate.Internal (migrationChecksumBytes)@@ -24,6 +26,7 @@ import Hasql.Connection.Settings qualified as Settings import Hasql.Decoders qualified as Decoders import Hasql.Encoders qualified as Encoders+import Hasql.Errors qualified as Errors import Hasql.Session qualified as Session import Hasql.Statement (Statement) import Hasql.Statement qualified as Statement@@ -39,7 +42,7 @@ main :: IO () main = hspec $ do describe "native Kiroku migration definition" $ do- it "tracks the eight native files in manifest order" $ do+ it "tracks the nine native files in manifest order" $ do directory <- findMigrationsDirectory manifest <- Text.lines <$> Text.IO.readFile (directory </> "manifest") manifest `shouldBe` Text.pack <$> nativeMigrationFiles@@ -52,7 +55,7 @@ bytes <- ByteString.readFile (directory </> nativeName) lookup legacyName lockEntries `shouldBe` Just (checksumText bytes) - it "builds component kiroku and an eight-migration plan" $ do+ it "builds component kiroku and a nine-migration plan" $ do component <- requireRight kirokuMigrations component `seq` pure () plan <- requirePlan@@ -62,7 +65,7 @@ ] validateHistoryMappingTargets plan kirokuCoddHistoryMappings `shouldBe` Right ()- length targetIds `shouldBe` 8+ length targetIds `shouldBe` length nativeMigrationFiles describe "native migration authoring" $ do it "creates the next numeric file and atomically appends the manifest" $@@ -88,18 +91,18 @@ `shouldReturn` "0007-existing.sql\n" describe "fresh native databases" $ do- it "applies all eight, verifies strictly, and reports AlreadyApplied on rerun" $ do+ it "applies all nine, verifies strictly, and reports AlreadyApplied on rerun" $ do plan <- requirePlan result <- withMigratedDatabase plan $ \connection -> do assertSchema connection let provider = providerFor connection rerun <- runMigrationPlanWith defaultRunOptions provider plan >>= requireMigration- reportOutcomes rerun `shouldBe` replicate 8 AlreadyApplied+ reportOutcomes rerun `shouldBe` replicate (length nativeMigrationFiles) AlreadyApplied verified <- verifyMigrationPlanWith defaultRunOptions provider plan >>= requireMigration case verified of VerificationReport verificationIssues applied _ _ -> do verificationIssues `shouldBe` []- length applied `shouldBe` 8+ length applied `shouldBe` length nativeMigrationFiles either (expectationFailure . show) pure result it "serializes concurrent applies through the pg-migrate advisory lock" $ do@@ -111,8 +114,140 @@ (runMigrationPlan defaultRunOptions settings plan >>= requireMigration) (runMigrationPlan defaultRunOptions settings plan >>= requireMigration) sort [reportOutcomes first, reportOutcomes second]- `shouldBe` sort [replicate 8 AppliedNow, replicate 8 AlreadyApplied]+ `shouldBe` sort+ [ replicate (length nativeMigrationFiles) AppliedNow+ , replicate (length nativeMigrationFiles) AlreadyApplied+ ] + describe "subscription checkpoint SQL relation" $ do+ it "publishes the frozen ordinary-view catalog contract" $ do+ plan <- requirePlan+ result <- withMigratedDatabase plan $ \connection -> do+ relation <- useSession connection (Session.statement () relationCatalogStatement)+ relation+ `shouldBe` ( "v"+ , "security_invoker=false"+ , True+ , "NO"+ , "NO"+ , "Stable read-only v1 relation of exact persisted subscription-member checkpoints; its columns and order are frozen, and its rows are unordered unless the caller supplies ORDER BY."+ )+ columns <- useSession connection (Session.statement () relationColumnsStatement)+ columns+ `shouldBe` [+ ( "subscription_name"+ , "text"+ , False+ , "Persisted subscription name; member zero does not distinguish a non-group subscription from member zero of a consumer group."+ )+ ,+ ( "consumer_group_member"+ , "integer"+ , False+ , "Persisted consumer-group member key; member zero carries no topology classification."+ )+ ,+ ( "checkpoint_position"+ , "bigint"+ , False+ , "Exact persisted global position for this subscription member; an explicit reset may move it backward or forward."+ )+ ,+ ( "checkpoint_updated_at"+ , "timestamp with time zone"+ , False+ , "Time of the latest checkpoint-row upsert; it does not imply position advancement or worker liveness."+ )+ ]+ either (expectationFailure . show) pure result++ it "returns zero rows for an empty checkpoint inventory" $ do+ plan <- requirePlan+ result <- withMigratedDatabase plan $ \connection -> do+ rows <- useSession connection (Session.statement () checkpointRowsStatement)+ rows `shouldBe` []+ either (expectationFailure . show) pure result++ it "returns exact non-null member rows and exposes reassignment only after commit" $ do+ plan <- requirePlan+ withKirokuPg $ \database -> do+ let settings = Pg.connectionSettings database+ _ <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+ withConnection settings $ \writer ->+ withConnection settings $ \reader -> do+ useSession writer (Session.script checkpointFixtureSql)+ rows <- useSession reader (Session.statement () checkpointRowsStatement)+ rows+ `shouldBe` [ ("grouped", 0, 20, checkpointFixtureTimestamp)+ , ("grouped", 1, 21, checkpointFixtureTimestamp)+ , ("grouped", 2, 22, checkpointFixtureTimestamp)+ , ("ordinary", 0, 11, checkpointFixtureTimestamp)+ ]++ useSession writer (Session.script "BEGIN")+ useSession writer (Session.script checkpointAdvanceSql)+ checkpointPosition writer `shouldReturn` 5+ checkpointPosition reader `shouldReturn` 11+ useSession writer (Session.script "COMMIT")+ checkpointPosition reader `shouldReturn` 5++ useSession writer (Session.script "BEGIN")+ useSession writer (Session.script checkpointRegressionSql)+ checkpointPosition writer `shouldReturn` 2+ checkpointPosition reader `shouldReturn` 5+ useSession writer (Session.script "ROLLBACK")+ checkpointPosition reader `shouldReturn` 5++ it "allows view-only readers while structurally rejecting owner updates" $ do+ plan <- requirePlan+ result <- withMigratedDatabase plan $ \connection ->+ withTemporaryReaderRole connection $ \role -> do+ let identifier = quoteTestIdentifier role+ useSession+ connection+ ( Session.script+ ( "GRANT USAGE ON SCHEMA kiroku TO "+ <> identifier+ <> "; GRANT SELECT ON kiroku.subscription_checkpoints_v1 TO "+ <> identifier+ <> "; SET ROLE "+ <> identifier+ )+ )+ viewCount <- useSession connection (Session.statement () checkpointRowCountStatement)+ viewCount `shouldBe` 0+ privatePrivilege <- useSession connection (Session.statement () privateTablePrivilegeStatement)+ privatePrivilege `shouldBe` False+ directPrivateRead <- Connection.use connection (Session.statement () privateTableCountStatement)+ directPrivateRead `shouldSatisfy` hasSqlState "42501"++ useSession connection (Session.script "RESET ROLE")+ ownerUpdate <- Connection.use connection (Session.statement () updatePublicViewStatement)+ ownerUpdate `shouldSatisfy` hasSqlState "55000"+ either (expectationFailure . show) pure result++ it "preserves a downstream view while private checkpoint storage is replaced" $ do+ plan <- requirePlan+ withKirokuPg $ \database -> do+ let settings = Pg.connectionSettings database+ _ <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+ withConnection settings $ \connection -> do+ useSession connection (Session.script downstreamReplacementFixtureSql)+ floors <- useSession connection (Session.statement () downstreamFloorsStatement)+ floors `shouldBe` [("dependency", 42)]++ it "pushes a subscription filter to the existing checkpoint index" $ do+ plan <- requirePlan+ result <- withMigratedDatabase plan $ \connection -> do+ useSession connection (Session.script checkpointPlanFixtureSql)+ version <- useSession connection (Session.statement () serverVersionStatement)+ planLines <- useSession connection (Session.statement () checkpointExplainStatement)+ let planText = Text.unlines planLines+ Text.IO.putStrLn ("PostgreSQL " <> version <> " checkpoint relation plan:\n" <> planText)+ planText `shouldSatisfy` Text.isInfixOf "ix_subscriptions_name_member"+ Text.isInfixOf "CTE Scan" planText `shouldBe` False+ either (expectationFailure . show) pure result+ describe "Codd history import" $ do it "imports a current codd V5 ledger, verifies, and never replays SQL" $ importFixture "codd"@@ -155,20 +290,23 @@ importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings >>= requireRight importOutcomes first `shouldBe` replicate 7 Imported- canaryId <- requireRight (migrationId "kiroku" "0008-schema-management-comment")+ pendingIds <-+ traverse+ (requireRight . migrationId "kiroku")+ ["0008-schema-management-comment", "0009"] verifiedBeforeCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration case verifiedBeforeCanary of VerificationReport verificationIssues _ _ _ -> verificationIssues- `shouldBe` [PendingMigration canaryId]+ `shouldBe` (PendingMigration <$> pendingIds) up <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration- reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> [AppliedNow]+ reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> replicate 2 AppliedNow verifiedAfterCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration case verifiedAfterCanary of VerificationReport verificationIssues _ _ _ -> verificationIssues `shouldBe` [] rerun <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration- reportOutcomes rerun `shouldBe` replicate 8 AlreadyApplied+ reportOutcomes rerun `shouldBe` replicate (length nativeMigrationFiles) AlreadyApplied second <- importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings >>= requireRight@@ -178,7 +316,7 @@ sourceRows <- useSession connection (Session.statement () (sourceRowCountStatement sourceSchema)) sourceRows `shouldBe` 7 facts <- useSession connection (Session.statement () importFactsStatement)- facts `shouldBe` (8, 7, True)+ facts `shouldBe` (fromIntegral (length nativeMigrationFiles), 7, True) nativeMigrationFiles :: [FilePath] nativeMigrationFiles =@@ -190,6 +328,7 @@ , "0006-stream-name-length-check.sql" , "0007-stream-truncate-before.sql" , "0008-schema-management-comment.sql"+ , "0009.sql" ] findMigrationsDirectory :: IO FilePath@@ -319,6 +458,313 @@ "INSERT INTO kiroku.streams (stream_name, stream_version) VALUES ($1, 0)" (Encoders.param (Encoders.nonNullable Encoders.text)) Decoders.noResult++relationCatalogStatement :: Statement () (Text, Text, Bool, Text, Text, Text)+relationCatalogStatement =+ Statement.preparable+ """+ SELECT c.relkind::text,+ COALESCE(array_to_string(c.reloptions, ','), ''),+ c.relowner = source.relowner,+ view_contract.is_updatable::text,+ view_contract.is_insertable_into::text,+ pg_catalog.obj_description(c.oid, 'pg_class')+ FROM pg_catalog.pg_class AS c+ JOIN pg_catalog.pg_namespace AS n ON n.oid = c.relnamespace+ JOIN pg_catalog.pg_class AS source+ ON source.oid = to_regclass('kiroku.subscriptions')+ JOIN information_schema.views AS view_contract+ ON view_contract.table_schema = n.nspname+ AND view_contract.table_name = c.relname+ WHERE n.nspname = 'kiroku'+ AND c.relname = 'subscription_checkpoints_v1'+ """+ Encoders.noParams+ ( Decoders.singleRow+ ( (,,,,,)+ <$> column Decoders.text+ <*> column Decoders.text+ <*> column Decoders.bool+ <*> column Decoders.text+ <*> column Decoders.text+ <*> column Decoders.text+ )+ )+ where+ column = Decoders.column . Decoders.nonNullable++relationColumnsStatement :: Statement () [(Text, Text, Bool, Text)]+relationColumnsStatement =+ Statement.preparable+ """+ SELECT attribute.attname::text,+ pg_catalog.format_type(attribute.atttypid, attribute.atttypmod),+ attribute.attnotnull,+ pg_catalog.col_description(relation.oid, attribute.attnum)+ FROM pg_catalog.pg_class AS relation+ JOIN pg_catalog.pg_namespace AS namespace+ ON namespace.oid = relation.relnamespace+ JOIN pg_catalog.pg_attribute AS attribute+ ON attribute.attrelid = relation.oid+ WHERE namespace.nspname = 'kiroku'+ AND relation.relname = 'subscription_checkpoints_v1'+ AND attribute.attnum > 0+ AND NOT attribute.attisdropped+ ORDER BY attribute.attnum+ """+ Encoders.noParams+ ( Decoders.rowList+ ( (,,,)+ <$> column Decoders.text+ <*> column Decoders.text+ <*> column Decoders.bool+ <*> column Decoders.text+ )+ )+ where+ column = Decoders.column . Decoders.nonNullable++checkpointRowsStatement :: Statement () [(Text, Int32, Int64, UTCTime)]+checkpointRowsStatement =+ Statement.preparable+ """+ SELECT subscription_name,+ consumer_group_member,+ checkpoint_position,+ checkpoint_updated_at+ FROM kiroku.subscription_checkpoints_v1+ ORDER BY subscription_name, consumer_group_member+ """+ Encoders.noParams+ ( Decoders.rowList+ ( (,,,)+ <$> column Decoders.text+ <*> column Decoders.int4+ <*> column Decoders.int8+ <*> column Decoders.timestamptz+ )+ )+ where+ column = Decoders.column . Decoders.nonNullable++checkpointFixtureTimestamp :: UTCTime+checkpointFixtureTimestamp =+ UTCTime+ (fromGregorian 2026 8 13)+ (secondsToDiffTime (12 * 60 * 60 + 34 * 60 + 56))++checkpointFixtureSql :: Text+checkpointFixtureSql =+ """+ INSERT INTO kiroku.subscriptions+ (subscription_name, consumer_group_member, last_seen, updated_at)+ VALUES+ ('ordinary', 0, 11, '2026-08-13 12:34:56+00'),+ ('grouped', 0, 20, '2026-08-13 12:34:56+00'),+ ('grouped', 1, 21, '2026-08-13 12:34:56+00'),+ ('grouped', 2, 22, '2026-08-13 12:34:56+00')+ """++checkpointAdvanceSql :: Text+checkpointAdvanceSql =+ """+ UPDATE kiroku.subscriptions+ SET last_seen = 5+ WHERE subscription_name = 'ordinary'+ AND consumer_group_member = 0+ """++checkpointRegressionSql :: Text+checkpointRegressionSql =+ """+ UPDATE kiroku.subscriptions+ SET last_seen = 2+ WHERE subscription_name = 'ordinary'+ AND consumer_group_member = 0+ """++checkpointPosition :: Connection.Connection -> IO Int64+checkpointPosition connection =+ useSession connection (Session.statement "ordinary" checkpointPositionStatement)++checkpointPositionStatement :: Statement Text Int64+checkpointPositionStatement =+ Statement.preparable+ """+ SELECT checkpoint_position+ FROM kiroku.subscription_checkpoints_v1+ WHERE subscription_name = $1+ AND consumer_group_member = 0+ """+ (Encoders.param (Encoders.nonNullable Encoders.text))+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))++withTemporaryReaderRole :: Connection.Connection -> (Text -> IO value) -> IO value+withTemporaryReaderRole connection action = do+ unique <- newUnique+ let role =+ "kiroku_checkpoint_reader_"+ <> Text.replace "-" "n" (Text.pack (show (hashUnique unique)))+ identifier = quoteTestIdentifier role+ useSession connection (Session.script ("CREATE ROLE " <> identifier))+ action role+ `finally` useSession+ connection+ ( Session.script+ ( "RESET ROLE; DROP OWNED BY "+ <> identifier+ <> "; DROP ROLE "+ <> identifier+ )+ )++quoteTestIdentifier :: Text -> Text+quoteTestIdentifier value =+ "\"" <> Text.replace "\"" "\"\"" value <> "\""++checkpointRowCountStatement :: Statement () Int64+checkpointRowCountStatement =+ Statement.preparable+ "SELECT count(*) FROM kiroku.subscription_checkpoints_v1"+ Encoders.noParams+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))++privateTablePrivilegeStatement :: Statement () Bool+privateTablePrivilegeStatement =+ Statement.preparable+ "SELECT has_table_privilege(current_user, 'kiroku.subscriptions', 'SELECT')"+ Encoders.noParams+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))++privateTableCountStatement :: Statement () Int64+privateTableCountStatement =+ Statement.preparable+ "SELECT count(*) FROM kiroku.subscriptions"+ Encoders.noParams+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))++updatePublicViewStatement :: Statement () ()+updatePublicViewStatement =+ Statement.unpreparable+ """+ UPDATE kiroku.subscription_checkpoints_v1+ SET checkpoint_position = checkpoint_position+ """+ Encoders.noParams+ Decoders.noResult++hasSqlState :: Text -> Either Errors.SessionError value -> Bool+hasSqlState expected = \case+ Left+ ( Errors.StatementSessionError+ _+ _+ _+ _+ _+ (Errors.ServerStatementError (Errors.ServerError actual _ _ _ _))+ ) -> actual == expected+ _ -> False++downstreamReplacementFixtureSql :: Text+downstreamReplacementFixtureSql =+ """+ INSERT INTO kiroku.subscriptions+ (subscription_name, consumer_group_member, last_seen,+ updated_at)+ VALUES ('dependency', 0, 42, '2026-08-13 12:34:56+00');++ CREATE VIEW public.subscription_checkpoint_floors AS+ SELECT subscription_name,+ min(checkpoint_position) AS checkpoint_floor+ FROM kiroku.subscription_checkpoints_v1+ GROUP BY subscription_name;++ CREATE TABLE kiroku.subscription_checkpoint_storage_v2 (+ subscription_name text NOT NULL,+ consumer_group_member integer NOT NULL,+ checkpoint_position bigint NOT NULL,+ checkpoint_updated_at timestamptz NOT NULL,+ PRIMARY KEY (subscription_name, consumer_group_member)+ );++ INSERT INTO kiroku.subscription_checkpoint_storage_v2+ SELECT subscription_name,+ consumer_group_member,+ last_seen,+ updated_at+ FROM kiroku.subscriptions;++ CREATE OR REPLACE VIEW kiroku.subscription_checkpoints_v1+ (subscription_name,+ consumer_group_member,+ checkpoint_position,+ checkpoint_updated_at)+ WITH (security_invoker = false)+ AS+ WITH checkpoint_rows AS NOT MATERIALIZED (+ SELECT subscription_name,+ consumer_group_member,+ checkpoint_position,+ checkpoint_updated_at+ FROM kiroku.subscription_checkpoint_storage_v2+ )+ SELECT subscription_name,+ consumer_group_member,+ checkpoint_position,+ checkpoint_updated_at+ FROM checkpoint_rows;++ DROP TABLE kiroku.subscriptions;+ """++downstreamFloorsStatement :: Statement () [(Text, Int64)]+downstreamFloorsStatement =+ Statement.preparable+ """+ SELECT subscription_name, checkpoint_floor+ FROM public.subscription_checkpoint_floors+ ORDER BY subscription_name+ """+ Encoders.noParams+ ( Decoders.rowList+ ( (,)+ <$> Decoders.column (Decoders.nonNullable Decoders.text)+ <*> Decoders.column (Decoders.nonNullable Decoders.int8)+ )+ )++checkpointPlanFixtureSql :: Text+checkpointPlanFixtureSql =+ """+ INSERT INTO kiroku.subscriptions+ (subscription_name, consumer_group_member, last_seen)+ SELECT 'subscription-' || lpad((value % 100)::text, 3, '0'),+ value::integer,+ value::bigint+ FROM generate_series(1, 10000) AS generated(value);++ ANALYZE kiroku.subscriptions;+ """++serverVersionStatement :: Statement () Text+serverVersionStatement =+ Statement.preparable+ "SELECT current_setting('server_version')"+ Encoders.noParams+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.text)))++checkpointExplainStatement :: Statement () [Text]+checkpointExplainStatement =+ Statement.preparable+ """+ EXPLAIN (COSTS OFF)+ SELECT min(checkpoint_position)+ FROM kiroku.subscription_checkpoints_v1+ WHERE subscription_name = 'subscription-042'+ """+ Encoders.noParams+ (Decoders.rowList (Decoders.column (Decoders.nonNullable Decoders.text))) applyNativeSqlFromDisk :: Connection.Connection -> FilePath -> IO () applyNativeSqlFromDisk connection directory =