/**
 * Copyright (c) 2018-present, heineiuo.
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
 */ import { assert } from "./DBHelper" ; import Version from "./Version" ; import { getCurrentFilename , getManifestFilename } from "./Filename" ; import Slice from "./Slice" ; import { getExpandedCompactionByteSizeLimit , getMaxBytesForLevel , } from "./VersionFormat" ; import Status from "./Status" ; import { FileMetaData } from "./VersionFormat" ; import VersionBuilder from "./VersionBuilder" ; import VersionEditRecord from "./VersionEditRecord" ; import LogReader from "./LogReader" ; import VersionEdit from "./VersionEdit" ; import { Config , InternalKeyComparator , InternalKey , Entry } from "./Format" ; import LogWriter from "./LogWriter" ; import Compaction from "./Compaction" ; import { Options , ReadOptions } from "./Options" ; import { TableCache } from "./SSTableCache" ; import Merger from "./Merger" ; import { decodeFixed64 } from "./Coding" ; interface RecoverResult { saveManifest ? : boolean ; } export default class VersionSet { compactPointers : Slice [ ] ; _current ! : Version ; _dummyVersions : Version ; hasLogNumber ? : boolean ; hasNextFileNumber ? : boolean ; hasPrevLogNumber ? : boolean ; logNumber = 0 ; prevLogNumber = 0 ; private _lastSequence = 0n ; hasLastSequence ? : boolean ; manifestFileNumber = 0 ; nextFileNumber = 2 ; private _dbpath : string ; _options : Options ; internalKeyComparator : InternalKeyComparator ; public tableCache : TableCache ; manifestWriter ? : LogWriter ; constructor ( dbpath , options , tableCache , internalKeyComparator ) { this . _dbpath = dbpath ; this . _options = options ; this . tableCache = tableCache ; this . internalKeyComparator = internalKeyComparator ; this . _dummyVersions = new Version ( this ) ; this . appendVersion ( new Version ( this ) ) ; this . compactPointers = [ ] ; } get lastSequence ( ) : bigint { return this . _lastSequence ; } set lastSequence ( value ) { this . _lastSequence = value ; } get current ( ) : Version { return this . _current ; } public getLevelSummary ( ) { let summary = "files [ " ; for ( let i = 0 ; i < this . _current . files . length ; i ++ ) { summary += `${ this . _current . files [ i ] . length } ` ; } summary += "]" ; return summary ; } public compactRange ( level , begin , end ) { let inputs = this . _current . getOverlappingInputs ( level , begin , end ) ; if ( inputs . length === 0 ) return ; if ( level > 0 ) { const limit = this . maxFileSizeForLevel ( this . _options , level ) ; let total = 0 ; for ( let i = 0 ; i < inputs . length ; i ++ ) { total += inputs [ i ] . fileSize ; if ( total >= limit ) { inputs = inputs . slice ( 0 , i + 1 ) ; break ; } } } const compaction = new Compaction ( this . _options , level ) ; compaction . inputVersion = this . _current ; compaction . inputVersion . ref ( ) ; compaction . inputs [ 0 ] = inputs ; this . setupOtherInputs ( compaction ) ; return compaction ; } private maxFileSizeForLevel ( options , level ) { return options . maxFileSize ; } public getNextFileNumber ( ) { return this . nextFileNumber ++ ; } public getNumLevelFiles ( level ) { assert ( level >= 0 ) ; assert ( level <= Config . kNumLevels ) ; return this . _current . files [ level ] . length ; } public recover = async ( ) : Promise < RecoverResult > => { const result = { } ; const current = await this . _options . env . readFile ( getCurrentFilename ( this . _dbpath ) , "utf8" ) ; if ( ! current || current [ current . length - 1 ] !== "\n" ) { throw new Error ( "Invalid format of CURRENT file." ) ; } let hasLogNumber = false ; let hasNextFileNumber = false ; let hasPrevLogNumber = false ; let hasLastSequence = false ; let logNumber = 0 ; let nextFileNumber = 0 ; let prevLogNumber = 0 ; let lastSequence = 0n ; const builder = new VersionBuilder ( this , this . _current ) ; const currentValue = current . substr ( 0 , current . length - 1 ) ; const manifestNumber = Number ( currentValue . substr ( "MANIFEST-" . length ) ) ; const reader = new LogReader ( await this . _options . env . open ( getManifestFilename ( this . _dbpath , manifestNumber ) , "r" ) ) ; for await ( const editSlice of reader . iterator ( ) ) { const edit = VersionEditRecord . decode ( editSlice ) ; builder . apply ( edit ) ; if ( edit . hasLogNumber ) { logNumber = edit . logNumber ; hasLogNumber = true ; } if ( edit . hasPrevLogNumber ) { prevLogNumber = edit . prevLogNumber ; hasPrevLogNumber = true ; } if ( edit . hasNextFileNumber ) { nextFileNumber = edit . nextFileNumber ; hasNextFileNumber = true ; } if ( edit . hasLastSequence ) { lastSequence = edit . lastSequence ; hasLastSequence = true ; } } if ( ! hasNextFileNumber ) { throw new Error ( "no meta-nextfile entry in descriptor" ) ; } else if ( ! hasLogNumber ) { throw new Error ( "no meta-lognumber entry in descriptor" ) ; } else if ( ! hasLastSequence ) { throw new Error ( "no last-sequence-number entry in descriptor" ) ; } if ( ! hasPrevLogNumber ) { prevLogNumber = 0 ; } this . markFileNumberUsed ( prevLogNumber ) ; this . markFileNumberUsed ( logNumber ) ; const version = new Version ( this ) ; builder . saveTo ( version ) ; this . finalize ( version ) ; this . appendVersion ( version ) ; this . manifestFileNumber = nextFileNumber ; this . nextFileNumber = nextFileNumber + 1 ; this . lastSequence = lastSequence ; this . logNumber = logNumber ; this . prevLogNumber = prevLogNumber ; if ( this . reuseManifest ( ) ) { } else { result . saveManifest = true ; } await reader . close ( ) ; return result ; } ; public markFileNumberUsed ( num ) { if ( this . nextFileNumber <= num ) { this . nextFileNumber = num + 1 ; } } private finalize ( ver ) { let bestLevel = - 1 ; let bestScore = - 1 ; for ( let level = 0 ; level < Config . kNumLevels ; level ++ ) { let score = 0 ; if ( level === 0 ) { score = ver . files [ level ] . length / Config . kL0CompactionTrigger ; } else { const levelBytes = this . getTotalBytes ( ver . files [ level ] ) ; if ( this . _options . debug ) this . _options . log ( `DEBUG level=${ level } levelBytes=${ levelBytes }` ) ; score = levelBytes / getMaxBytesForLevel ( level ) ; } if ( score > bestScore ) { bestScore = score ; bestLevel = level ; } } ver . compactionLevel = bestLevel ; ver . compactionScore = bestScore ; } private getTotalBytes ( files ) { let sum = 0 ; for ( const f of files ) { sum += f . fileSize ; } return sum ; } public async logAndApply ( edit ) { if ( edit . hasLogNumber ) { assert ( edit . logNumber >= this . logNumber ) ; assert ( edit . logNumber < this . nextFileNumber ) ; } else { edit . logNumber = this . logNumber ; } if ( ! edit . hasPrevLogNumber ) { edit . prevLogNumber = this . prevLogNumber ; } edit . nextFileNumber = this . nextFileNumber ; edit . lastSequence = this . lastSequence ; const ver = new Version ( this ) ; const builder = new VersionBuilder ( this , this . _current ) ; builder . apply ( edit ) ; builder . saveTo ( ver ) ; this . finalize ( ver ) ; let manifestFilename = "" ; let status = new Status ( ) ; if ( ! this . manifestWriter ) { manifestFilename = getManifestFilename ( this . _dbpath , this . manifestFileNumber ) ; edit . nextFileNumber = this . nextFileNumber ; this . manifestWriter = new LogWriter ( await this . _options . env . open ( manifestFilename , "a" ) ) ; if ( this . _options . debug ) this . _options . log ( "DEBUG writeSnapshot starting..." ) ; status = this . writeSnapshot ( this . manifestWriter ) ; } if ( await status . ok ( ) ) { const record = VersionEditRecord . add ( edit ) ; status = new Status ( this . manifestWriter . addRecord ( record ) ) ; } else { if ( this . _options . debug ) this . _options . log ( "DEBUG writeSnapshot fail" ) ; } if ( ( await status . ok ( ) ) && manifestFilename . length > 0 ) { status = new Status ( this . writeCurrentFile ( this . _dbpath , this . manifestFileNumber ) ) ; } if ( await status . ok ( ) ) { if ( ! ! manifestFilename ) { await this . manifestWriter . close ( ) ; delete this . manifestWriter ; } this . _options . log ( "DEBUG LogAndApply success, Install the new version" ) ; this . appendVersion ( ver ) ; this . logNumber = edit . logNumber ; this . prevLogNumber = edit . prevLogNumber ; } else { this . _options . log ( "DEBUG LogAndApply fail, Delete ver" ) ; if ( ! ! manifestFilename ) { await this . manifestWriter . close ( ) ; delete this . manifestWriter ; await this . _options . env . unlink ( manifestFilename ) ; } } return status ; } public needsCompaction ( ) { return this . _current . compactionScore >= 1 || ! ! this . _current . fileToCompact ; } /**
     * update this._current
     */ private appendVersion ( ver ) { assert ( ver . refs === 0 ) ; assert ( ver !== this . _current ) ; if ( this . _current ) { this . _current . unref ( ) ; } this . _current = ver ; ver . ref ( ) ; ver . prev = this . _dummyVersions . prev ; ver . next = this . _dummyVersions ; ver . prev . next = ver ; ver . next . prev = ver ; } private reuseManifest ( ) { return false ; } /**
     * dump current to manifest
     */ private writeSnapshot ( writer ) { const edit = new VersionEdit ( ) ; edit . comparator = this . internalKeyComparator . userComparator . getName ( ) ; for ( let level = 0 ; level < Config . kNumLevels ; level ++ ) { if ( ! ! this . compactPointers [ level ] && this . compactPointers [ level ] . length !== 0 ) { const key = new InternalKey ( ) ; key . decodeFrom ( this . compactPointers [ level ] ) ; edit . setCompactPointer ( level , key ) ; } } for ( let level = 0 ; level < Config . kNumLevels ; level ++ ) { const files = this . _current . files [ level ] ; for ( let i = 0 ; i < files . length ; i ++ ) { const f = files [ i ] ; edit . addFile ( level , f . number , f . fileSize , f . smallest , f . largest ) ; } } const record = VersionEditRecord . add ( edit ) ; const status = new Status ( writer . addRecord ( record ) ) ; return status ; } private async writeCurrentFile ( dbpath , manifestFileNumber ) { const currentFilename = getCurrentFilename ( dbpath ) ; let manifestFilename = getManifestFilename ( dbpath , manifestFileNumber ) ; assert ( manifestFilename . startsWith ( dbpath + "/" ) ) ; manifestFilename = manifestFilename . substr ( dbpath . length + 1 ) ; await this . _options . env . writeFile ( currentFilename , manifestFilename + "\n" ) ; } public pickCompaction ( ) { const shouldSizeCompaction = this . _current . compactionScore >= 1 ; const shouldSeekCompaction = ! ! this . _current . fileToCompact ; let c ; let level ; if ( shouldSizeCompaction ) { level = this . _current . compactionLevel ; assert ( level >= 0 ) ; assert ( level + 1 < Config . kNumLevels ) ; c = new Compaction ( this . _options , level ) ; for ( const f of this . _current . files [ level ] ) { if ( ! this . compactPointers [ level ] || this . compactPointers [ level ] . length === 0 || this . internalKeyComparator . compare ( f . largest , this . compactPointers [ level ] ) > 0 ) { c . inputs [ 0 ] . push ( f ) ; break ; } } if ( c . inputs [ 0 ] . length === 0 ) { c . inputs [ 0 ] . push ( this . _current . files [ level ] [ 0 ] ) ; } } else if ( shouldSeekCompaction ) { level = this . _current . fileToCompactLevel ; c = new Compaction ( this . _options , level ) ; c . inputs [ 0 ] . push ( this . _current . fileToCompact ) ; } else { return ; } c . inputVersion = this . current ; c . inputVersion . ref ( ) ; if ( level === 0 ) { const smallest = new InternalKey ( ) ; const largest = new InternalKey ( ) ; this . getRange ( c . inputs [ 0 ] , smallest , largest ) ; c . inputs [ 0 ] = this . _current . getOverlappingInputs ( 0 , smallest , largest ) ; assert ( c . inputs [ 0 ] . length > 0 ) ; } this . setupOtherInputs ( c ) ; return c ; } /**
     * Stores the minimal range that covers all entries in inputs in
     * smallest, *largest.
     * REQUIRES: inputs is not empty
     */ private getRange ( inputs , smallest , largest ) { assert ( inputs . length > 0 ) ; smallest . clear ( ) ; largest . clear ( ) ; for ( let i = 0 ; i < inputs . length ; i ++ ) { const fileMetaData = inputs [ i ] ; if ( i === 0 ) { smallest . buffer = fileMetaData . smallest . buffer ; largest . buffer = fileMetaData . largest . buffer ; } else { if ( this . internalKeyComparator . compare ( fileMetaData . smallest , smallest ) < 0 ) { smallest . buffer = fileMetaData . smallest . buffer ; } if ( this . internalKeyComparator . compare ( fileMetaData . largest , largest ) > 0 ) { largest . buffer = fileMetaData . largest . buffer ; } } } } /**
     * Stores the minimal range that covers all entries in inputs1 and inputs2
     * in *smallest, *largest.
     * REQUIRES: inputs is not empty
     */ private getRange2 ( inputs1 , inputs2 , smallest , largest ) { const all = inputs1 . concat ( inputs2 ) ; this . getRange ( all , smallest , largest ) ; } private findLargestKey ( icmp , files , largestKey ) { if ( files . length === 0 ) return false ; largestKey . buffer = files [ 0 ] . largest . buffer ; for ( let i = 0 ; i < files . length ; i ++ ) { const f = files [ i ] ; if ( icmp . compare ( f . largest , largestKey ) > 0 ) { largestKey . buffer = f . largest . buffer ; } } return true ; } private addBoundaryInputs ( icmp , levelFiles , compactionFiles ) { let largestKey = new InternalKey ( ) ; if ( ! this . findLargestKey ( icmp , compactionFiles , largestKey ) ) { return ; } while ( true ) { const smallestBoundaryFile = this . findSmallestBoundaryFile ( icmp , levelFiles , largestKey ) ; if ( ! smallestBoundaryFile ) break ; largestKey = smallestBoundaryFile . largest ; compactionFiles . push ( smallestBoundaryFile ) ; } } public findSmallestBoundaryFile ( icmp , levelFiles , largestKey ) { const userComparator = icmp . userComparator ; let smallestBoundaryFile ! ; for ( let i = 0 ; i < levelFiles . length ; i ++ ) { const f = levelFiles [ i ] ; if ( icmp . compare ( f . smallest , largestKey ) > 0 && userComparator . compare ( f . smallest . userKey , largestKey . userKey ) === 0 ) { if ( ! smallestBoundaryFile || icmp . compare ( f . smallest , smallestBoundaryFile . smallest ) < 0 ) { smallestBoundaryFile = f ; } } } return smallestBoundaryFile ; } public setupOtherInputs ( c ) { const level = c . level ; let smallest = new InternalKey ( ) ; let largest = new InternalKey ( ) ; this . addBoundaryInputs ( this . internalKeyComparator , this . _current . files [ level ] , c . inputs [ 0 ] ) ; this . getRange ( c . inputs [ 0 ] , smallest , largest ) ; c . inputs [ 1 ] = this . current . getOverlappingInputs ( level + 1 , smallest , largest ) ; const allStart = new InternalKey ( ) ; const allLimit = new InternalKey ( ) ; this . getRange2 ( c . inputs [ 0 ] , c . inputs [ 1 ] , allStart , allLimit ) ; if ( c . inputs . length > 0 ) { const expand0 = this . current . getOverlappingInputs ( level , allStart , allLimit ) ; this . addBoundaryInputs ( this . internalKeyComparator , this . _current . files [ level ] , expand0 ) ; const input0Size = this . getTotalBytes ( c . inputs [ 0 ] ) ; const input1Size = this . getTotalBytes ( c . inputs [ 1 ] ) ; const expand0Size = this . getTotalBytes ( expand0 ) ; if ( expand0 . length > c . inputs [ 0 ] . length && input1Size + expand0Size < getExpandedCompactionByteSizeLimit ( this . _options ) ) { const newStart = new InternalKey ( ) ; const newLimit = new InternalKey ( ) ; this . getRange ( expand0 , newStart , newLimit ) ; const expand1 = this . _current . getOverlappingInputs ( level + 1 , newStart , newLimit ) ; if ( expand1 . length === c . inputs [ 1 ] . length ) { this . _options . log ( `Expanding@${ level } ${ c . inputs [ 0 ] . length }+${ c . inputs [ 1 ] . length }` + ` (${ input0Size }+${ input1Size } bytes) to ${ expand0Size }` + `+${ expand1 . length } (${ expand0Size }+${ input1Size } bytes)` ) ; smallest = newStart ; largest = newLimit ; c . inputs [ 0 ] = expand0 ; c . inputs [ 1 ] = expand1 ; this . getRange2 ( c . inputs [ 0 ] , c . inputs [ 1 ] , allStart , allLimit ) ; } } } if ( level + 2 < Config . kNumLevels ) { c . grandparents = this . _current . getOverlappingInputs ( level + 2 , allStart , allLimit ) ; } this . compactPointers [ level ] = new Slice ( largest . buffer ) ; c . edit . compactPointers . push ( { level , internalKey : largest } ) ; } public addLiveFiles ( live ) { for ( let ver = this . _dummyVersions . next ; ver != this . _dummyVersions ; ver = ver . next ) { for ( let level = 0 ; level < Config . kNumLevels ; level ++ ) { const files = ver . files [ level ] ; for ( let i = 0 ; i < files . length ; i ++ ) { live . add ( files [ i ] . number ) ; } } } } public async * makeInputIterator ( currentCompaction ) { let num = 0 ; const options = { } as ReadOptions ; const space = currentCompaction . level === 0 ? currentCompaction . inputs [ 0 ] . length + 1 : 2 ; const list = Array . from ( { length : space } ) ; for ( let which = 0 ; which < 2 ; which ++ ) { if ( currentCompaction . inputs [ which ] . length > 0 ) { if ( currentCompaction . level + which === 0 ) { const files = currentCompaction . inputs [ which ] ; for ( let i = 0 ; i < files . length ; i ++ ) { list [ num ++ ] = this . tableCache . entryIterator ( options , files [ i ] . number , files [ i ] . fileSize ) ; } } else { const files = currentCompaction . inputs [ which ] ; list [ num ++ ] = this . levelFileEntryIterator ( files ) ; } } } assert ( num <= space ) ; const merger = new Merger ( this . internalKeyComparator , list , num ) ; yield * merger . iterator ( ) ; } private async * levelFileEntryIterator ( files ) { const options = { } as ReadOptions ; for ( const fileEntry of Version . levelFileNumIterator ( this . internalKeyComparator , files ) ) { const fileNumber = decodeFixed64 ( fileEntry . value . buffer . slice ( 0 , 8 ) ) ; const fileSize = decodeFixed64 ( fileEntry . value . buffer . slice ( 8 ) ) ; yield * this . tableCache . entryIterator ( options , Number ( fileNumber ) , Number ( fileSize ) ) ; } } }