[no-release-notes] go: sqle: dolt_gc: Purge read caches associated with the DoltDB as we begin GC.

go/libraries/doltcore/doltdb/doltdb.go

@@ -2191,6 +2191,15 @@ func (ddb *DoltDB) PersistGhostCommits(ctx context.Context, ghostCommits hash.Ha
 	return ddb.db.Database.PersistGhostCommitIDs(ctx, ghostCommits)
 }
 
+// Purge in-memory read caches associated with this DoltDB. This needs
+// to be done at a specific point during a GC operation to ensure that
+// everything the application layer sees still exists in the database
+// after the GC operation is completed.
+func (ddb *DoltDB) PurgeCaches() {
+	ddb.vrw.PurgeCaches()
+	ddb.ns.PurgeCaches()
+}
+
 type FSCKReport struct {
 	ChunkCount uint32
 	Problems   []error

go/libraries/doltcore/sqle/dprocedures/dolt_gc.go

@@ -27,6 +27,7 @@ import (
 	"github.com/dolthub/dolt/go/cmd/dolt/cli"
 	"github.com/dolthub/dolt/go/libraries/doltcore/branch_control"
 	"github.com/dolthub/dolt/go/libraries/doltcore/dconfig"
+	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess"
 	"github.com/dolthub/dolt/go/store/hash"
 	"github.com/dolthub/dolt/go/store/types"
@@ -73,10 +74,12 @@ var ErrServerPerformedGC = errors.New("this connection was established when this
 // invalidated in such a way that all future queries on it return an error.
 type killConnectionsSafepointController struct {
 	callCtx   *sql.Context
+	doltDB    *doltdb.DoltDB
 	origEpoch int
 }
 
 func (sc killConnectionsSafepointController) BeginGC(ctx context.Context, keeper func(hash.Hash) bool) error {
+	sc.doltDB.PurgeCaches()
 	return nil
 }
 
@@ -157,6 +160,7 @@ type sessionAwareSafepointController struct {
 	controller *dsess.GCSafepointController
 	callCtx    *sql.Context
 	origEpoch  int
+	doltDB     *doltdb.DoltDB
 
 	waiter *dsess.GCSafepointWaiter
 	keeper func(hash.Hash) bool
@@ -167,6 +171,7 @@ func (sc *sessionAwareSafepointController) visit(ctx context.Context, sess *dses
 }
 
 func (sc *sessionAwareSafepointController) BeginGC(ctx context.Context, keeper func(hash.Hash) bool) error {
+	sc.doltDB.PurgeCaches()
 	sc.keeper = keeper
 	thisSess := dsess.DSessFromSess(sc.callCtx.Session)
 	err := sc.visit(ctx, thisSess)
@@ -256,11 +261,13 @@ func doDoltGC(ctx *sql.Context, args []string) (int, error) {
 				origEpoch:  origepoch,
 				callCtx:    ctx,
 				controller: gcSafepointController,
+				doltDB:     ddb,
 			}
 		} else {
 			sc = killConnectionsSafepointController{
 				origEpoch: origepoch,
 				callCtx:   ctx,
+				doltDB:    ddb,
 			}
 		}
 		err = ddb.GC(ctx, mode, sc)

go/store/prolly/tree/node_cache.go

@@ -48,6 +48,12 @@ func (c nodeCache) insert(addr hash.Hash, node Node) {
 	s.insert(addr, node)
 }
 
+func (c nodeCache) purge() {
+	for _, s := range c.stripes {
+		s.purge()
+	}
+}
+
 type centry struct {
 	a    hash.Hash
 	n    Node
@@ -83,6 +89,15 @@ func removeFromList(e *centry) {
 	e.next = e
 }
 
+func (s *stripe) purge() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.chunks = make(map[hash.Hash]*centry)
+	s.head = nil
+	s.sz = 0
+	s.rev = 0
+}
+
 func (s *stripe) moveToFront(e *centry) {
 	e.i = s.rev
 	s.rev++

go/store/prolly/tree/node_cache_test.go

@@ -0,0 +1,48 @@
+// Copyright 2021 Dolthub, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package tree
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/dolthub/dolt/go/store/hash"
+)
+
+func TestNodeCache(t *testing.T) {
+	t.Run("InsertGetPurge", func(t *testing.T) {
+		// Simple smoke screen test of insert, get, purge.
+		var addr hash.Hash
+		var n Node
+		n.msg = make([]byte, 1024)
+		cache := newChunkCache(64 * 1024)
+		for i := 0; i < 32; i++ {
+			addr[0] = byte(i)
+			cache.insert(addr, n)
+		}
+		for i := 0; i < 32; i++ {
+			addr[0] = byte(i)
+			_, ok := cache.get(addr)
+			assert.True(t, ok)
+		}
+		cache.purge()
+		for i := 0; i < 32; i++ {
+			addr[0] = byte(i)
+			_, ok := cache.get(addr)
+			assert.False(t, ok)
+		}
+	})
+}

go/store/prolly/tree/node_store.go

@@ -49,6 +49,8 @@ type NodeStore interface {
 
 	BlobBuilder() *BlobBuilder
 	PutBlobBuilder(*BlobBuilder)
+
+	PurgeCaches()
 }
 
 type nodeStore struct {
@@ -194,3 +196,7 @@ func (ns nodeStore) Format() *types.NomsBinFormat {
 	}
 	return nbf
 }
+
+func (ns nodeStore) PurgeCaches() {
+	ns.cache.purge()
+}

go/store/prolly/tree/testutils.go

@@ -333,6 +333,10 @@ func (v nodeStoreValidator) PutBlobBuilder(bb *BlobBuilder) {
 	v.bbp.Put(bb)
 }
 
+func (v nodeStoreValidator) PurgeCaches() {
+	v.ns.PurgeCaches()
+}
+
 func (v nodeStoreValidator) Format() *types.NomsBinFormat {
 	return v.ns.Format()
 }

go/store/types/value_store.go

@@ -51,6 +51,7 @@ type ValueReader interface {
 // package that implements Value writing.
 type ValueWriter interface {
 	WriteValue(ctx context.Context, v Value) (Ref, error)
+	PurgeCaches()
 }
 
 // ValueReadWriter is an interface that knows how to read and write Noms
@@ -739,10 +740,6 @@ func (lvs *ValueStore) GC(ctx context.Context, mode GCMode, oldGenRefs, newGenRe
 		return chunks.ErrUnsupportedOperation
 	}
 
-	// TODO: The decodedChunks cache can potentially allow phantom reads of
-	// already collected chunks until we clear it...
-	lvs.decodedChunks.Purge()
-
 	if tfs, ok := lvs.cs.(chunks.TableFileStore); ok {
 		return tfs.PruneTableFiles(ctx)
 	}
@@ -809,6 +806,10 @@ func (lvs *ValueStore) gc(ctx context.Context,
 	return finalizer, sweeper.Close(ctx)
 }
 
+func (lvs *ValueStore) PurgeCaches() {
+	lvs.decodedChunks.Purge()
+}
+
 // Close closes the underlying ChunkStore
 func (lvs *ValueStore) Close() error {
 	return lvs.cs.Close()

go/store/valuefile/file_value_store.go

@@ -161,6 +161,9 @@ func (f *FileValueStore) CacheHas(h hash.Hash) bool {
 	return ok
 }
 
+func (f *FileValueStore) PurgeCaches() {
+}
+
 // HasMany returns the set of hashes that are absent from the store
 func (f *FileValueStore) HasMany(ctx context.Context, hashes hash.HashSet) (absent hash.HashSet, err error) {
 	f.chunkLock.Lock()