Allow ManagedDB to set a discard below timestamp. So, only versions b…

…elow this timestamp get discarded, hence ensuring that we keep all the versions since the last snapshot.

levels.go

@@ -308,10 +308,10 @@ func (s *levelsController) compactBuildTables(
 
 	it.Rewind()
 
-	// Pick up the currently pending transactions' min readTs, so we can discard versions below this
-	// readTs. We should never discard any versions starting from above this timestamp, because that
-	// would affect the snapshot view guarantee provided by transactions.
-	minReadTs := s.kv.orc.readMark.MinReadTs()
+	// Pick a discard ts, so we can discard versions below this ts. We should
+	// never discard any versions starting from above this timestamp, because
+	// that would affect the snapshot view guarantee provided by transactions.
+	discardTs := s.kv.orc.discardAtOrBelow()
 
 	// Start generating new tables.
 	type newTableResult struct {
@@ -350,7 +350,7 @@ func (s *levelsController) compactBuildTables(
 
 			vs := it.Value()
 			version := y.ParseTs(it.Key())
-			if version <= minReadTs {
+			if version <= discardTs {
 				// Keep track of the number of versions encountered for this key. Only consider the
 				// versions which are below the minReadTs, otherwise, we might end up discarding the
 				// only valid version for a running transaction.

managed_db.go

@@ -77,3 +77,10 @@ func (txn *Txn) CommitAt(commitTs uint64, callback func(error)) error {
 func (db *ManagedDB) GetSequence(_ []byte, _ uint64) (*Sequence, error) {
 	panic("Cannot use GetSequence for ManagedDB.")
 }
+
+// SetDiscardTs sets a timestamp at or below which, any invalid or deleted
+// versions can be discarded from the LSM tree, and thence from the value log to
+// reclaim disk space.
+func (db *ManagedDB) SetDiscardTs(ts uint64) {
+	db.orc.setDiscardTs(ts)
+}

transaction.go

@@ -40,7 +40,10 @@ type oracle struct {
 	writeLock  sync.Mutex
 	nextCommit uint64
 
-	readMark y.WaterMark
+	// Either of these is used to determine which versions can be permanently
+	// discarded during compaction.
+	discardTs uint64      // Used by ManagedDB.
+	readMark  y.WaterMark // Used by DB.
 
 	// commits stores a key fingerprint and latest commit counter for it.
 	// refCount is used to clear out commits map to avoid a memory blowup.
@@ -81,6 +84,23 @@ func (o *oracle) commitTs() uint64 {
 	return o.nextCommit
 }
 
+// Any deleted or invalid versions at or below ts would be discarded during
+// compaction to reclaim disk space in LSM tree and thence value log.
+func (o *oracle) setDiscardTs(ts uint64) {
+	o.Lock()
+	defer o.Unlock()
+	o.discardTs = ts
+}
+
+func (o *oracle) discardAtOrBelow() uint64 {
+	if o.isManaged {
+		o.Lock()
+		defer o.Unlock()
+		return o.discardTs
+	}
+	return o.readMark.MinReadTs()
+}
+
 // hasConflict must be called while having a lock.
 func (o *oracle) hasConflict(txn *Txn) bool {
 	if len(txn.reads) == 0 {