Make merging trees structurally non-empty

lsm-tree/src-core/Database/LSMTree/Internal/MergingTree.hs

@@ -13,12 +13,13 @@ module Database.LSMTree.Internal.MergingTree (
   , newPendingLevelMerge
   , newPendingUnionMerge
   , getCompleted
-  , isStructurallyEmpty
   , remainingMergeDebt
   , supplyCredits
     -- * Internal state
   , MergingTreeState (..)
   , PendingMerge (..)
+  , pattern PendingLevelMerge
+  , pattern PendingUnionMerge
   ) where
 
 import           Control.ActionRegistry
@@ -113,18 +114,56 @@ data MergingTreeState m h =
       !(PendingMerge m h)
 
 -- | A merge that is waiting for its inputs to complete.
+--
+-- The constructors are unsafe and only exported to allow deriving instances in
+-- other modules. Use the patterns 'PendingLevelMerge' and 'PendingUnionMerge'
+-- instead.
 data PendingMerge m h =
     -- | The collection of inputs is the entire contents of a table,
     -- i.e. its (merging) runs and finally a union merge (if that table
     -- already contained a union).
-    PendingLevelMerge
+    --
+    -- INVARIANT: At least one 'PreExistingRun' must be present. This is
+    -- enforced by in 'mkPendingLevelMerge'. Don't use the constructor directly.
+    UnsafePendingLevelMerge
       !(Vector (PreExistingRun m h))
       !(Maybe (Ref (MergingTree m h)))
 
     -- | Each input is the entire content of a table (as a merging tree).
-  | PendingUnionMerge
+    --
+    -- INVARIANT: At least two inputs ('MergingTree') must be present. This is
+    -- enforced in 'newPendingUnionMerge'. Don't use the constructor directly.
+  | UnsafePendingUnionMerge
       !(Vector (Ref (MergingTree m h)))
 
+{-# COMPLETE PendingLevelMerge, PendingUnionMerge #-}
+pattern PendingLevelMerge ::
+     Vector (PreExistingRun m h)
+  -> Maybe (Ref (MergingTree m h))
+  -> PendingMerge m h
+pattern PendingLevelMerge prs mmt <- UnsafePendingLevelMerge prs mmt
+
+pattern PendingUnionMerge ::
+     Vector (Ref (MergingTree m h))
+  -> PendingMerge m h
+pattern PendingUnionMerge mts <- UnsafePendingUnionMerge mts
+
+mkPendingLevelMerge ::
+     PreExistingRun m h
+  -> [PreExistingRun m h]
+  -> Maybe (Ref (MergingTree m h))
+  -> PendingMerge m h
+mkPendingLevelMerge pr prs mmt =
+    UnsafePendingLevelMerge (V.fromList (pr : prs)) mmt
+
+mkPendingUnionMerge ::
+     Ref (MergingTree m h)
+  -> Ref (MergingTree m h)
+  -> [Ref (MergingTree m h)]
+  -> PendingMerge m h
+mkPendingUnionMerge mt1 mt2 mts =
+    UnsafePendingUnionMerge (V.fromList (mt1 : mt2 : mts))
+
 pendingContent ::
      PendingMerge m h
   -> ( MR.TreeMergeType
@@ -176,15 +215,14 @@ newOngoingMerge refCtx mr = mkMergingTree refCtx . OngoingTreeMerge =<< dupRef m
      RefCtx
   -> [PreExistingRun IO h]
   -> Maybe (Ref (MergingTree IO h))
-  -> IO (Ref (MergingTree IO h)) #-}
+  -> IO (Maybe (Ref (MergingTree IO h))) #-}
 -- | Create a new 'MergingTree' representing the merge of a sequence of
 -- pre-existing runs (completed or ongoing, plus a optional final tree).
 -- This is for merging the entire contents of a table down to a single run
 -- (while sharing existing ongoing merges).
 --
--- Shape: if the list of runs is empty and the optional input tree is
--- structurally empty, the result will also be structurally empty. See
--- 'isStructurallyEmpty'.
+-- If the list of runs is empty and the optional input tree is not present, the
+-- resulting tree would be empty, so we return @Nothing@.
 --
 -- Resource tracking:
 -- * This allocates a new 'Ref' which the caller is responsible for releasing
@@ -200,8 +238,8 @@ newPendingLevelMerge ::
   => RefCtx
   -> [PreExistingRun m h]
   -> Maybe (Ref (MergingTree m h))
-  -> m (Ref (MergingTree m h))
-newPendingLevelMerge _ [] (Just t) = dupRef t
+  -> m (Maybe (Ref (MergingTree m h)))
+newPendingLevelMerge _ [] mt = traverse dupRef mt
 newPendingLevelMerge refCtx [PreExistingRun r] Nothing = do
     -- No need to create a pending merge here.
     --
@@ -214,40 +252,30 @@ newPendingLevelMerge refCtx [PreExistingRun r] Nothing = do
     r' <- dupRef r
     -- There are no interruption points here, and thus provided async
     -- exceptions are masked then there can be no async exceptions here at all.
-    mkMergingTree refCtx (CompletedTreeMerge r')
-
-newPendingLevelMerge refCtx prs mmt = do
-    -- isStructurallyEmpty is an interruption point, and can receive async
-    -- exceptions even when masked. So we use it first, *before* allocating
-    -- new references.
-    mmt' <- dupMaybeMergingTree mmt
-    prs' <- traverse dupPreExistingRun (V.fromList prs)
-    mkMergingTree refCtx (PendingTreeMerge (PendingLevelMerge prs' mmt'))
+    Just <$> mkMergingTree refCtx (CompletedTreeMerge r')
+
+newPendingLevelMerge refCtx (pr : prs) mmt = do
+    pr'  <- dupPreExistingRun pr
+    prs' <- traverse dupPreExistingRun prs
+    mmt' <- traverse dupRef mmt
+    let pending = mkPendingLevelMerge pr' prs' mmt'
+    Just <$> mkMergingTree refCtx (PendingTreeMerge pending)
   where
     dupPreExistingRun (PreExistingRun r) =
       PreExistingRun <$!> dupRef r
     dupPreExistingRun (PreExistingMergingRun mr) =
       PreExistingMergingRun <$!> dupRef mr
 
-    dupMaybeMergingTree :: Maybe (Ref (MergingTree m h))
-                        -> m (Maybe (Ref (MergingTree m h)))
-    dupMaybeMergingTree Nothing   = pure Nothing
-    dupMaybeMergingTree (Just mt) = do
-      isempty <- isStructurallyEmpty mt
-      if isempty
-        then pure Nothing
-        else Just <$!> dupRef mt
-
 {-# SPECIALISE newPendingUnionMerge ::
      RefCtx
   -> [Ref (MergingTree IO h)]
-  -> IO (Ref (MergingTree IO h)) #-}
+  -> IO (Maybe (Ref (MergingTree IO h))) #-}
 -- | Create a new 'MergingTree' representing the union of one or more merging
 -- trees. This is for unioning the content of multiple tables (represented
 -- themselves as merging trees).
 --
--- Shape: if all of the input trees are structurally empty, the result will
--- also be structurally empty. See 'isStructurallyEmpty'.
+-- If the list of inputs is empty, the resulting tree would be empty, so we
+-- return @Nothing@.
 --
 -- Resource tracking:
 -- * This allocates a new 'Ref' which the caller is responsible for releasing
@@ -261,16 +289,15 @@ newPendingUnionMerge ::
      (MonadMVar m, MonadMask m, PrimMonad m)
   => RefCtx
   -> [Ref (MergingTree m h)]
-  -> m (Ref (MergingTree m h))
-newPendingUnionMerge refCtx mts = do
-    mts' <- V.filterM (fmap not . isStructurallyEmpty) (V.fromList mts)
-    -- isStructurallyEmpty is interruptible even with async exceptions masked,
-    -- but we use it before allocating new references.
-    mts'' <- V.mapM dupRef mts'
-    case V.uncons mts'' of
-      Just (mt, x) | V.null x
-        -> pure mt
-      _ -> mkMergingTree refCtx (PendingTreeMerge (PendingUnionMerge mts''))
+  -> m (Maybe (Ref (MergingTree m h)))
+newPendingUnionMerge _ [] =
+    pure Nothing
+newPendingUnionMerge _ [mt] =
+    -- No need to create a new node, directly use the single input tree.
+    Just <$> dupRef mt
+newPendingUnionMerge refCtx (mt1 : mt2 : mts) = do
+    state <- mkPendingUnionMerge <$> dupRef mt1 <*> dupRef mt2 <*> mapM dupRef mts
+    Just <$> mkMergingTree refCtx (PendingTreeMerge state)
 
 {-# SPECIALISE getCompleted ::
      Ref (MergingTree IO h)
@@ -298,23 +325,6 @@ getCompleted (DeRef MergingTree {mergeState}) =
       OngoingTreeMerge{}   -> pure Nothing
       PendingTreeMerge{}   -> pure Nothing
 
-{-# SPECIALISE isStructurallyEmpty :: Ref (MergingTree IO h) -> IO Bool #-}
--- | Test if a 'MergingTree' is \"obviously\" empty by virtue of its structure.
--- This is not the same as being empty due to a pending or ongoing merge
--- happening to produce an empty run.
---
-isStructurallyEmpty :: MonadMVar m => Ref (MergingTree m h) -> m Bool
-isStructurallyEmpty (DeRef MergingTree {mergeState}) =
-    isStructurallyEmptyState <$> readMVar mergeState
-
-isStructurallyEmptyState :: MergingTreeState m h -> Bool
-isStructurallyEmptyState = \case
-    PendingTreeMerge (PendingLevelMerge prs Nothing) -> V.null prs
-    PendingTreeMerge (PendingUnionMerge mts)         -> V.null mts
-    _                                                -> False
-    -- It may also turn out to be useful to consider CompletedTreeMerge with
-    -- a zero length runs as empty.
-
 {-# SPECIALISE mkMergingTree ::
      RefCtx
   -> MergingTreeState IO h
@@ -467,22 +477,6 @@ supplyCredits hfs hbio refCtx resolve salt runParams threshold root uc = \mt0 c0
                 -- all work is done, we can't spend any more credits
                 pure (CompletedTreeMerge r, leftovers)
 
-          PendingTreeMerge _
-            | isStructurallyEmptyState state -> do
-            -- make a completely fresh empty run. this can only happen at the
-            -- root. the structurally empty tree still has debt 1, so we want to
-            -- merge it into a single run.
-            -- we handle this as a special case here since in several places
-            -- below we require the list of children to be non-empty.
-            runPaths <- mkFreshRunPaths
-            run <-
-              withRollback reg
-                -- TODO: the builder's handles aren't cleaned up if we fail
-                -- before fromBuilder closes them
-                (Run.newEmpty hfs hbio refCtx salt runParams runPaths)
-                releaseRef
-            pure (CompletedTreeMerge run, credits)
-
           PendingTreeMerge pm -> do
             leftovers <- supplyPending pm credits
             if leftovers <= 0

lsm-tree/src-core/Database/LSMTree/Internal/MergingTree/Lookup.hs

@@ -91,8 +91,6 @@ mergeLookupAcc resolve mt accs =
 -- | Create a 'LookupTree' of batches of runs, e.g. to do lookups on. The
 -- entries within each batch are to be combined using 'MR.MergeLevel'.
 --
--- Assumes that the merging tree is not 'MT.isStructurallyEmpty'.
---
 -- This function duplicates the references to all the tree's runs. These
 -- references later need to be released.
 {-# SPECIALISE buildLookupTree ::

lsm-tree/src-core/Database/LSMTree/Internal/Snapshot.hs

@@ -49,7 +49,7 @@ import           Data.Text (Text)
 import qualified Data.Vector as V
 import qualified Database.LSMTree.Internal.BloomFilter as Bloom
 import           Database.LSMTree.Internal.Config
-import           Database.LSMTree.Internal.CRC32C (checkCRC)
+import           Database.LSMTree.Internal.CRC32C (checkCRC, expectValidFile)
 import qualified Database.LSMTree.Internal.CRC32C as CRC
 import qualified Database.LSMTree.Internal.FS as FS
 import           Database.LSMTree.Internal.IncomingRun
@@ -203,9 +203,13 @@ instance NFData r => NFData (SnapMergingTreeState r) where
   rnf (SnapOngoingTreeMerge a)   = rnf a
 
 data SnapPendingMerge r =
+    -- | INVARIANT: There is always at least one 'SnapPreExistingRun'. See
+    -- 'MT.PendingMerge'.
     SnapPendingLevelMerge
       ![SnapPreExistingRun r]
       !(Maybe (SnapMergingTree r))
+    -- | INVARIANT: There is always at least one 'SnapMergingRun'. See
+    -- 'MT.PendingMerge'.
   | SnapPendingUnionMerge
       ![SnapMergingTree r]
   deriving stock (Eq, Functor, Foldable, Traversable)
@@ -234,6 +238,7 @@ instance NFData r => NFData (SnapPreExistingRun r) where
   -> Bloom.Salt
   -> UniqCounter IO
   -> ResolveSerialisedValue
+  -> FS.FsPath
   -> ActiveDir
   -> ActionRegistry IO
   -> SnapMergingTree (Ref (Run IO h))
@@ -250,11 +255,12 @@ fromSnapMergingTree ::
   -> Bloom.Salt
   -> UniqCounter m
   -> ResolveSerialisedValue
+  -> FS.FsPath -- ^ Path to source file for error reporting
   -> ActiveDir
   -> ActionRegistry m
   -> SnapMergingTree (Ref (Run m h))
   -> m (Ref (MT.MergingTree m h))
-fromSnapMergingTree hfs hbio refCtx salt uc resolve dir =
+fromSnapMergingTree hfs hbio refCtx salt uc resolve contentPath dir =
     go
   where
     -- Reference strategy:
@@ -276,21 +282,32 @@ fromSnapMergingTree hfs hbio refCtx salt uc resolve dir =
                               (SnapPendingLevelMerge prs mmt))) = do
       prs' <- traverse (fromSnapPreExistingRun reg) prs
       mmt' <- traverse (go reg) mmt
-      mt   <- withRollback reg
+      mt   <- withRollbackMaybe reg
                 (MT.newPendingLevelMerge refCtx prs' mmt')
                 releaseRef
       traverse_ (delayedCommit reg . releasePER) prs'
       traverse_ (delayedCommit reg . releaseRef) mmt'
-      pure mt
+      expectValidFile hfs contentPath CRC.FormatSnapshotMetaData $
+        case mt of
+          Nothing ->
+            Left "structurally empty pending level merge in snapshot"
+          Just mt' ->
+            Right mt'
 
     go reg (SnapMergingTree (SnapPendingTreeMerge
                               (SnapPendingUnionMerge mts))) = do
       mts' <- traverse (go reg) mts
-      mt   <- withRollback reg
+      mt   <- withRollbackMaybe reg
                 (MT.newPendingUnionMerge refCtx mts')
                 releaseRef
       traverse_ (delayedCommit reg . releaseRef) mts'
-      pure mt
+      expectValidFile hfs contentPath CRC.FormatSnapshotMetaData $
+        case mt of
+          Nothing ->
+            -- TODO: fail properly, or enforce invariant in SnapPendingUnionMerge
+            Left "structurally empty pending union merge in snapshot"
+          Just mt' ->
+            Right mt'
 
     go reg (SnapMergingTree (SnapOngoingTreeMerge smrs)) = do
       mr <- withRollback reg