diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,21 @@ # Changelog for srtree-db +## 0.1.2.0 + +- **Split-DB architecture**: e-graph and per-dataset fit data now live in separate SQLite files to eliminate WAL bloat during fitting. Schema split into `egraphSchemaSQL` and `fitSchemaSQL`. +- **Blob format**: switched `cstore_page` from hex-encoded TEXT to binary BLOB; added `SqlBlob` and `loadPagesBulk` for bulk page preloading. +- **New CLI subcommands**: `ingest` (streaming line-by-line expression parser with batched inserts), `eqsat` (standalone eqsat on paged graphs), `fitdata` (resumable batch fitting with periodic commit checkpoints), `status` (fit summary per dataset), `refit` (clear and re-fit from scratch). +- **fitdata performance overhaul**: batch page preloading (`loadPagesBulk`), persistent page cache, sub-expression expansion for bulk dependency loading, analytical fitting for parameter-free expressions, parallel fitting with `setMTPopParallel`, `--no-header` flag (default True), `compileLossAndGrad` reuse across restarts. +- **Ingest performance**: `writeNode` writes class pages inline (O(1) per new class); removed `writeMissingPages`. Added `--reparam` flag for float constants to parameters. +- **Frontier re-saturation**: `frontier` table marks changed classes; matcher restricted to frontier via `cpsBeginFrontier`/`cpsEndFrontier`. +- **Page streaming**: `pushFit` streams the page store via `SqlBackend.streamPages` (SQLite cursor / Postgres grid fallback), bounded O(1) memory. +- **Legacy cleanup**: removed `fit` table; `enode_child` populated during eqsat by write-through. +- **NaN propagation**: parameter-less subexpressions that evaluate to NaN/Infinity detected analytically and propagated to ancestors. +- **DB bloat fixes**: `INSERT OR REPLACE` replaced with upsert; batched transactions; secondary indexes dropped; `PRAGMA journal_mode=DELETE` in fitdata. +- **In-memory vs DB eqsat equivalence test**: `testEquivInMemDB` proves paged and in-memory eqsat converge to the same merge structure. +- **Extract module**: `extractTreeFromDB` (standalone SRTree reconstruction from DB pages) and `reconstructFromCache` (pure IntMap-based). +- **Conversion script**: `tools/convert_db.py` converts single-DB to split format. + ## 0.1.1.0 - Added cli tools to populate and fit data into a database diff --git a/app/EqSat.hs b/app/EqSat.hs --- a/app/EqSat.hs +++ b/app/EqSat.hs @@ -21,7 +21,7 @@ import Algorithm.EqSat.Storage.SQLite (loadGraphLazy, saveGraph, flushStore) import Algorithm.EqSat.Storage.Query (getOrCreateDataset) -import Database.SQLite3 (Database, open, close) +import Database.SQLite3 (Database, open, close, exec) -- | CLI options for the eqsat sub-command. data EqSatOpts = EqSatOpts @@ -93,4 +93,9 @@ -- | Open a SQLite database, run an action, and close it. withSQLite :: String -> (Database -> IO a) -> IO a -withSQLite path = bracket (open (T.pack path)) close +withSQLite path f = bracket openDb close f + where + openDb = do + db <- open (T.pack path) + exec db "PRAGMA journal_mode=WAL" + pure db diff --git a/app/FitData.hs b/app/FitData.hs --- a/app/FitData.hs +++ b/app/FitData.hs @@ -1,56 +1,74 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE ScopedTypeVariables #-} module FitData ( FitDataOpts(..) , fitdataParser , runFitData + , runRefit ) where -import Control.Exception (bracket, SomeException, catch, displayException) -import Control.Monad.State.Strict (runStateT) +import Control.Concurrent (getNumCapabilities, threadDelay) +import Control.Concurrent.Async (mapConcurrently_, mapConcurrently) +import Control.Monad (replicateM, when, unless, void, forM_) +import Control.Exception (bracket, SomeException, catch, SomeAsyncException(..)) import Data.IORef -import Data.List (maximumBy) +import Data.List (maximumBy, foldl', sortBy) +import Data.Maybe (catMaybes) import Data.Ord (comparing) +import qualified Data.IntMap.Strict as IntMap +import qualified Data.IntSet as IntSet import qualified Data.Text as T import qualified Data.Vector.Unboxed as VU import Options.Applicative hiding (Const) +import System.IO (hPutStrLn, hFlush, stdout, stderr) import System.Random (randomRIO) import Data.SRTree (Fix(..), SRTree(..), Op(..), relabelParams, countParamsUniq, countNodes) import Data.SRTree.Print (showExpr) -import Data.SRTree.Eval (Target) +import Data.SRTree.Eval (Target, compile) import Data.SRTree.Datasets (loadDataset) -import Algorithm.SRTree.NonlinearOpt (minimizeNLL') -import Algorithm.SRTree.Likelihoods (Loss(..), Distribution(..), readLoss) +import Algorithm.SRTree.NonlinearOpt (compileLossAndGrad, minimizeNLLWith) +import Algorithm.SRTree.Likelihoods (Loss(..), Distribution(..)) import Algorithm.SRTree.AD (ADBackEnd(..)) +import Algorithm.SRTree.AD.Unboxed (setMTPopParallel) import Numeric.Optimization.NLOPT (LocalAlgorithm(..)) -import Algorithm.EqSat.Egraph (EGraph(..), EClassId, getBestExpr, canonical) +import Algorithm.EqSat.Egraph (EClassId, EClass(..)) import Algorithm.EqSat.Storage.Backend (SqlBackend(..), SqlValue(..), sqlToInt) -import Algorithm.EqSat.Storage.SQLite (loadGraphLazy, saveGraph, flushStore) +import Algorithm.EqSat.Storage.Extract (reconstructFromCache, expandTreeIds) +import Algorithm.EqSat.Storage.SQLite (loadPagesBulk) import Algorithm.EqSat.Storage.Query (getOrCreateDataset, writeDatasetFit) +import Algorithm.EqSat.Storage.Schema (createSchemaFit) import Algorithm.EqSat.Storage.Types (serializeTheta) -import Database.SQLite3 (Database, open, close) +import Database.SQLite3 (Database, open, close, exec) -- | CLI options for the fitdata sub-command. data FitDataOpts = FitDataOpts - { fitdataDb :: String - , fitdataDataset :: String - , fitdataData :: String - , fitdataLoss :: Loss - , fitdataHasHeader :: Bool - , fitdataNRep :: Int - , fitdataNIter :: Int - , fitdataBatchSize :: Int + { fitdataEgraph :: String + , fitdataFitdb :: String + , fitdataDataset :: String + , fitdataData :: String + , fitdataLoss :: Loss + , fitdataHasHeader :: Bool + , fitdataNRep :: Int + , fitdataNIter :: Int + , fitdataBatchSize :: Int + , fitdataQuiet :: Bool } deriving (Show) fitdataParser :: Parser FitDataOpts fitdataParser = FitDataOpts <$> strOption - ( long "db" + ( long "egraph" <> metavar "FILE" - <> help "SQLite database file path" ) + <> help "Path to e-graph database" ) <*> strOption + ( long "fitdb" + <> metavar "FILE" + <> help "Path to fit database" ) + <*> strOption ( long "dataset" <> metavar "NAME" <> help "Dataset name" ) @@ -63,9 +81,9 @@ <> value (NLL Gaussian) <> metavar "LOSS" <> help "Loss function (MSE, NLL Gaussian, etc.)" ) - <*> switch - ( long "has-header" - <> help "CSV has header row (default: True)" ) + <*> flag True False + ( long "no-header" + <> help "CSV has no header row (default: has header)" ) <*> option auto ( long "n-rep" <> value 1 @@ -73,21 +91,25 @@ <> help "Number of random restarts per expression" ) <*> option auto ( long "n-iter" - <> value 100 + <> value 30 <> metavar "N" <> help "Max NLopt iterations" ) <*> option auto ( long "batch-size" - <> value 100 + <> value 10000 <> metavar "N" - <> help "Fit N expressions per commit batch" ) + <> help "Fit N expressions per batch" ) + <*> switch + ( long "quiet" + <> short 'q' + <> help "Suppress per-expression output; print progress every 10k expressions" ) -- | Run the fitdata sub-command. runFitData :: FitDataOpts -> IO () runFitData opts = do let FitDataOpts{..} = opts - -- 1. Load dataset putStrLn $ "Loading dataset: " ++ fitdataData + hFlush stdout ((xTrain, yTrain, _xVal, _yVal), (mYErr, _), _varnames, _target) <- loadDataset fitdataData fitdataHasHeader @@ -96,103 +118,339 @@ NLL ROXY -> 3 _ -> 0 - -- 2. Load paged graph and query for unfitted e-classes - putStrLn $ "Loading paged graph from " ++ fitdataDb ++ "..." - withSQLite fitdataDb $ \db -> do - dsid <- getOrCreateDataset db fitdataDataset - er <- loadGraphLazy db dsid - case er of - Left err -> putStrLn $ "loadGraphLazy failed: " ++ err - Right eg -> do - -- Query for unfitted e-classes - unfitted <- queryUnfitted db dsid - let total = length unfitted - putStrLn $ "Found " ++ show total ++ " unfitted e-classes" + putStrLn $ "Opening egraph: " ++ fitdataEgraph ++ "..." + putStrLn $ "Opening fitdb: " ++ fitdataFitdb ++ "..." + hFlush stdout + withSQLite fitdataFitdb $ \fitDb -> do + createSchemaFit fitDb + withSQLite fitdataEgraph $ \egDb -> do + dsid <- getOrCreateDataset fitDb fitdataDataset - if total == 0 - then putStrLn "Nothing to fit." - else do - -- Process in batches - counter <- newIORef (0 :: Int) - let processBatch [] = pure () - processBatch batch = do - mapM_ (fitOne db dsid eg xTrain yTrain mYErr fitdataLoss fitdataNIter fitdataNRep nNoiseParams counter) batch - -- Commit checkpoint - flushStore eg - putStrLn $ " [checkpoint] committed batch" + total <- countUnfitted egDb fitDb dsid + putStrLn $ "Found " ++ show total ++ " unfitted e-classes" + hFlush stdout - let batches = chunk fitdataBatchSize unfitted - mapM_ processBatch batches + if total == 0 + then putStrLn "Nothing to fit." + else do + nCaps <- getNumCapabilities + counter <- newIORef (0 :: Int) + nanSet <- newIORef IntSet.empty + nanCount <- newIORef (0 :: Int) - fitted <- readIORef counter - putStrLn $ "Fitted " ++ show fitted ++ "/" ++ show total - ++ " expressions" + -- Phase 0-1: load pages + expand (reads from egraph DB) + -- Accumulates invalid/analytical fits in an IORef (no DB writes) + let loadPhase pendingRef batch = do + let batchIds = IntSet.fromList batch + batchPages <- loadPagesBulk egDb (IntSet.toList batchIds) + let !cache0 = batchPages + let expandLoop !cache = do + let needed = foldl' (\s eid -> s `IntSet.union` expandTreeIds cache eid) IntSet.empty batch + missing = IntSet.toList (IntSet.difference needed (IntSet.fromList (IntMap.keys cache))) + if null missing + then pure (cache, needed) + else do + putStrLn $ " Loading " ++ show (length missing) ++ " sub-expression pages..." + hFlush stdout + newPages <- loadPagesBulk egDb missing + expandLoop (cache `IntMap.union` newPages) + (cache1, needed) <- expandLoop cache0 --- | Fit a single e-class and write the result. -fitOne :: SqlBackend db - => db -> Int -> EGraph - -> [VU.Vector Double] -> VU.Vector Double -> Maybe (VU.Vector Double) - -> Loss -> Int -> Int -> Int -> IORef Int -> EClassId -> IO () -fitOne db dsid eg xTrain yTrain mYErr loss nIter nRep nNoiseParams counter eid = do - n <- readIORef counter - let label = "[" ++ show (n+1) ++ "]" - -- Extract best expression from the e-class - mTree <- runStateT (getBestExpr eid) eg >>= pure . Just . fst - case mTree of + -- Phase 2: build jobs (reconstruct, handle cache misses) + let toFit = IntSet.toList needed + mjobs <- mapM (buildJobNoWrite cache1 nNoiseParams counter pendingRef) toFit + let jobs = sortBy (comparing jobSize) (catMaybes mjobs) + + -- Phase 3: classify bottom-up, pruning NaN, collecting survivors + survivorRef <- newIORef ([] :: [FitJob]) + beforeNan <- readIORef nanCount + mapM_ (classifyNoWrite fitdataQuiet nanSet counter nanCount survivorRef pendingRef dsid cache1 xTrain yTrain mYErr fitdataLoss nNoiseParams) jobs + survivors <- readIORef survivorRef + afterNan <- readIORef nanCount + when (afterNan > beforeNan) $ + unless fitdataQuiet $ do + putStrLn $ " +" ++ show (afterNan - beforeNan) + ++ " eclasses inserted with NaN this batch (total " ++ show afterNan ++ ")" + hFlush stdout + pure survivors + + -- Phase 4-5: parallel NLopt + batch write (uses fitDb) + let fitPhase pendingRef survivors = do + let chunks = chunk nCaps survivors + setMTPopParallel False + results <- fmap concat $ mapConcurrently (mapM (fitOneNLopt fitdataQuiet xTrain yTrain mYErr fitdataLoss fitdataNIter fitdataNRep counter)) chunks + setMTPopParallel True + -- Batch write all pending fits (invalid + analytical + NLopt) + pending <- atomicModifyIORef' pendingRef (\ps -> ([], ps)) + execDb fitDb "BEGIN" + forM_ pending $ \(FitPending eid fit theta sz) -> + case fit of + Nothing -> writeInvalidFit fitDb dsid eid -- invalid: fitted=1, fitness=NULL + Just f -> writeDatasetFit fitDb dsid eid (Just f) Nothing theta sz + forM_ results $ \(FitResult eid fit theta sz) -> + writeDatasetFit fitDb dsid eid (Just fit) Nothing theta sz + execDb fitDb "COMMIT" + unless fitdataQuiet $ putStrLn " [checkpoint] committed batch" + + -- Stream IDs in batches, process sequentially (O(1) memory for IDs) + pendingRef <- newIORef ([] :: [FitPending]) + let processBatch batch = do + survivors <- loadPhase pendingRef batch + fitPhase pendingRef survivors + processStreamingBatches egDb fitDb dsid fitdataBatchSize processBatch + + fitted <- readIORef counter + putStrLn $ "Fitted " ++ show fitted ++ "/" ++ show total + ++ " expressions" + +-- | A unit of NLopt work: a reconstructed, relabeled expression with its +-- parameter count and node size precomputed. +data FitJob = FitJob + { jobEid :: !EClassId + , jobTree :: !(Fix SRTree) + , jobFree :: !Bool -- ^ structurally parameter-free (no Param nodes) + , jobNp :: !Int -- ^ total free params incl. loss noise params + , jobSize :: !Int + } + +-- | Result of fitting one expression (pure data, no DB side effects). +data FitResult = FitResult + { frEid :: !EClassId + , frFitness :: !Double + , frTheta :: !T.Text + , frSize :: !Int + } + +-- | A pending DB write (accumulated during loadPhase, written in fitPhase). +data FitPending = FitPending + { fpEid :: !EClassId + , fpFitness :: !(Maybe Double) + , fpTheta :: !T.Text + , fpSize :: !Int + } + +-- | Reconstruct a job, accumulating invalid writes instead of writing immediately. +buildJobNoWrite :: IntMap.IntMap EClass -> Int -> IORef Int -> IORef [FitPending] + -> EClassId -> IO (Maybe FitJob) +buildJobNoWrite cache nNoiseParams counter pendingRef eid = do + case reconstructFromCache cache eid of Nothing -> do - putStrLn $ label ++ " eclass " ++ show eid ++ ": could not extract expression (skipped)" - modifyIORef' counter (+1) + atomicModifyIORef' pendingRef (\ps -> (FitPending eid Nothing (T.pack (serializeTheta [])) 0 : ps, ())) + atomicModifyIORef' counter (\n -> let !n' = n + 1 in (n', ())) + pure Nothing Just tree -> do - let tree' = relabelParams tree - np = countParamsUniq tree' + nNoiseParams - sz = countNodes tree' - -- Skip expressions with no parameters (pure constants/variables) - if np == 0 - then do - putStrLn $ label ++ " eclass " ++ show eid ++ ": no parameters (skipped)" - modifyIORef' counter (+1) - else do - -- Try multiple random restarts, pick the best - results <- mapM (\_ -> fitOneRandom xTrain yTrain mYErr loss nIter tree' np) [1..nRep] - let (bestFitness, bestTheta) = maximumBy (comparing fst) results - -- Write result to DB - writeDatasetFit db dsid eid (Just bestFitness) Nothing - (T.pack (serializeTheta [bestTheta])) sz - putStrLn $ label ++ " eclass " ++ show eid ++ " (" ++ takeExpr tree' ++ "): fitness=" ++ showFit bestFitness - modifyIORef' counter (+1) + let !tree' = relabelParams tree + !nup = countParamsUniq tree' + !np = nup + nNoiseParams + !free = nup == 0 + !sz = countNodes tree' + pure (Just (FitJob eid tree' free np sz)) --- | Fit an expression with a random initial theta. -fitOneRandom :: [VU.Vector Double] -> VU.Vector Double -> Maybe (VU.Vector Double) - -> Loss -> Int -> Fix SRTree -> Int -> IO (Double, VU.Vector Double) -fitOneRandom xTrain yTrain mYErr loss nIter tree np = do - theta0 <- VU.replicateM np (randomRIO (-1, 1)) - let (theta, lossVal, _) = minimizeNLL' VAR1 SingleThread loss mYErr nIter xTrain yTrain tree theta0 - fitness = negate lossVal - pure (fitness, theta) +-- | Classify without writing to DB. Accumulates pending writes. +classifyNoWrite :: Bool -> IORef IntSet.IntSet -> IORef Int -> IORef Int + -> IORef [FitJob] -> IORef [FitPending] + -> Int -> IntMap.IntMap EClass + -> [VU.Vector Double] -> VU.Vector Double -> Maybe (VU.Vector Double) + -> Loss -> Int -> FitJob -> IO () +classifyNoWrite quiet nanSet counter nanCount survivorRef pendingRef dsid cache xTrain yTrain mYErr loss nNoiseParams (FitJob eid tree' free np sz) = do + let desc = expandTreeIds cache eid + nan <- readIORef nanSet + if not (IntSet.null (IntSet.intersection desc nan)) + then pruneNoWrite + else if not free + then survivor + else do + let fitness = analyticalFit loss xTrain yTrain tree' + if isInvalid fitness + then pruneNoWrite + else if np == 0 + then do + atomicModifyIORef' pendingRef (\ps -> (FitPending eid (Just fitness) (T.pack (serializeTheta [])) sz : ps, ())) + atomicModifyIORef' counter (\n -> let !n' = n + 1 in (n', ())) + unless quiet $ putStrLn $ " eclass " ++ show eid ++ " (" ++ takeExpr tree' ++ "): fitness=" ++ showFit fitness ++ " [analytical]" + else survivor + where + pruneNoWrite = do + atomicModifyIORef' pendingRef (\ps -> (FitPending eid Nothing (T.pack (serializeTheta [])) 0 : ps, ())) + atomicModifyIORef' nanSet (\s -> (IntSet.insert eid s, ())) + atomicModifyIORef' nanCount (\n -> let !n' = n + 1 in (n', ())) + atomicModifyIORef' counter (\n -> let !n' = n + 1 in (n', ())) + survivor = do + atomicModifyIORef' survivorRef (\xs -> (FitJob eid tree' free np sz : xs, ())) --- | Query for e-class IDs that are not yet fitted for a dataset. -queryUnfitted :: SqlBackend db => db -> Int -> IO [EClassId] -queryUnfitted db dsid = do - rows <- queryDb db - "SELECT e.eid FROM eclass e \ - \LEFT JOIN dataset_fit df ON df.eid = e.eid AND df.dataset_id = ? \ - \WHERE df.fitted IS NULL OR df.fitted = 0" +-- | Classify a single eclass (run in bottom-up order). If the expression +-- contains any known-NaN subexpression, or is a parameter-less expression that +-- evaluates to NaN/infinity, it is pruned (written as a NULL-fitness fitted row) +-- and recorded in @nanSet@ so its ancestors propagate. Otherwise it is queued as +-- an NLopt survivor. +classify :: SqlBackend db + => Bool -> IORef IntSet.IntSet -> IORef Int -> IORef Int -> IORef [FitJob] + -> db -> Int -> IntMap.IntMap EClass + -> [VU.Vector Double] -> VU.Vector Double -> Maybe (VU.Vector Double) + -> Loss -> Int -> FitJob -> IO () +classify quiet nanSet counter nanCount survivorRef db dsid cache xTrain yTrain mYErr loss nNoiseParams (FitJob eid tree' free np sz) = do + let desc = expandTreeIds cache eid + nan <- readIORef nanSet + if not (IntSet.null (IntSet.intersection desc nan)) + then prune + else if not free + then survivor + else do + -- Structurally parameter-free: evaluate analytically to decide NaN. + -- Handles the noise-param loss case (e.g. NLL Gaussian, np = nNoiseParams). + let fitness = analyticalFit loss xTrain yTrain tree' + if isInvalid fitness + then prune + else if np == 0 + then do + -- Fully parameter-less: analytic fit is exact. + writeDatasetFit db dsid eid (Just fitness) Nothing + (T.pack (serializeTheta [])) sz + atomicModifyIORef' counter (\n -> let !n' = n + 1 in (n', ())) + unless quiet $ putStrLn $ " eclass " ++ show eid ++ " (" ++ takeExpr tree' ++ "): fitness=" ++ showFit fitness ++ " [analytical]" + else survivor + where + -- An eclass doomed to NaN (contains a known-NaN subexpr, or is itself NaN): + -- write a NULL-fitness fitted row, record it, and count it. + prune = do + writeInvalidFit db dsid eid + atomicModifyIORef' nanSet (\s -> (IntSet.insert eid s, ())) + atomicModifyIORef' nanCount (\n -> let !n' = n + 1 in (n', ())) + atomicModifyIORef' counter (\n -> let !n' = n + 1 in (n', ())) + survivor = do + atomicModifyIORef' survivorRef (\xs -> (FitJob eid tree' free np sz : xs, ())) + +-- | Fit one expression via NLopt. Only called on survivors that are not +-- statically NaN. Runs @nRep@ restarts and keeps the best. +-- Writes result to DB immediately. +fitOne :: SqlBackend db + => Bool -> db -> Int + -> [VU.Vector Double] -> VU.Vector Double -> Maybe (VU.Vector Double) + -> Loss -> Int -> Int -> IORef Int -> FitJob -> IO () +fitOne quiet db dsid xTrain yTrain mYErr loss nIter nRep counter job = do + fr <- fitOneNLopt quiet xTrain yTrain mYErr loss nIter nRep counter job + writeDatasetFit db dsid (frEid fr) (Just (frFitness fr)) Nothing (frTheta fr) (frSize fr) + +-- | Pure NLopt fit (no DB side effects). Returns a FitResult. +fitOneNLopt :: Bool + -> [VU.Vector Double] -> VU.Vector Double -> Maybe (VU.Vector Double) + -> Loss -> Int -> Int -> IORef Int -> FitJob -> IO FitResult +fitOneNLopt quiet xTrain yTrain mYErr loss nIter nRep counter (FitJob eid tree' _ np sz) = do + let funAndGrad = compileLossAndGrad MultiThread loss mYErr xTrain yTrain tree' + runRestart = do + theta0 <- VU.replicateM np (randomRIO (-1, 1)) + let (theta, lossVal, _) = minimizeNLLWith funAndGrad VAR1 nIter theta0 + pure (negate lossVal, theta) + results <- replicateM nRep runRestart + let (bestFitness, bestTheta) = maximumBy (comparing fst) results + !thetaText = T.pack (serializeTheta [bestTheta]) + atomicModifyIORef' counter (\n -> let !n' = n + 1 in (n', ())) + unless quiet $ putStrLn $ " eclass " ++ show eid ++ " (" ++ takeExpr tree' ++ "): fitness=" ++ showFit bestFitness + pure (FitResult eid bestFitness thetaText sz) + +-- | Whether a fitness value is unusable (NaN or +/-Infinity), so it can be +-- pruned and propagated to ancestors. +isInvalid :: Double -> Bool +isInvalid f = isNaN f || isInfinite f + +-- | Write a fitted row with NULL fitness (used for pruned / cache-miss eclasses). +-- Marks @fitted = 1@ so the eclass is removed from the unfitted queue but stays +-- distinguishable from rows with a real fitness. +writeInvalidFit :: SqlBackend db => db -> Int -> EClassId -> IO () +writeInvalidFit db dsid eid = + runDb db + "INSERT INTO dataset_fit (dataset_id, eid, fitness, dl, theta, size, evaluated, fitted) \ + \VALUES (?, ?, NULL, NULL, '', 0, 0, 1) \ + \ON CONFLICT (dataset_id, eid) DO UPDATE SET \ + \fitness = NULL, dl = NULL, theta = '', size = 0, evaluated = 0, fitted = 1" + [SqlInteger (fromIntegral dsid), SqlInteger (fromIntegral eid)] + +-- | Analytical fitness for parameter-free expressions. +-- No NLopt needed — compute loss directly. +analyticalFit :: Loss -> [VU.Vector Double] -> VU.Vector Double -> Fix SRTree -> Double +analyticalFit loss xTrain yTrain tree = + let preds = compile xTrain tree VU.empty -- evaluate with empty theta + m = fromIntegral (VU.length yTrain) :: Double + residuals = VU.zipWith (-) preds yTrain + in case loss of + NLL Gaussian -> + let mse = VU.sum (VU.map (\r -> r * r) residuals) / m + sigma2 = mse + nll = negate (m / 2 * log (2 * pi * sigma2) + m / 2) + in if isNaN nll || isInfinite nll then -(1/0) else nll + MSE -> + let mse = VU.sum (VU.map (\r -> r * r) residuals) / m + in negate mse + _ -> + let mse = VU.sum (VU.map (\r -> r * r) residuals) / m + in negate mse -- fallback: use MSE + +-- | Count unfitted e-classes for a dataset (cross-DB: egraph + fit). +countUnfitted :: (SqlBackend db1, SqlBackend db2) => db1 -> db2 -> Int -> IO Int +countUnfitted egDb fitDb dsid = do + -- Total eclasses in egraph + totalRows <- queryDb egDb "SELECT COUNT(*) FROM eclass" [] + let totalEclasses = case totalRows of { [[cnt]] -> sqlToInt cnt; _ -> 0 } + -- Fitted eclasses in fit DB + fittedRows <- queryDb fitDb + "SELECT COUNT(*) FROM dataset_fit WHERE dataset_id = ? AND fitted = 1" [SqlInteger (fromIntegral dsid)] - pure [ sqlToInt eid | [eid] <- rows ] + let totalFitted = case fittedRows of { [[cnt]] -> sqlToInt cnt; _ -> 0 } + pure (totalEclasses - totalFitted) --- | Chunk a list into sub-lists of the given size. +-- | Load all fitted eclass IDs for a dataset into an IntSet. +loadFittedSet :: SqlBackend db => db -> Int -> IO IntSet.IntSet +loadFittedSet fitDb dsid = do + rows <- queryDb fitDb + "SELECT eid FROM dataset_fit WHERE dataset_id = ? AND fitted = 1" + [SqlInteger (fromIntegral dsid)] + pure $ IntSet.fromList [ sqlToInt eid | [eid] <- rows ] + +-- | Stream unfitted e-class IDs in batches, processing each batch +-- without materializing the full ID list in memory. +-- Cross-DB: streams from egraph, filters against fit DB. +processStreamingBatches :: (SqlBackend db1, SqlBackend db2) + => db1 -> db2 -> Int -> Int -> ([EClassId] -> IO ()) -> IO () +processStreamingBatches egDb fitDb dsid batchSize processBatch = do + fittedSet <- loadFittedSet fitDb dsid + batchRef <- newIORef ([] :: [EClassId]) + countRef <- newIORef (0 :: Int) + foldQueryDb egDb + "SELECT eid FROM eclass ORDER BY eid" + [] + () + (\() cols -> case cols of + [eidCol] -> do + let !eid = sqlToInt eidCol + if IntSet.member eid fittedSet + then pure () + else do + batch <- readIORef batchRef + let !batch' = eid : batch + n <- readIORef countRef + let !n' = n + 1 + writeIORef countRef n' + if n' >= batchSize + then do + processBatch (reverse batch') + writeIORef batchRef [] + writeIORef countRef 0 + else writeIORef batchRef batch' + pure () + _ -> pure ()) + remaining <- readIORef batchRef + when (not (null remaining)) $ processBatch (reverse remaining) + chunk :: Int -> [a] -> [[a]] chunk _ [] = [] chunk n xs = let (h, t) = splitAt n xs in h : chunk n t --- | Take first few chars of an expression for display. takeExpr :: Fix SRTree -> String takeExpr t | length s > 40 = take 40 s ++ "..." | otherwise = s where s = showExpr t --- | Format fitness for display. showFit :: Double -> String showFit f | f == (-1/0) = "-Infinity" @@ -200,6 +458,27 @@ | isNaN f = "NaN" | otherwise = show (fromIntegral (round (f * 1000) :: Int) / 1000 :: Double) --- | Open a SQLite database, run an action, and close it. withSQLite :: String -> (Database -> IO a) -> IO a -withSQLite path = bracket (open (T.pack path)) close +withSQLite path f = bracket openDb close f + where + openDb = do + db <- open (T.pack path) + exec db "PRAGMA journal_mode=DELETE" + exec db "PRAGMA busy_timeout = 30000" + pure db + +-- | Run refit: clear all fitted data for a dataset, then re-fit everything. +runRefit :: FitDataOpts -> IO () +runRefit opts = do + let FitDataOpts{..} = opts + putStrLn $ "Refitting dataset: " ++ fitdataDataset + putStrLn $ "Clearing previous fit data..." + hFlush stdout + withSQLite fitdataFitdb $ \fitDb -> do + createSchemaFit fitDb + dsid <- getOrCreateDataset fitDb fitdataDataset + runDb fitDb "DELETE FROM dataset_fit WHERE dataset_id = ?" [SqlInteger (fromIntegral dsid)] + putStrLn $ "Cleared fit data for dataset " ++ show dsid ++ "." + hFlush stdout + -- Now run the normal fitdata flow + runFitData opts diff --git a/app/Ingest.hs b/app/Ingest.hs --- a/app/Ingest.hs +++ b/app/Ingest.hs @@ -26,9 +26,9 @@ import Algorithm.SRTree.Likelihoods (Loss(..), Distribution(..)) import Algorithm.SRTree.AD (ADBackEnd(..)) import Numeric.Optimization.NLOPT (LocalAlgorithm(..)) -import Algorithm.EqSat.Storage.Import (importEqs, ImportSummary(..)) +import Algorithm.EqSat.Storage.Import (importEqs, importEqsInit, ImportSummary(..)) import Algorithm.EqSat.Storage.SQLite () -import Database.SQLite3 (Database, open, close) +import Database.SQLite3 (Database, open, close, exec) -- | CLI options for the ingest sub-command. data IngestOpts = IngestOpts @@ -43,6 +43,7 @@ , ingestEqsatSteps :: Int , ingestReparam :: Bool , ingestHasHeader :: Bool + , ingestQuiet :: Bool } deriving (Show) ingestParser :: Parser IngestOpts @@ -95,6 +96,10 @@ <*> switch ( long "has-header" <> help "CSV has header row (default: True)" ) + <*> switch + ( long "quiet" + <> short 'q' + <> help "Suppress per-expression output; print progress every 10k expressions" ) -- | Run the ingest sub-command. runIngest :: IngestOpts -> IO () @@ -108,6 +113,7 @@ alg = ingestFormat varnames = ingestVarnames batchSize = 1000 :: Int + progressInterval = if ingestQuiet then 10000 else batchSize -- Open the expression file (or stdin) h <- if null ingestExprs then pure stdin else openFile ingestExprs ReadMode @@ -115,7 +121,11 @@ -- Open DB putStrLn $ "Opening " ++ ingestDb ++ "..." db <- open (T.pack ingestDb) + exec db "PRAGMA journal_mode=WAL" + -- One-time schema + index setup (skipped by repeated importEqs calls) + importEqsInit db + -- Process line by line, batch and insert putStrLn "Processing expressions..." totalRef <- newIORef (0 :: Int) @@ -140,7 +150,7 @@ else case parseSR alg (B.pack varnames) False (B.pack line) of Left err -> do modifyIORef' failedRef (+1) - hPutStrLn stderr $ " FAILED: " ++ line ++ " -- " ++ err + unless ingestQuiet $ hPutStrLn stderr $ " FAILED: " ++ line ++ " -- " ++ err Right tree -> do modifyIORef' validRef (+1) modifyIORef' batchRef ((relabelParams tree, [], Nothing) :) @@ -148,8 +158,9 @@ when (length batch >= batchSize) $ do flushBatch v <- readIORef validRef - hPutStrLn stderr $ " ... " ++ show v ++ " expressions processed" - hFlush stderr + when (v `mod` progressInterval < batchSize) $ do + hPutStrLn stderr $ " ... " ++ show v ++ " expressions processed" + hFlush stderr loop = do done <- hIsEOF h diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -5,9 +5,10 @@ import Options.Applicative import Ingest (IngestOpts, ingestParser, runIngest) import EqSat (EqSatOpts, eqsatParser, runEqSatCmd) -import FitData (FitDataOpts, fitdataParser, runFitData) +import FitData (FitDataOpts, fitdataParser, runFitData, runRefit) +import Status (StatusOpts, statusParser, runStatus) -data Cmd = Ingest IngestOpts | EqSat EqSatOpts | FitData FitDataOpts +data Cmd = Ingest IngestOpts | EqSat EqSatOpts | FitData FitDataOpts | Refit FitDataOpts | Status StatusOpts main :: IO () main = execParser cmdParser >>= dispatch @@ -19,9 +20,13 @@ ( command "ingest" (Ingest <$> info (ingestParser <**> helper) (progDesc "Ingest expressions into DB")) <> command "eqsat" (EqSat <$> info (eqsatParser <**> helper) (progDesc "Run equality saturation")) <> command "fitdata" (FitData <$> info (fitdataParser <**> helper) (progDesc "Fit expressions to dataset")) + <> command "refit" (Refit <$> info (fitdataParser <**> helper) (progDesc "Clear fit data and re-fit all expressions")) + <> command "status" (Status <$> info (statusParser <**> helper) (progDesc "Show fit status for a dataset")) ) dispatch :: Cmd -> IO () dispatch (Ingest opts) = runIngest opts dispatch (EqSat opts) = runEqSatCmd opts dispatch (FitData opts) = runFitData opts +dispatch (Refit opts) = runRefit opts +dispatch (Status opts) = runStatus opts diff --git a/app/Status.hs b/app/Status.hs new file mode 100644 --- /dev/null +++ b/app/Status.hs @@ -0,0 +1,95 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} + +module Status + ( StatusOpts(..) + , statusParser + , runStatus + ) where + +import qualified Data.Text as T +import Options.Applicative +import System.IO (hPutStrLn, hFlush, stdout, stderr) + +import Algorithm.EqSat.Storage.Backend (SqlBackend(..), SqlValue(..), sqlToInt) +import Algorithm.EqSat.Storage.Schema (createSchemaFit) +import Algorithm.EqSat.Storage.SQLite () +import Database.SQLite3 (Database, open, close, exec) +import Control.Exception (bracket) + +data StatusOpts = StatusOpts + { statusEgraph :: String + , statusFitdb :: String + , statusDataset :: String + } deriving (Show) + +statusParser :: Parser StatusOpts +statusParser = StatusOpts + <$> strOption + ( long "egraph" + <> metavar "FILE" + <> help "Path to e-graph database" ) + <*> strOption + ( long "fitdb" + <> metavar "FILE" + <> help "Path to fit database" ) + <*> strOption + ( long "dataset" + <> metavar "NAME" + <> help "Dataset name" ) + +runStatus :: StatusOpts -> IO () +runStatus StatusOpts{..} = do + withSQLite statusFitdb $ \fitDb -> do + createSchemaFit fitDb + -- Look up dataset id (don't create if missing) + dsRows <- queryDb fitDb "SELECT id FROM dataset WHERE name = ?" + [SqlText (T.pack statusDataset)] + case dsRows of + [] -> do + putStrLn $ "Dataset '" ++ statusDataset ++ "' not found." + hFlush stdout + pure () + [[dsIdVal]] -> do + let dsid = sqlToInt dsIdVal + putStrLn $ "Dataset: " ++ statusDataset ++ " (id=" ++ show dsid ++ ")" + hFlush stdout + + withSQLite statusEgraph $ \egDb -> do + totalRows <- queryDb egDb "SELECT COUNT(*) FROM eclass" [] + let totalEclasses = case totalRows of { [[cnt]] -> sqlToInt cnt; _ -> 0 } + + fittedRows <- queryDb fitDb + "SELECT COUNT(*) FROM dataset_fit WHERE dataset_id = ? AND fitted = 1" + [SqlInteger (fromIntegral dsid)] + let totalFitted = case fittedRows of { [[cnt]] -> sqlToInt cnt; _ -> 0 } + + finiteRows <- queryDb fitDb + "SELECT COUNT(*) FROM dataset_fit WHERE dataset_id = ? AND fitted = 1 AND fitness IS NOT NULL" + [SqlInteger (fromIntegral dsid)] + let totalFinite = case finiteRows of { [[cnt]] -> sqlToInt cnt; _ -> 0 } + + prunedRows <- queryDb fitDb + "SELECT COUNT(*) FROM dataset_fit WHERE dataset_id = ? AND fitted = 1 AND fitness IS NULL" + [SqlInteger (fromIntegral dsid)] + let totalPruned = case prunedRows of { [[cnt]] -> sqlToInt cnt; _ -> 0 } + + let totalUnfitted = totalEclasses - totalFitted + + putStrLn $ " Total eclasses: " ++ show totalEclasses + putStrLn $ " Fitted: " ++ show totalFitted + putStrLn $ " Finite fitness: " ++ show totalFinite + putStrLn $ " Pruned (NULL): " ++ show totalPruned + putStrLn $ " Unfitted: " ++ show totalUnfitted + hFlush stdout + _ -> do + putStrLn $ "Dataset '" ++ statusDataset ++ "' query returned unexpected result." + hFlush stdout + +withSQLite :: String -> (Database -> IO a) -> IO a +withSQLite path f = bracket openDb close f + where + openDb = do + db <- open (T.pack path) + exec db "PRAGMA busy_timeout = 5000" + pure db diff --git a/src/Algorithm/EqSat/Storage/Backend.hs b/src/Algorithm/EqSat/Storage/Backend.hs --- a/src/Algorithm/EqSat/Storage/Backend.hs +++ b/src/Algorithm/EqSat/Storage/Backend.hs @@ -22,13 +22,16 @@ , sqlToInt , sqlToMaybeDouble , sqlToText + , sqlToBlob ) where import Data.Int (Int64) import Data.Maybe (fromMaybe, listToMaybe) import Data.Text (Text) import qualified Data.Text as T +import qualified Data.Text.Encoding as TE import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy as BL import Algorithm.EqSat.Egraph (EClassId) @@ -36,6 +39,7 @@ data SqlValue = SqlInteger Int64 | SqlFloat Double | SqlText Text + | SqlBlob BS.ByteString | SqlNull deriving (Eq, Show) @@ -54,6 +58,10 @@ insertIgnore :: db -> Text -> [SqlValue] -> IO () -- | Run a parameterized query and return the raw result grid. queryDb :: db -> Text -> [SqlValue] -> IO [[SqlValue]] + -- | Fold over query results one row at a time without materializing the + -- full result list. This is O(1) memory in the row accumulator (unlike + -- 'queryDb' which builds a spine-strict list of all rows). + foldQueryDb :: db -> Text -> [SqlValue] -> a -> (a -> [SqlValue] -> IO a) -> IO a -- | Stream (bounded) the distinct e-class ids whose e-class contains a node -- with the given @op_detail@, for the streaming matcher, skipping any ids in -- @exclude@ (the already-attempted seen-set, so the per-rule budget advances @@ -67,8 +75,10 @@ -- implement this; others fall back to a grid 'queryDb' (unbounded, documented). -- The blob is delivered hex-decoded (raw) to the callback. streamPages :: db -> Text -> (Int64 -> BS.ByteString -> IO ()) -> IO () - -- | Create the schema (tables, indexes) for this driver. + -- | Create the egraph schema (tables, indexes) for this driver. createSchemaDb :: db -> IO () + -- | Create the fit dataset schema (tables, indexes) for this driver. + createSchemaDbFit :: db -> IO () sqlToInt :: SqlValue -> Int sqlToInt (SqlInteger n) = fromIntegral n @@ -88,4 +98,10 @@ sqlToText (SqlText t) = t sqlToText (SqlInteger n) = T.pack (show n) sqlToText (SqlFloat d) = T.pack (show d) +sqlToText (SqlBlob bs) = TE.decodeUtf8 bs sqlToText SqlNull = "" + +sqlToBlob :: SqlValue -> BS.ByteString +sqlToBlob (SqlBlob bs) = bs +sqlToBlob (SqlText t) = TE.encodeUtf8 t +sqlToBlob _ = BS.empty diff --git a/src/Algorithm/EqSat/Storage/ClassStore.hs b/src/Algorithm/EqSat/Storage/ClassStore.hs --- a/src/Algorithm/EqSat/Storage/ClassStore.hs +++ b/src/Algorithm/EqSat/Storage/ClassStore.hs @@ -17,9 +17,8 @@ -- O(n) per call). -- * Every access refreshes recency (true LRU), so the hot classes a -- rebuild/rewrite pass revisits stay resident. --- * Pages are stored hex-encoded in a TEXT column so the same DDL/CRUD --- runs unchanged on SQLite and PostgreSQL (a bytea/BLOB column is a --- possible later optimization). +-- * Pages are stored as binary BLOBs for compact storage and zero-copy +-- reads. -- -- Driver-neutrality note: the store talks only through 'SqlBackend' and -- spells writes as DELETE+INSERT inside one transaction (both drivers @@ -49,8 +48,6 @@ , clearFrontier , setFrontierActive , initFrontier - , hex - , unhex ) where import Control.Monad (forM_, unless, when) @@ -66,32 +63,13 @@ import qualified Data.ByteString.Lazy as BL import Algorithm.EqSat.Storage.Backend - ( SqlValue(..), SqlBackend(..), sqlToInt, sqlToText ) + ( SqlValue(..), SqlBackend(..), sqlToInt, sqlToText, sqlToBlob ) import Algorithm.EqSat.Storage.Types ( enodeKey, enodeOpTag, enodeOpDetail, opDetailOf ) import Algorithm.EqSat.Egraph ( EClass, EClassPageStore(..), ENode(..), _eClassId ) -- --------------------------------------------------------------------------- --- hex encoding of page blobs (driver-neutral TEXT storage) - -hex :: BS.ByteString -> T.Text -hex = T.pack . concatMap go . BS.unpack - where - go b = - let hi = fromIntegral (b `div` 16) - lo = fromIntegral (b `mod` 16) - in "0123456789abcdef" !! hi : ["0123456789abcdef" !! lo] - -unhex :: T.Text -> BS.ByteString -unhex = BS.pack . go . T.unpack - where - go (a:b:r) = fromIntegral (hexv a * 16 + hexv b) : go r - go _ = [] - hexv c | c >= '0' && c <= '9' = fromEnum c - fromEnum '0' - | otherwise = fromEnum c - fromEnum 'a' + 10 - --- --------------------------------------------------------------------------- -- LRU page cache data PageCache = PageCache @@ -144,7 +122,7 @@ -- --------------------------------------------------------------------------- -- store --- | A paging store over a table @key TEXT PRIMARY KEY, blob TEXT NOT NULL@, +-- | A paging store over a table @key INTEGER PRIMARY KEY, blob BLOB NOT NULL@, -- connected to a 'SqlBackend' database. data PageStore db = PageStore { psDb :: db @@ -167,7 +145,7 @@ newPageStore db tbl cap flushEvery' = do execDb db ("CREATE TABLE IF NOT EXISTS " <> tbl <> - " (key TEXT PRIMARY KEY, blob TEXT NOT NULL)") + " (key INTEGER PRIMARY KEY, blob BLOB NOT NULL)") cache <- newIORef (emptyCache cap) nodes <- newIORef Set.empty canons <- newIORef IM.empty @@ -197,8 +175,12 @@ [SqlInteger (fromIntegral eid)] case rows of [] -> pure Nothing + [[SqlBlob page]] -> do + writeIORef (psCache ps) (insertEvict eid page c0) + pure (Just page) [[SqlText hv]] -> do - let page = unhex hv + -- backward compat: old databases may still have hex-encoded TEXT + let page = sqlToBlob (SqlText hv) writeIORef (psCache ps) (insertEvict eid page c0) pure (Just page) _ -> fail "ClassStore.readPage: unexpected row shape" @@ -236,7 +218,7 @@ runDb db ("DELETE FROM " <> tbl <> " WHERE key = ?") [SqlInteger (fromIntegral eid)] runDb db ("INSERT INTO " <> tbl <> " (key, blob) VALUES (?, ?)") - [ SqlInteger (fromIntegral eid), SqlText (hex page) ] + [ SqlInteger (fromIntegral eid), SqlBlob page ] execDb db "COMMIT" modifyIORef' (psCache ps) (\cc -> cc { pcPend = IM.empty, pcPendN = 0 }) flushNodes ps @@ -312,7 +294,7 @@ allPages :: SqlBackend db => PageStore db -> IO [(Int, BS.ByteString)] allPages ps = do rows <- queryDb (psDb ps) ("SELECT key, blob FROM " <> psTable ps) [] - pure [ (sqlToInt k, unhex (sqlToText b)) | [k, b] <- rows ] + pure [ (sqlToInt k, sqlToBlob b) | [k, b] <- rows ] -- | Read every e-class id currently stored in the page table (keys only). This -- does NOT load the page blobs, so callers that only need the id set (e.g. diff --git a/src/Algorithm/EqSat/Storage/Extract.hs b/src/Algorithm/EqSat/Storage/Extract.hs new file mode 100644 --- /dev/null +++ b/src/Algorithm/EqSat/Storage/Extract.hs @@ -0,0 +1,314 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE BangPatterns #-} + +-- | Standalone SRTree reconstruction from the relational DB, without loading +-- the full e-graph. Walks @cstore_page@ blobs one class at a time, resolving +-- children recursively. Memory is O(depth) — no page cache, no in-memory maps. +module Algorithm.EqSat.Storage.Extract + ( extractTreeFromDB + , extractBestFromDB + , readPage + , reconstructFromCache + , expandTreeIds + ) where + +import Data.Binary (decode) +import qualified Data.ByteString.Lazy as BL +import qualified Data.IntMap as IntMap +import qualified Data.HashSet as Set +import Data.IntSet (IntSet) +import qualified Data.IntSet as IntSet + +import Data.SRTree (Fix(..), SRTree(..), Op(..), Function(..)) +import Algorithm.EqSat.Egraph (EClassId, EClass(..), EClassData(..), ENode(..), NOp(..), toOp) +import Algorithm.EqSat.Storage.Backend + ( SqlBackend(..), SqlValue(..), sqlToInt, sqlToBlob ) +import Algorithm.EqSat.Storage.ClassStore (classStoreTable) + +-- | Reconstruct a 'Fix SRTree' for the given e-class by walking the page +-- store and relational tables directly. No 'EGraph' state is needed. +-- +-- Returns 'Nothing' if the e-class has no page blob or the tree exceeds the +-- expansion budget (200 nodes, same as 'getBestExprBounded'). +extractTreeFromDB :: SqlBackend db => db -> EClassId -> IO (Maybe (Fix SRTree)) +extractTreeFromDB db root = go IntSet.empty 0 root + where + budget :: Int + budget = 200 + + go :: IntSet -> Int -> EClassId -> IO (Maybe (Fix SRTree)) + go _ n _ | n >= budget = pure Nothing + go seen n eid + | IntSet.member eid seen = pure Nothing + | otherwise = do + mPage <- readPage db eid + case mPage of + Nothing -> pure Nothing + Just page -> do + let ec = decode page :: EClass + nodes = Set.toList (_eNodes ec) + case nodes of + [] -> pure Nothing + (en : _) -> expandNode (IntSet.insert eid seen) n en + + expandNode :: IntSet -> Int -> ENode -> IO (Maybe (Fix SRTree)) + expandNode _ _ (EVar ix) = pure (Just (Fix (Var ix))) + expandNode _ _ (EParam ix) = pure (Just (Fix (Param ix))) + expandNode _ _ (EConst x) = pure (Just (Fix (Const x))) + expandNode seen n (EUni f t) = do + mt <- go seen (n + 1) t + case mt of + Nothing -> pure Nothing + Just t' -> pure (Just (Fix (Uni f t'))) + expandNode seen n (EBin op l r) = do + ml <- go seen (n + 1) l + case ml of + Nothing -> pure Nothing + Just l' -> do + mr <- go seen (n + 1) r + case mr of + Nothing -> pure Nothing + Just r' -> pure (Just (Fix (Bin op l' r'))) + expandNode seen n (ENAry op m) = do + let children = IntMap.toAscList m + mts <- expandNary seen n children + pure $ naryTree op <$> mts + + -- Expand each child in the ENAry multiset, collecting results. + -- Each child is expanded once, then replicated by its multiplicity. + expandNary :: IntSet -> Int -> [(EClassId, Int)] -> IO (Maybe [Fix SRTree]) + expandNary _ _ [] = pure (Just []) + expandNary seen n ((cid, cnt) : rest) = do + mc <- go seen (n + 1) cid + case mc of + Nothing -> pure Nothing + Just c -> do + mrest <- expandNary seen (n + 1) rest + case mrest of + Nothing -> pure Nothing + Just rs -> pure (Just (replicate (min cnt (budget - n)) c ++ rs)) + + -- Right-fold a list of child expressions into a binary Fix SRTree, + -- then normalize Sub/Div (same as Egraph.naryTree). + naryTree :: NOp -> [Fix SRTree] -> Fix SRTree + naryTree _ [] = Fix (Var 0) + naryTree op ts = normalizeSubDiv (foldr1 (\a b -> Fix (Bin (toOp op) a b)) ts) + + normalizeSubDiv :: Fix SRTree -> Fix SRTree + normalizeSubDiv (Fix (Bin Add l r)) = case pick l r of + Just (pos, neg) -> Fix (Bin Sub pos neg) + Nothing -> Fix (Bin Add (normalizeSubDiv l) (normalizeSubDiv r)) + where + pick a b = case negated a of + Just t -> Just (b, t) + Nothing -> case negated b of + Just t -> Just (a, t) + Nothing -> Nothing + negated (Fix (Bin Mul (Fix (Const c)) t)) | c == -1 = Just t + negated (Fix (Bin Mul t (Fix (Const c)))) | c == -1 = Just t + negated (Fix (Const c)) | c < 0 = Just (Fix (Const (-c))) + negated _ = Nothing + normalizeSubDiv (Fix (Bin Mul l r)) = case pick l r of + Just (num, den) -> Fix (Bin Div num den) + Nothing -> Fix (Bin Mul (normalizeSubDiv l) (normalizeSubDiv r)) + where + pick a b = case a of + Fix (Uni Recip t) -> Just (b, t) + _ -> case b of + Fix (Uni Recip t) -> Just (a, t) + _ -> Nothing + normalizeSubDiv (Fix (Uni f t)) = Fix (Uni f (normalizeSubDiv t)) + normalizeSubDiv t = t + +-- | Like 'extractTreeFromDB' but follows @_best@ pointers (the cost-minimal +-- e-node chosen by eqsat) instead of taking the first node from @_eNodes@. +-- This is what 'getBestExpr' does, but without loading the full EGraph -- +-- pages are read directly from the DB, one per class, O(depth) memory. +extractBestFromDB :: SqlBackend db => db -> EClassId -> IO (Maybe (Fix SRTree)) +extractBestFromDB db root = go IntSet.empty 0 root + where + budget :: Int + budget = 200 + + go :: IntSet -> Int -> EClassId -> IO (Maybe (Fix SRTree)) + go _ n _ | n >= budget = pure Nothing + go seen n eid + | IntSet.member eid seen = pure Nothing + | otherwise = do + mPage <- readPage db eid + case mPage of + Nothing -> pure Nothing + Just page -> do + let ec = decode page :: EClass + best = _best (_info ec) + expandNode (IntSet.insert eid seen) n best + + expandNode :: IntSet -> Int -> ENode -> IO (Maybe (Fix SRTree)) + expandNode _ _ (EVar ix) = pure (Just (Fix (Var ix))) + expandNode _ _ (EParam ix) = pure (Just (Fix (Param ix))) + expandNode _ _ (EConst x) = pure (Just (Fix (Const x))) + expandNode seen n (EUni f t) = do + mt <- go seen (n + 1) t + case mt of + Nothing -> pure Nothing + Just t' -> pure (Just (Fix (Uni f t'))) + expandNode seen n (EBin op l r) = do + ml <- go seen (n + 1) l + case ml of + Nothing -> pure Nothing + Just l' -> do + mr <- go seen (n + 1) r + case mr of + Nothing -> pure Nothing + Just r' -> pure (Just (Fix (Bin op l' r'))) + expandNode seen n (ENAry op m) = do + let children = IntMap.toAscList m + mts <- expandNary seen n children + pure $ naryTree op <$> mts + + expandNary :: IntSet -> Int -> [(EClassId, Int)] -> IO (Maybe [Fix SRTree]) + expandNary _ _ [] = pure (Just []) + expandNary seen n ((cid, cnt) : rest) = do + mc <- go seen (n + 1) cid + case mc of + Nothing -> pure Nothing + Just c -> do + mrest <- expandNary seen (n + 1) rest + case mrest of + Nothing -> pure Nothing + Just rs -> pure (Just (replicate (min cnt (budget - n)) c ++ rs)) + + naryTree :: NOp -> [Fix SRTree] -> Fix SRTree + naryTree _ [] = Fix (Var 0) + naryTree op ts = normSubDiv (foldr1 (\a b -> Fix (Bin (toOp op) a b)) ts) + + normSubDiv :: Fix SRTree -> Fix SRTree + normSubDiv (Fix (Bin Add l r)) = case pick l r of + Just (pos, neg) -> Fix (Bin Sub pos neg) + Nothing -> Fix (Bin Add (normSubDiv l) (normSubDiv r)) + where + pick a b = case negated a of + Just t -> Just (b, t) + Nothing -> case negated b of + Just t -> Just (a, t) + Nothing -> Nothing + negated (Fix (Bin Mul (Fix (Const c)) t)) | c == -1 = Just t + negated (Fix (Bin Mul t (Fix (Const c)))) | c == -1 = Just t + negated (Fix (Const c)) | c < 0 = Just (Fix (Const (-c))) + negated _ = Nothing + normSubDiv (Fix (Bin Mul l r)) = case pick l r of + Just (num, den) -> Fix (Bin Div num den) + Nothing -> Fix (Bin Mul (normSubDiv l) (normSubDiv r)) + where + pick a b = case a of + Fix (Uni Recip t) -> Just (b, t) + _ -> case b of + Fix (Uni Recip t) -> Just (a, t) + _ -> Nothing + normSubDiv (Fix (Uni f t)) = Fix (Uni f (normSubDiv t)) + normSubDiv t = t + +-- | Read a single page blob for an e-class (raw binary, no decoding). +readPage :: SqlBackend db => db -> EClassId -> IO (Maybe BL.ByteString) +readPage db eid = do + rows <- queryDb db + ("SELECT blob FROM " <> classStoreTable <> " WHERE key = ?") + [SqlInteger (fromIntegral eid)] + case rows of + [[SqlBlob bs]] -> pure (Just (BL.fromStrict bs)) + [[SqlText hv]] -> pure (Just (BL.fromStrict (sqlToBlob (SqlText hv)))) -- backward compat + _ -> pure Nothing + +-- | Pure SRTree reconstruction from a pre-loaded IntMap cache. +-- No IO, no SQL — O(1) per node lookup. +-- +-- Returns 'Nothing' if the e-class is not in the cache or the tree exceeds +-- the expansion budget (200 nodes). +reconstructFromCache :: IntMap.IntMap EClass -> EClassId -> Maybe (Fix SRTree) +reconstructFromCache cache root = go IntSet.empty 0 root + where + go seen n eid + | n >= 200 = Nothing + | IntSet.member eid seen = Nothing + | otherwise = case IntMap.lookup eid cache of + Nothing -> Nothing + Just ec -> + let nodes = Set.toList (_eNodes ec) + in case nodes of + [] -> Nothing + (en : _) -> expandNode (IntSet.insert eid seen) n en + + expandNode _ _ (EVar ix) = Just (Fix (Var ix)) + expandNode _ _ (EParam ix) = Just (Fix (Param ix)) + expandNode _ _ (EConst x) = Just (Fix (Const x)) + expandNode seen n (EUni f t) = Fix . Uni f <$> go seen (n + 1) t + expandNode seen n (EBin op l r) = do + l' <- go seen (n + 1) l + r' <- go seen (n + 1) r + pure (Fix (Bin op l' r')) + expandNode seen n (ENAry op m) = do + let children = IntMap.toAscList m + ts <- expandNary seen n op children + pure (naryTree op ts) + + expandNary _ _ _ [] = Just [] + expandNary seen n op ((cid, cnt) : rest) = do + c <- go seen (n + 1) cid + rs <- expandNary seen (n + 1) op rest + pure (replicate (min cnt (200 - n)) c ++ rs) + + naryTree _ [] = Fix (Var 0) + naryTree op ts = normalizeSubDiv (foldr1 (\a b -> Fix (Bin (toOp op) a b)) ts) + + normalizeSubDiv (Fix (Bin Add l r)) = case pick l r of + Just (pos, neg) -> Fix (Bin Sub pos neg) + Nothing -> Fix (Bin Add (normalizeSubDiv l) (normalizeSubDiv r)) + where + pick a b = case negated a of + Just t -> Just (b, t) + Nothing -> case negated b of + Just t -> Just (a, t) + Nothing -> Nothing + negated (Fix (Bin Mul (Fix (Const c)) t)) | c == -1 = Just t + negated (Fix (Bin Mul t (Fix (Const c)))) | c == -1 = Just t + negated (Fix (Const c)) | c < 0 = Just (Fix (Const (-c))) + negated _ = Nothing + normalizeSubDiv (Fix (Bin Mul l r)) = case pick l r of + Just (num, den) -> Fix (Bin Div num den) + Nothing -> Fix (Bin Mul (normalizeSubDiv l) (normalizeSubDiv r)) + where + pick a b = case a of + Fix (Uni Recip t) -> Just (b, t) + _ -> case b of + Fix (Uni Recip t) -> Just (a, t) + _ -> Nothing + normalizeSubDiv (Fix (Uni f t)) = Fix (Uni f (normalizeSubDiv t)) + normalizeSubDiv t = t + +-- | Collect all eclass IDs referenced by the tree rooted at @eid@, +-- including the root itself. Used to pre-expand dependencies for bulk loading. +expandTreeIds :: IntMap.IntMap EClass -> EClassId -> IntSet +expandTreeIds cache root = go IntSet.empty 0 root + where + go seen n eid + | n >= 200 = seen + | IntSet.member eid seen = seen + | otherwise = case IntMap.lookup eid cache of + Nothing -> IntSet.insert eid seen + Just ec -> + let seen' = IntSet.insert eid seen + nodes = Set.toList (_eNodes ec) + in case nodes of + [] -> seen' + (en : _) -> expandNode seen' n en + + expandNode seen _ (EVar _) = seen + expandNode seen _ (EParam _) = seen + expandNode seen _ (EConst _) = seen + expandNode seen n (EUni _ t) = go seen (n + 1) t + expandNode seen n (EBin _ l r) = + let !seen' = go seen (n + 1) l + in go seen' (n + 1) r + expandNode seen n (ENAry _ m) = + foldl' (\s (cid, _) -> go s (n + 1) cid) seen (IntMap.toAscList m) diff --git a/src/Algorithm/EqSat/Storage/Import.hs b/src/Algorithm/EqSat/Storage/Import.hs --- a/src/Algorithm/EqSat/Storage/Import.hs +++ b/src/Algorithm/EqSat/Storage/Import.hs @@ -8,13 +8,12 @@ -- The import holds **no** graph-size data in RAM: e-nodes are content-addressed -- against the @enode@/@eclass_node@ tables (the @enode_key@ is their content -- address), child lookups for n-ary flattening read each child's node from the --- DB, and parent edges are written straight to the @parent@ table. Only a --- scalar next-id counter is kept in memory, so peak memory is bounded (one --- e-class at a time) regardless of how many expressions are imported. +-- DB. Only a scalar next-id counter is kept in memory, so peak memory is +-- bounded (one e-class at a time) regardless of how many expressions are +-- imported. -- --- Class pages (@cstore_page@) are written at the end in a single linear pass by --- reconstructing each class from the relational tables, so hot classes are --- never rewritten repeatedly. +-- Class pages (@cstore_page@) are written inline during the batch fold, so +-- each new class gets its page in O(1) — no post-pass needed. -- -- The produced database is byte-compatible with 'saveGraph': the same page -- blobs and relational rows, so 'loadGraphLazy' / 'dbEqSat' work on it @@ -23,6 +22,7 @@ module Algorithm.EqSat.Storage.Import ( ImportSummary(..) , importEqs + , importEqsInit , recordExpressionIndex ) where @@ -45,7 +45,7 @@ ( EClassId, ENode(..), NOp(..), EClass(..), EClassData(..), Consts(..) ) import Algorithm.EqSat.Storage.Backend ( SqlValue(..), SqlBackend(..), sqlToInt, sqlToMaybeDouble, sqlToText ) -import Algorithm.EqSat.Storage.ClassStore (classStoreTable, hex) +import Algorithm.EqSat.Storage.ClassStore (classStoreTable) import Algorithm.EqSat.Storage.Types ( enodeKey, enodeOpTag, enodeOpDetail, serializeTheta, parseTheta, parseEnodeKey ) import Algorithm.EqSat.Storage.Schema (createSchema) @@ -64,42 +64,41 @@ { stNextId :: !Int } +-- | One-time schema + index setup for import. Call this once before the first +-- 'importEqs' call (e.g. in the ingest CLI before the batch loop) so that +-- repeated 'importEqs' calls skip redundant DDL. +importEqsInit :: SqlBackend db => db -> IO () +importEqsInit db = do + createSchema db + execDb db "CREATE INDEX IF NOT EXISTS idx_eclass_node_enode_key ON eclass_node(enode_key)" + -- | Insert a list of @(expression, theta, fitness)@ into the database, -- structurally expanding every subexpression into its own e-class, then write --- the class pages in a final linear pass. Runs inside a single transaction +-- only the pages for newly created classes. Runs inside a single transaction -- (rolled back on error). -- -- When a dataset name is provided (@Just ds@), dataset_fit rows and -- expression_index entries are written so fitness queries and dedup work. -- When @Nothing@, only the structural e-graph is built (enode, eclass, -- parent, cstore_page, meta) — the e-graph is reusable across datasets. +-- +-- Call 'importEqsInit' once before the first invocation to set up the schema +-- and indexes; subsequent calls skip the DDL. importEqs :: SqlBackend db => db -> Maybe String -> [(Fix SRTree, [Target], Maybe Double)] -> IO (Either String ImportSummary) importEqs db mds eqs = do createSchema db - -- content-address dedup queries by enode_key, but eclass_node's PK is - -- (eid, enode_key); index enode_key so those lookups are O(log n) not a scan. execDb db "CREATE INDEX IF NOT EXISTS idx_eclass_node_enode_key ON eclass_node(enode_key)" mdsid <- traverse (getOrCreateDataset db) mds - -- Read the current next_id from the meta table so repeated importEqs calls - -- don't collide on eclass eids. curNextId <- readMetaNextId db ref <- newIORef (ImportState curNextId) r <- try $ do execDb db "BEGIN" - -- stream the expression list through a fold (rather than forM_ + length - -- eqs) so the lazy list is unreferenced after consumption and GC'd as it - -- is processed; the fold accumulator carries the count, so we never retain - -- the whole parsed expression list in memory. n <- foldM (\c (t, theta, fit) -> do (eid, h) <- insertTree ref db fit mdsid t - -- dataset_fit and expression_index only when a dataset is given case mdsid of Nothing -> pure () Just dsid -> do writeDatasetFit db dsid eid fit Nothing (T.pack (serializeTheta theta)) h - -- record the root expression in the registry (keyed by its - -- canonical root node) so "was this expression seen/tested?" - -- can be answered per dataset. mroot <- lookupClassNode db eid forM_ mroot $ \en -> runDb db @@ -109,7 +108,6 @@ , SqlInteger (fromIntegral dsid) ] pure (c + 1)) 0 eqs writeMeta db ref - writeAllPages db execDb db "COMMIT" pure n case r of @@ -118,7 +116,7 @@ pure (Left ("importEqs failed: " <> displayException e)) Right n -> do st <- readIORef ref - pure (Right (ImportSummary (stNextId st) (stNextId st) n)) + pure (Right (ImportSummary (stNextId st) (stNextId st - curNextId) n)) -- | Insert a full tree bottom-up, returning its root e-class id and height. insertTree :: SqlBackend db => IORef ImportState -> db -> Maybe Double -> Maybe Int -> Fix SRTree -> IO (EClassId, Int) @@ -188,8 +186,9 @@ . IntMap.fromListWith (\(n1, h1) (n2, h2) -> (n1 + n2, max h1 h2)) . map (\(c, n, h) -> (c, (n, h))) --- | Write the relational rows for a brand-new e-node. Reverse parent edges go --- straight to the @parent@ table (reconstructed into pages by 'writeAllPages'). +-- | Write the relational rows for a brand-new e-node AND its class page. +-- This is O(1) per new class — the page is written inline during the batch fold, +-- eliminating the O(n) post-pass that 'writeMissingPages' used to do. writeNode :: SqlBackend db => db -> EClassId -> ENode -> String -> [(EClassId, Int, Int)] -> Int -> Maybe Double -> Maybe Int -> IO () writeNode db eid en key children h fit mdsid = do runDb db "INSERT INTO enode (key, op, op_detail) VALUES (?, ?, ?)" @@ -209,11 +208,11 @@ [ SqlText (T.pack key) , SqlInteger (fromIntegral c) , SqlInteger (fromIntegral n) ] - forM_ children $ \(c, _, _) -> - runDb db "INSERT INTO parent (child_eid, parent_eid, parent_enode_key) VALUES (?, ?, ?)" - [ SqlInteger (fromIntegral c) - , SqlInteger (fromIntegral eid) - , SqlText (T.pack key) ] + -- Write the class page inline (O(1) per new class, no post-pass needed) + let parents = HashSet.fromList + [ (c, fromMaybe (error "importEqs: bad parent key in page write") (parseEnodeKey key)) + | (c, _, _) <- children ] + writeClassPage db eid (EClass eid (HashSet.singleton en) parents h (defaultInfo en h)) -- every class gets a dataset_fit row so the graph is fitness-annotated per -- dataset; the root's proper params/theta are written by the caller. case mdsid of @@ -267,41 +266,13 @@ , SqlInteger (fromIntegral eid) , SqlInteger (fromIntegral dsid) ] --- | Write every class page once, reconstructing each class from the relational --- tables (node + height from @eclass_node@/@eclass@, parents from @parent@, --- metrics from @fit@). Memory stays bounded to a single class at a time. -writeAllPages :: SqlBackend db => db -> IO () -writeAllPages db = do - cids <- queryDb db "SELECT eid FROM eclass ORDER BY eid" [] - forM_ cids $ \[eidCol] -> do - let eid = sqlToInt eidCol - nh <- queryDb db - "SELECT n.enode_key, c.height FROM eclass_node n \ - \JOIN eclass c ON c.eid = n.eid WHERE n.eid = ?" - [SqlInteger (fromIntegral eid)] - case nh of - ([SqlText k, hcol] : _) -> do - let en = fromMaybe (error ("importEqs: bad node key for eid " <> show eid)) - (parseEnodeKey (T.unpack k)) - h = sqlToInt hcol - pr <- queryDb db "SELECT parent_eid, parent_enode_key FROM parent WHERE child_eid = ?" - [SqlInteger (fromIntegral eid)] - let parents = HashSet.fromList - [ (sqlToInt pe, fromMaybe (error "importEqs: bad parent key") (parseEnodeKey (T.unpack (sqlToText pk)))) - | [pe, pk] <- pr ] - -- Pages are structural-only: cost/best are derived on load, and - -- fitness/dl/theta are dataset metadata (dataset_fit), so they are NOT - -- baked into the e-graph blob (keeps the graph reusable across datasets). - writeClassPage db eid (EClass eid (HashSet.singleton en) parents h (defaultInfo en h)) - _ -> pure () - -- | Serialize an e-class to the page store (INSERT OR REPLACE so the final -- pass is idempotent). writeClassPage :: SqlBackend db => db -> EClassId -> EClass -> IO () writeClassPage db eid ec = runDb db ("INSERT OR REPLACE INTO " <> classStoreTable <> " (key, blob) VALUES (?, ?)") [ SqlInteger (fromIntegral eid) - , SqlText (hex (BL.toStrict (encode ec))) ] + , SqlBlob (BL.toStrict (encode ec)) ] -- | Per-class data. Cost/best are derived quantities recomputed on load -- ('recalculateBestAll'). Fitness/dl/theta/size are baked into the page (so diff --git a/src/Algorithm/EqSat/Storage/Postgres.hs b/src/Algorithm/EqSat/Storage/Postgres.hs --- a/src/Algorithm/EqSat/Storage/Postgres.hs +++ b/src/Algorithm/EqSat/Storage/Postgres.hs @@ -1,4 +1,5 @@ {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE BangPatterns #-} -- | PostgreSQL-backed persistence for srtree e-graphs. -- @@ -12,13 +13,15 @@ -- 'closePostgres'); the reggression layer dispatches on a @postgres://@ / -- @postgresql://@ DSN. module Algorithm.EqSat.Storage.Postgres - ( schemaPostgres + ( schemaEgraphPostgres + , schemaFitPostgres , connectPostgres , closePostgres ) where import Control.Monad (forM, forM_) import Data.ByteString (ByteString) +import qualified Data.ByteString as BS import qualified Data.IntSet as IntSet import Data.Text (Text) import qualified Data.Text as T @@ -29,16 +32,15 @@ , ntuples, resultErrorMessage, resultStatus, toColumn, toRow ) import Algorithm.EqSat.Storage.Backend (SqlValue(..), SqlBackend(..), sqlToInt, sqlToText) -import Algorithm.EqSat.Storage.ClassStore (unhex) --- | PostgreSQL DDL. Mirrors 'Algorithm.EqSat.Storage.Schema.schemaSQL'. +-- | PostgreSQL DDL for the egraph section. -- -- Differences from SQLite: @BIGINT@ identity keys, @DOUBLE PRECISION@ -- metrics, and foreign keys declared @DEFERRABLE INITIALLY DEFERRED@ so the -- writer can insert @eclass_node@/@fit@ rows before their referenced -- @eclass@ rows within the @BEGIN@..@COMMIT@ transaction of 'saveGraph'. -schemaPostgres :: [Text] -schemaPostgres = +schemaEgraphPostgres :: [Text] +schemaEgraphPostgres = [ "CREATE TABLE IF NOT EXISTS meta (" <> " key TEXT PRIMARY KEY," <> " value TEXT NOT NULL)" @@ -62,24 +64,24 @@ <> " eid BIGINT NOT NULL REFERENCES eclass(eid) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED," <> " enode_key TEXT NOT NULL REFERENCES enode(key) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED," <> " PRIMARY KEY (eid, enode_key))" - , "CREATE TABLE IF NOT EXISTS parent (" - <> " child_eid BIGINT NOT NULL REFERENCES eclass(eid) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED," - <> " parent_eid BIGINT NOT NULL," - <> " parent_enode_key TEXT NOT NULL REFERENCES enode(key) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED," - <> " PRIMARY KEY (child_eid, parent_eid, parent_enode_key))" , "CREATE TABLE IF NOT EXISTS cstore_page (" - <> " key TEXT PRIMARY KEY," - <> " blob TEXT NOT NULL)" + <> " key BIGINT PRIMARY KEY," + <> " blob BYTEA NOT NULL)" , "CREATE TABLE IF NOT EXISTS frontier (" <> " eid BIGINT PRIMARY KEY REFERENCES eclass(eid) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED," <> " updated_at TEXT)" - , "CREATE TABLE IF NOT EXISTS dataset (" + ] + +-- | PostgreSQL DDL for a per-dataset fit DB (no FK to eclass). +schemaFitPostgres :: [Text] +schemaFitPostgres = + [ "CREATE TABLE IF NOT EXISTS dataset (" <> " id BIGSERIAL PRIMARY KEY," <> " name TEXT NOT NULL UNIQUE," <> " created TEXT)" , "CREATE TABLE IF NOT EXISTS dataset_fit (" <> " dataset_id BIGINT NOT NULL REFERENCES dataset(id) ON DELETE CASCADE," - <> " eid BIGINT NOT NULL REFERENCES eclass(eid) ON DELETE CASCADE," + <> " eid BIGINT NOT NULL," <> " fitness DOUBLE PRECISION," <> " dl DOUBLE PRECISION," <> " theta TEXT," @@ -89,12 +91,9 @@ <> " stale INTEGER NOT NULL DEFAULT 0," <> " updated_at TEXT," <> " PRIMARY KEY (dataset_id, eid))" - , "CREATE INDEX IF NOT EXISTS idx_dsfit_fitness ON dataset_fit(fitness)" - , "CREATE INDEX IF NOT EXISTS idx_dsfit_size ON dataset_fit(size)" - , "CREATE INDEX IF NOT EXISTS idx_dsfit_dl ON dataset_fit(dl)" , "CREATE TABLE IF NOT EXISTS expression_index (" <> " expression_key TEXT PRIMARY KEY," - <> " eclass BIGINT NOT NULL REFERENCES eclass(eid) ON DELETE CASCADE," + <> " eclass BIGINT NOT NULL," <> " dataset_id BIGINT REFERENCES dataset(id) ON DELETE CASCADE," <> " first_seen TEXT)" ] @@ -140,8 +139,33 @@ statusOK r "query" pure [] - createSchemaDb conn = mapM_ (execDb conn) schemaPostgres + foldQueryDb conn sql params seed0 f = do + r <- pgExecParams conn sql params + st <- resultStatus r + case st of + TuplesOk -> do + ns <- ntuples r + nf <- nfields r + let n = fromEnum ns + m = fromEnum nf + let go !acc i + | i >= n = pure acc + | otherwise = do + row <- forM [0 .. m - 1] $ \j -> do + v <- getvalue r (toRow i) (toColumn j) + pure $ case v of + Nothing -> SqlNull + Just bs -> SqlText (TE.decodeUtf8 bs) + acc' <- f acc row + go acc' (i + 1) + go seed0 0 + _ -> do + statusOK r "foldQuery" + pure seed0 + createSchemaDb conn = mapM_ (execDb conn) schemaEgraphPostgres + createSchemaDbFit conn = mapM_ (execDb conn) schemaFitPostgres + -- Grid fallback (Postgres is not the out-of-core target): the cursor-based -- streaming matcher needs 'Database.SQLite3'; here we return the full -- distinct set up to @budget@, documented as unbounded memory. @@ -156,7 +180,11 @@ -- out-of-core target). streamPages conn tbl k = do rows <- queryDb conn ("SELECT key, blob FROM " <> tbl) [] - forM_ rows $ \[key, blob] -> k (fromIntegral (sqlToInt key)) (unhex (sqlToText blob)) + forM_ rows $ \[key, blob] -> + -- Postgres bytea returns hex-encoded text with \x prefix in text protocol + let raw = sqlToText blob + hexStr = if T.isPrefixOf "\\x" raw then T.drop 2 raw else raw + in k (fromIntegral (sqlToInt key)) (unhex hexStr) -- | Raise an exception unless the status is @CommandOk@/@TuplesOk@. statusOK :: Result -> Text -> IO () @@ -195,6 +223,7 @@ renderParam (SqlInteger n) = Just (invalidOid, TE.encodeUtf8 (T.pack (show n)), Text) renderParam (SqlFloat d) = Just (invalidOid, TE.encodeUtf8 (T.pack (show d)), Text) renderParam (SqlText t) = Just (invalidOid, TE.encodeUtf8 t, Text) +renderParam (SqlBlob bs) = Just (invalidOid, TE.encodeUtf8 (T.pack (show bs)), Text) renderParam SqlNull = Nothing -- | Rewrite the shared positional @?@ placeholders to libpq's @$n@ form @@ -211,3 +240,13 @@ skip [] = [] go n ('?' : r) = '$' : show n ++ go (n + 1) r go n (c : r) = c : go n r + +-- | Decode a hex string to a ByteString (local copy; Postgres bytea returns +-- hex-encoded text in the text protocol). +unhex :: Text -> ByteString +unhex = BS.pack . go . T.unpack + where + go (a:b:r) = fromIntegral (hexv a * 16 + hexv b) : go r + go _ = [] + hexv c | c >= '0' && c <= '9' = fromEnum c - fromEnum '0' + | otherwise = fromEnum c - fromEnum 'a' + 10 diff --git a/src/Algorithm/EqSat/Storage/Query.hs b/src/Algorithm/EqSat/Storage/Query.hs --- a/src/Algorithm/EqSat/Storage/Query.hs +++ b/src/Algorithm/EqSat/Storage/Query.hs @@ -67,22 +67,29 @@ :: SqlBackend db => db -> Int -> EClassId -> Maybe Double -> Maybe Double -> Text -> Int -> IO () writeDatasetFit db ds eid fit dl theta sz = do - let (fitCol, fitVal) = case fit of - Nothing -> ("NULL", Nothing) - Just f -> ("?", Just (SqlFloat f)) - (dlCol, dlVal) = case dl of - Nothing -> ("NULL", Nothing) - Just d -> ("?", Just (SqlFloat d)) + let (fitVal, fitExcluded) = case fit of + Nothing -> ("NULL", "excluded.fitness") + Just f -> ("?", "excluded.fitness") + (dlVal, dlExcluded) = case dl of + Nothing -> ("NULL", "excluded.dl") + Just d -> ("?", "excluded.dl") + isFitted = case fit of { Nothing -> 0; Just _ -> 1 } + isEvaluated = case fit of { Nothing -> 0; Just _ -> 1 } + fitParams = case fit of { Nothing -> []; Just f -> [SqlFloat f] } + dlParams = case dl of { Nothing -> []; Just d -> [SqlFloat d] } runDb db - ("INSERT OR REPLACE INTO dataset_fit \ + ("INSERT INTO dataset_fit \ \(dataset_id, eid, fitness, dl, theta, size, evaluated, fitted) \ - \VALUES (?, ?, " <> fitCol <> ", " <> dlCol <> ", ?, ?, 1, 1)") - (catMaybes [ Just (SqlInteger (fromIntegral ds)) - , Just (SqlInteger (fromIntegral eid)) - , fitVal - , dlVal - , Just (SqlText theta) - , Just (SqlInteger (fromIntegral sz)) ]) + \VALUES (?, ?, " <> fitVal <> ", " <> dlVal <> ", ?, ?, " <> T.pack (show isEvaluated) <> ", " <> T.pack (show isFitted) <> ") \ + \ON CONFLICT (dataset_id, eid) DO UPDATE SET \ + \fitness = " <> fitExcluded <> ", dl = " <> dlExcluded <> ", theta = excluded.theta, size = excluded.size, \ + \evaluated = " <> T.pack (show isEvaluated) <> ", fitted = " <> T.pack (show isFitted)) + ([ SqlInteger (fromIntegral ds) + , SqlInteger (fromIntegral eid) + ] ++ fitParams ++ dlParams ++ + [ SqlText theta + , SqlInteger (fromIntegral sz) + ]) -- | Read per-(dataset, e-class) fit rows. readDatasetFit diff --git a/src/Algorithm/EqSat/Storage/SQLite.hs b/src/Algorithm/EqSat/Storage/SQLite.hs --- a/src/Algorithm/EqSat/Storage/SQLite.hs +++ b/src/Algorithm/EqSat/Storage/SQLite.hs @@ -1,5 +1,6 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE BangPatterns #-} -- | SQLite-backed persistence for srtree e-graphs. -- @@ -25,9 +26,10 @@ , refreshFitness , query , flushStore + , loadPagesBulk ) where -import Control.Monad (forM, forM_, when) +import Control.Monad (forM, forM_, when, foldM) import Control.Exception (SomeException, catch, displayException) import Control.Monad.Identity (runIdentity) import Control.Monad.State.Strict (execStateT) @@ -35,7 +37,7 @@ import Data.Maybe (catMaybes, fromMaybe, listToMaybe) import Data.Text (Text) import qualified Data.Text as T -import Data.List (foldl') +import Data.List (foldl', intercalate) import qualified Data.IntSet as IntSet import qualified Data.IntMap as IntMap import qualified Data.HashMap.Strict as HashMap @@ -44,6 +46,8 @@ import qualified Data.Set as RangeSet import Data.Binary (decode, encode) import qualified Data.ByteString.Lazy as BL +import qualified Data.ByteString as BS +import qualified Data.Text.Encoding as TE import Database.SQLite3 ( Database, SQLData(..), StepResult(..) @@ -59,17 +63,21 @@ import Algorithm.EqSat.Store ( GraphRows(..), EClassRow(..), exportEGraph, importEGraph, rebuildDBs ) import Algorithm.EqSat.Storage.Backend - ( SqlValue(..), SqlBackend(..), sqlToInt, sqlToMaybeDouble, sqlToText ) + ( SqlValue(..), SqlBackend(..), sqlToInt, sqlToMaybeDouble, sqlToText, sqlToBlob ) import Algorithm.EqSat.Storage.ClassStore - ( classStoreTable, openClassStore, allPages, classStoreHandle, hex, unhex ) + ( classStoreTable, openClassStore, allPages, classStoreHandle ) import Algorithm.EqSat.Storage.Query (readDatasetFit, writeDatasetFit, firstDatasetId) import Algorithm.EqSat.Storage.Stream (streamRootsByOp) import Algorithm.EqSat.Storage.Types -import Algorithm.EqSat.Storage.Schema (createSchema, schemaSQL) +import Algorithm.EqSat.Storage.Schema (egraphSchemaSQL, fitSchemaSQL, createSchema, createSchemaFit) -- | Default cache capacity (pages) for the lazily paged e-class store. +-- Kept small: the cache is an LRU of deserialized EClass objects (each with +-- HashSets of nodes/parents), so a large cap wastes resident memory. The +-- out-of-core matcher streams roots directly from the DB and the extraction +-- path reads one class at a time, so a small working set suffices. defaultClassCap :: Int -defaultClassCap = 50000 +defaultClassCap = 1000 -- --------------------------------------------------------------------------- -- SQLite driver instance @@ -95,6 +103,18 @@ Row -> do cols <- columns stmt go stmt (map fromSqlData cols : acc) + foldQueryDb db sql params seed0 go = withStatement db sql $ \stmt -> do + bind stmt (map toSqlData params) + goRows seed0 stmt + where + goRows !acc stmt = do + r <- step stmt + case r of + Done -> pure acc + Row -> do + cols <- columns stmt + acc' <- go acc (map fromSqlData cols) + goRows acc' stmt -- O(1)-memory candidate-root enumeration: stream the distinct e-class ids by -- operator through a cursor, skipping the already-attempted set, and stop -- after @budget@ rows, so the matcher never materializes the whole (operator @@ -112,23 +132,27 @@ Row -> do cols <- columns stmt let eid = case cols of (SQLInteger i : _) -> i; _ -> 0 - hv = case cols of (_ : SQLText t : _) -> t; _ -> "" - k eid (unhex hv) + blob = case cols of (_ : SQLBlob bs : _) -> bs + (_ : SQLText t : _) -> TE.encodeUtf8 t + _ -> BS.empty + k eid blob go stmt - createSchemaDb db = mapM_ (exec db) schemaSQL + createSchemaDb db = mapM_ (exec db) egraphSchemaSQL + createSchemaDbFit db = mapM_ (exec db) fitSchemaSQL toSqlData :: SqlValue -> SQLData toSqlData (SqlInteger n) = SQLInteger n toSqlData (SqlFloat d) = SQLFloat d toSqlData (SqlText t) = SQLText t +toSqlData (SqlBlob bs) = SQLBlob bs toSqlData SqlNull = SQLNull fromSqlData :: SQLData -> SqlValue fromSqlData (SQLInteger n) = SqlInteger n fromSqlData (SQLFloat d) = SqlFloat d fromSqlData (SQLText t) = SqlText t +fromSqlData (SQLBlob bs) = SqlBlob bs fromSqlData SQLNull = SqlNull -fromSqlData _ = SqlNull -- | Driver-neutral parameterized query (abstracts the concrete backend). query :: SqlBackend db => db -> Text -> [SqlValue] -> IO [[SqlValue]] @@ -170,7 +194,6 @@ writeMeta db rows writeNodes db rows writeClasses db rows - writeParents db rows writeDatasetFitRows db dsid rows writeClassPages db rows where @@ -204,7 +227,6 @@ clearTables :: SqlBackend db => db -> IO () clearTables db = do - execDb db "DELETE FROM parent" execDb db "DELETE FROM meta" execDb db "DELETE FROM enode_child" execDb db "DELETE FROM eclass_node" @@ -219,7 +241,7 @@ forM_ (IntMap.toAscList (_grEClasses rows)) $ \(eid, r) -> run db ("INSERT INTO " <> classStoreTable <> " (key, blob) VALUES (?, ?)") [ SqlInteger (fromIntegral eid) - , SqlText (hex (BL.toStrict (encode (EClass eid (_rcNodes r) (_rcParents r) (_rcHeight r) (_rcInfo r))))) ] + , SqlBlob (BL.toStrict (encode (EClass eid (_rcNodes r) (_rcParents r) (_rcHeight r) (_rcInfo r)))) ] writeMeta :: SqlBackend db => db -> GraphRows -> IO () writeMeta db rows = do @@ -257,18 +279,6 @@ , SqlInteger (fromIntegral canon) , SqlInteger (fromIntegral (maybe 0 _rcHeight (IntMap.lookup eid (_grEClasses rows)))) ] --- | Persist the reverse edges: for every (parent class, parent e-node) in each --- class's @_parents@, a @parent@ row keyed by the child e-class. This makes the --- parent relation queryable per class without scanning @enode@/@eclass_node@. -writeParents :: SqlBackend db => db -> GraphRows -> IO () -writeParents db rows = - forM_ (IntMap.toAscList (_grEClasses rows)) $ \(eid, r) -> - forM_ (Set.toList (_rcParents r)) $ \(pEid, pEn) -> - run db "INSERT INTO parent (child_eid, parent_eid, parent_enode_key) VALUES (?, ?, ?)" - [ SqlInteger (fromIntegral eid) - , SqlInteger (fromIntegral pEid) - , SqlText (T.pack (enodeKey pEn)) ] - -- | Write the per-(dataset, e-class) fitness rows for every class in a graph. writeDatasetFitRows :: SqlBackend db => db -> Int -> GraphRows -> IO () writeDatasetFitRows db dsid rows = @@ -312,11 +322,9 @@ if null pages then do -- fully relational path (databases written before the page store) - parents <- readParents db - let storedParents = IntMap.fromListWith Set.union - [ (c, Set.singleton (pEid, pEn)) - | (c, pEid, pEn) <- parents ] - classes = buildClasses canon nodeToEClass storedParents (IntMap.fromList fit) (IntMap.fromList [ (eid, h) | (eid, _, h) <- ecLst ]) + -- Parent pointers are recomputed from nodeToEClass (the parent table + -- has been removed; buildClasses handles the fallback). + let classes = buildClasses canon nodeToEClass IntMap.empty (IntMap.fromList fit) (IntMap.fromList [ (eid, h) | (eid, _, h) <- ecLst ]) rows = GraphRows canon nodeToEClass classes nextId trackDBs pure (importEGraph rows) else do @@ -438,11 +446,7 @@ rep eid = IntMap.findWithDefault eid eid canon0 nodeToEClass0 = HashMap.fromList enodes nodeToEClass = HashMap.map rep nodeToEClass0 - parents <- readParents db - let storedParents = IntMap.fromListWith Set.union - [ (c, Set.singleton (pEid, pEn)) - | (c, pEid, pEn) <- parents ] - classes = buildClasses canon0 nodeToEClass storedParents fitMap + classes = buildClasses canon0 nodeToEClass IntMap.empty fitMap (IntMap.fromList [ (eid, h) | (eid, _, h) <- ecLst ]) rows = GraphRows canon0 nodeToEClass classes nextId trackDBs pure (importEGraph rows) @@ -504,21 +508,9 @@ rows <- query db "SELECT eid, canonical, height FROM eclass" [] pure [ (sqlToInt eid, sqlToInt c, sqlToInt h) | [eid, c, h] <- rows ] --- | Read (child e-class, parent e-class, parent e-node) edges from the @parent@ --- table. The rows are grouped per child class by 'loadGraph'. -readParents :: SqlBackend db => db -> IO [(EClassId, EClassId, ENode)] -readParents db = do - rows <- query db "SELECT child_eid, parent_eid, parent_enode_key FROM parent" [] - pure (catMaybes - [ do - en <- parseEnodeKey (T.unpack (sqlToText k)) - pure (sqlToInt c, sqlToInt p, en) - | [c, p, k] <- rows ]) - -- | Rebuild @_grEClasses@ rows: only canonical roots carry real class rows. --- Parent pointers come from the stored @parent@ relation when present, falling --- back to recomputation from the node -> class map (e.g. databases written --- before the @parent@ table existed, or hand-built rows). +-- Parent pointers are recomputed from the node -> class map (the parent table +-- has been removed; the cstore_page blobs store the authoritative parent set). buildClasses :: IntMap.IntMap EClassId -- ^ canonical eid -> eid (self-map for roots) -> HashMap.HashMap ENode EClassId -- ^ node -> class @@ -590,3 +582,28 @@ c <- canonical eid insertFitness c f (parseTheta (T.unpack theta)) pure (Right (runIdentity $ execStateT m eg)) + +-- | Bulk-load page blobs for a list of e-class IDs in one query, returning +-- an IntMap of deserialized EClass values. For lists exceeding SQLite's +-- parameter limit, the query is chunked automatically. +loadPagesBulk :: SqlBackend db => db -> [EClassId] -> IO (IntMap.IntMap EClass) +loadPagesBulk _ [] = pure IntMap.empty +loadPagesBulk db eids = do + let chunks = chunkList 500 eids + foldM (\acc chunkIds -> do + pages <- loadChunk chunkIds + pure $! IntMap.union acc pages) IntMap.empty chunks + where + chunkList _ [] = [] + chunkList n xs = let (h, t) = splitAt n xs in h : chunkList n t + + loadChunk ids = do + let placeholders = intercalate "," (replicate (length ids) "?") + params = map (SqlInteger . fromIntegral) ids + rows <- queryDb db + ("SELECT key, blob FROM cstore_page WHERE key IN (" <> T.pack placeholders <> ")") + params + pure $ IntMap.fromList + [ (sqlToInt k, decode (BL.fromStrict (sqlToBlob b))) + | [k, b] <- rows + ] diff --git a/src/Algorithm/EqSat/Storage/Schema.hs b/src/Algorithm/EqSat/Storage/Schema.hs --- a/src/Algorithm/EqSat/Storage/Schema.hs +++ b/src/Algorithm/EqSat/Storage/Schema.hs @@ -2,29 +2,30 @@ -- | Schema for persisting srtree e-graphs. -- --- Layout (shared by the SQLite and PostgreSQL backends): --- * @meta@ - scalar settings (@next_id@, @track_dbs@, cost-function tag) --- * @enode@ - content-addressable e-nodes (@key@ = canonical serialization) --- * @enode_child@ - ENAry multiset children (@child_eid@, @cnt@) --- * @eclass@ - e-class id -> canonical representative + height --- * @eclass_node@ - canonical e-node -> e-class membership --- * @parent@ - reverse edges: child e-class -> (parent e-class, parent e-node) --- * @fit@ - per-class risk metrics (fitness, dl, size, theta) +-- Two logical sections, potentially in separate DB files: +-- * E-graph section (dataset-agnostic, read-only during fitting): +-- @meta@, @enode@, @enode_child@, @eclass@, @eclass_node@, +-- @cstore_page@, @frontier@ +-- * Dataset-fit section (per-dataset, write-heavy during fitting): +-- @dataset@, @dataset_fit@, @expression_index@ -- --- 'schemaSQL' is the SQLite DDL; 'Algorithm.EqSat.Storage.Postgres' carries --- the equivalent PostgreSQL DDL (identity keys, deferred FK checks, --- @DOUBLE PRECISION@). Dataset-specific fit tables are a later phase. +-- 'egraphSchemaSQL' is the DDL for the egraph DB. +-- 'fitSchemaSQL' is the DDL for a per-dataset fit DB (no FK to eclass). module Algorithm.EqSat.Storage.Schema - ( schemaSQL + ( egraphSchemaSQL + , fitSchemaSQL , createSchema + , createSchemaFit ) where import Data.Text (Text) import Algorithm.EqSat.Storage.Backend (SqlBackend(..)) -schemaSQL :: [Text] -schemaSQL = +-- | DDL for the egraph database. +-- Full schema including dataset tables for backward compatibility with importEqs. +egraphSchemaSQL :: [Text] +egraphSchemaSQL = [ "CREATE TABLE IF NOT EXISTS meta (" <> " key TEXT PRIMARY KEY," <> " value TEXT NOT NULL)" @@ -48,14 +49,9 @@ <> " eid INTEGER NOT NULL REFERENCES eclass(eid) ON DELETE CASCADE," <> " enode_key TEXT NOT NULL REFERENCES enode(key) ON DELETE CASCADE," <> " PRIMARY KEY (eid, enode_key))" - , "CREATE TABLE IF NOT EXISTS parent (" - <> " child_eid INTEGER NOT NULL REFERENCES eclass(eid) ON DELETE CASCADE," - <> " parent_eid INTEGER NOT NULL," - <> " parent_enode_key TEXT NOT NULL REFERENCES enode(key) ON DELETE CASCADE," - <> " PRIMARY KEY (child_eid, parent_eid, parent_enode_key))" , "CREATE TABLE IF NOT EXISTS cstore_page (" - <> " key TEXT PRIMARY KEY," - <> " blob TEXT NOT NULL)" + <> " key INTEGER PRIMARY KEY," + <> " blob BLOB NOT NULL)" , "CREATE TABLE IF NOT EXISTS frontier (" <> " eid INTEGER PRIMARY KEY REFERENCES eclass(eid) ON DELETE CASCADE," <> " updated_at TEXT)" @@ -75,9 +71,6 @@ <> " stale INTEGER NOT NULL DEFAULT 0," <> " updated_at TEXT," <> " PRIMARY KEY (dataset_id, eid))" - , "CREATE INDEX IF NOT EXISTS idx_dsfit_fitness ON dataset_fit(fitness)" - , "CREATE INDEX IF NOT EXISTS idx_dsfit_size ON dataset_fit(size)" - , "CREATE INDEX IF NOT EXISTS idx_dsfit_dl ON dataset_fit(dl)" , "CREATE TABLE IF NOT EXISTS expression_index (" <> " expression_key TEXT PRIMARY KEY," <> " eclass INTEGER NOT NULL REFERENCES eclass(eid) ON DELETE CASCADE," @@ -85,6 +78,37 @@ <> " first_seen TEXT)" ] --- | Create (or ensure) the schema on the given backend. +-- | DDL for a per-dataset fit database. +-- No FK to eclass (e-graph lives in a separate DB). +fitSchemaSQL :: [Text] +fitSchemaSQL = + [ "CREATE TABLE IF NOT EXISTS dataset (" + <> " id INTEGER PRIMARY KEY," + <> " name TEXT NOT NULL UNIQUE," + <> " created TEXT)" + , "CREATE TABLE IF NOT EXISTS dataset_fit (" + <> " dataset_id INTEGER NOT NULL REFERENCES dataset(id) ON DELETE CASCADE," + <> " eid INTEGER NOT NULL," + <> " fitness REAL," + <> " dl REAL," + <> " theta TEXT," + <> " size INTEGER NOT NULL DEFAULT 0," + <> " evaluated INTEGER NOT NULL DEFAULT 0," + <> " fitted INTEGER NOT NULL DEFAULT 0," + <> " stale INTEGER NOT NULL DEFAULT 0," + <> " updated_at TEXT," + <> " PRIMARY KEY (dataset_id, eid))" + , "CREATE TABLE IF NOT EXISTS expression_index (" + <> " expression_key TEXT PRIMARY KEY," + <> " eclass INTEGER NOT NULL," + <> " dataset_id INTEGER REFERENCES dataset(id) ON DELETE CASCADE," + <> " first_seen TEXT)" + ] + +-- | Create (or ensure) the egraph schema on the given backend. createSchema :: SqlBackend db => db -> IO () createSchema = createSchemaDb + +-- | Create (or ensure) the fit schema on the given backend. +createSchemaFit :: SqlBackend db => db -> IO () +createSchemaFit = createSchemaDbFit diff --git a/srtree-db.cabal b/srtree-db.cabal --- a/srtree-db.cabal +++ b/srtree-db.cabal @@ -1,12 +1,11 @@ cabal-version: 2.4 name: srtree-db -version: 0.1.1.0 +version: 0.1.2.0 synopsis: SQL persistence and querying for srtree e-graphs description: Reusable storage layer for srtree multiset e-graphs: a driver-neutral serialization of the Algorithm.EqSat.Store rows (backed by SQLite and PostgreSQL) plus SQL queries - (topN, pareto, distribution counts, patterns). - Dataset-specific fit tables are a later phase. + (topN, pareto, distribution counts, patterns, dataset-aware fit). license: BSD-3-Clause license-file: LICENSE author: Fabricio Olivetti de França @@ -28,6 +27,7 @@ Algorithm.EqSat.Storage.Types Algorithm.EqSat.Storage.Backend Algorithm.EqSat.Storage.ClassStore + Algorithm.EqSat.Storage.Extract Algorithm.EqSat.Storage.Import Algorithm.EqSat.Storage.Stream Algorithm.EqSat.Storage.Schema @@ -43,7 +43,7 @@ , text >=1.2 && <2.2 , direct-sqlite >=2.3 && <2.4 , postgresql-libpq >=0.10 && <0.12 - , srtree >=3.0.0.2 && <3.1 + , srtree >=3.0.0.3 && <3.1 , vector >=0.12 && <0.14 , mtl >=2.2 && <2.4 default-language: Haskell2010 @@ -62,22 +62,24 @@ , directory >=1.3 && <1.4 , direct-sqlite >=2.3 && <2.4 , postgresql-libpq >=0.10 && <0.12 - , srtree >=3.0 && <3.1 + , srtree >=3.0.0.3 && <3.1 , srtree-db >=0.1 && <0.2 default-language: Haskell2010 executable srtree-db hs-source-dirs: app main-is: Main.hs - other-modules: Ingest, EqSat, FitData + other-modules: Ingest, EqSat, FitData, Status ghc-options: -O2 -threaded -rtsopts -with-rtsopts=-N build-depends: base >=4.14 && <5 , optparse-applicative >=0.17 && <0.19 , srtree-db >=0.1 && <0.2 - , srtree >=3.0 && <3.1 + , srtree >=3.0.0.3 && <3.1 , bytestring >=0.10 && <0.13 + , binary >=0.8 && <0.9 , containers >=0.6 && <0.9 + , unordered-containers >=0.2 && <0.3 , text >=1.2 && <2.2 , vector >=0.12 && <0.14 , directory >=1.3 && <1.4 @@ -85,6 +87,7 @@ , direct-sqlite >=2.3 && <2.4 , random >=1.2 && <1.3 , mtl >=2.2 && <2.4 + , async >=2.2 && <2.3 default-language: Haskell2010 test-suite srtree-db-test @@ -102,7 +105,7 @@ , text >=1.2 && <2.2 , direct-sqlite >=2.3 && <2.4 , postgresql-libpq >=0.10 && <0.12 - , srtree >=3.0 && <3.1 + , srtree >=3.0.0.3 && <3.1 , srtree-db >=0.1 && <0.2 , vector >=0.12 && <0.14 , mtl >=2.2 && <2.4 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -44,6 +44,7 @@ myCost (Var _) = 1 myCost (Const _) = 1 myCost (Param _) = 1 +myCost (Y _) = 1 myCost (Bin _ l r) = 2 + l + r myCost (Uni _ t) = 3 + t @@ -172,28 +173,6 @@ mkPage :: Int -> BS.ByteString mkPage i = BS.replicate (10 + i) (fromIntegral i) --- | The @parent@ table round-trips every reverse edge: the row count matches --- the live graph's total @_parents@ entries, and a reload reconstructs the --- identical per-class parent sets. -testParents :: SqlBackend db => IO db -> (db -> IO ()) -> Test -testParents openDb closeDb = TestCase $ do - db <- openDb - (eg, _, _, _) <- buildGraph - _ <- saveGraphTest db eg - let expected = sum - [ Set.size (_parents ec) - | (_, ec) <- IntMap.toList (_eClass eg) ] - rows <- query db "SELECT COUNT(*) FROM parent" [] - case rows of - [[SqlInteger n]] -> assertEqual "parent row count" expected (fromIntegral n) - [[SqlText t]] -> assertEqual "parent row count" expected - (read (T.unpack t) :: Int) - _ -> assertFailure "parent count: unexpected row shape" - Right eg' <- loadGraph db - let parentsMap g = IntMap.map _parents (_eClass g) - assertEqual "parents preserved per class" (parentsMap eg) (parentsMap eg') - closeDb db - testStoreRoundtrip :: SqlBackend db => IO db -> (db -> IO ()) -> Test testStoreRoundtrip openDb closeDb = TestCase $ do db <- openDb @@ -566,7 +545,6 @@ runSuite :: SqlBackend db => String -> (IO db, db -> IO ()) -> [Test] runSuite tag (openDb, closeDb) = [ TestLabel (tag <> " save-load-roundtrip") (testSaveLoadRT openDb closeDb) - , TestLabel (tag <> " parents") (testParents openDb closeDb) , TestLabel (tag <> " queries") (testQueries openDb closeDb) , TestLabel (tag <> " sync") (testSync openDb closeDb) ]