Merge pull request #9726 from dolthub/aaron/auto-gc-for-dolt-sql

go: cmd/dolt: sql.go: Enable Auto GC when running `dolt sql`.

Author: reltuk

go/cmd/dolt/cli/cli_context.go

@@ -33,6 +33,12 @@ type LateBindQueryistResult struct {
 	Closer   func()
 }
 
+type LateBindQueryistConfig struct {
+	EnableAutoGC bool
+}
+
+type LateBindQueryistOption func(*LateBindQueryistConfig)
+
 // LateBindQueryist is a function that will be called the first time Queryist is needed for use. Input is a context which
 // is appropriate for the call to commence. Output is a LateBindQueryistResult, which includes a Queryist, a sql.Context, and
 // a closer function. It can also result in an error.
@@ -48,15 +54,15 @@ type LateBindQueryistResult struct {
 // This state is useful for determining whether a command making use of the CliContext is being run within the context of
 // another command. This is particularly interesting when running a \checkout in a dolt sql session. It makes sense to do
 // so in the context of `dolt sql`, but not in the context of `dolt checkout` when connected to a remote server.
-type LateBindQueryist func(ctx context.Context) (LateBindQueryistResult, error)
+type LateBindQueryist func(ctx context.Context, opts ...LateBindQueryistOption) (LateBindQueryistResult, error)
 
 // CliContexct is used to pass top level command information down to subcommands.
 type CliContext interface {
 	// GlobalArgs returns the arguments passed before the subcommand.
 	GlobalArgs() *argparser.ArgParseResults
 	WorkingDir() filesys.Filesys
 	Config() *env.DoltCliConfig
-	QueryEngine(ctx context.Context) (QueryEngineResult, error)
+	QueryEngine(ctx context.Context, opts ...LateBindQueryistOption) (QueryEngineResult, error)
 	// Release resources associated with the CliContext, including
 	// any QueryEngines which were provisioned over the lifetime
 	// of the CliContext.
@@ -115,7 +121,7 @@ func (lbc LateBindCliContext) GlobalArgs() *argparser.ArgParseResults {
 // QueryEngine returns a Queryist, a sql.Context, a closer function, and an error. It ensures that only one call to the
 // LateBindQueryist is made, and caches the result. Note that if this is called twice, the closer function returns will
 // be nil, callers should check if is nil.
-func (lbc LateBindCliContext) QueryEngine(ctx context.Context) (res QueryEngineResult, err error) {
+func (lbc LateBindCliContext) QueryEngine(ctx context.Context, opts ...LateBindQueryistOption) (res QueryEngineResult, err error) {
 	if lbc.activeContext != nil && lbc.activeContext.qryist != nil && lbc.activeContext.sqlCtx != nil {
 		res.Queryist = *lbc.activeContext.qryist
 		res.Context = lbc.activeContext.sqlCtx
@@ -125,7 +131,7 @@ func (lbc LateBindCliContext) QueryEngine(ctx context.Context) (res QueryEngineR
 		return res, nil
 	}
 
-	bindRes, err := lbc.bind(ctx)
+	bindRes, err := lbc.bind(ctx, opts...)
 	if err != nil {
 		return res, err
 	}

go/cmd/dolt/commands/engine/sqlengine.go

@@ -87,6 +87,8 @@ type SqlEngineConfig struct {
 	EventSchedulerStatus       eventscheduler.SchedulerStatus
 }
 
+type SqlEngineConfigOption func(*SqlEngineConfig)
+
 // NewSqlEngine returns a SqlEngine
 func NewSqlEngine(
 	ctx context.Context,

go/cmd/dolt/commands/init_test.go

@@ -71,7 +71,9 @@ func TestInit(t *testing.T) {
 			gCfg, _ := dEnv.Config.GetConfig(env.GlobalConfig)
 			gCfg.SetStrings(test.GlobalConfig)
 			apr := argparser.ArgParseResults{}
-			latebind := func(ctx context.Context) (res cli.LateBindQueryistResult, err error) { return res, nil }
+			latebind := func(ctx context.Context, opts ...cli.LateBindQueryistOption) (res cli.LateBindQueryistResult, err error) {
+				return res, nil
+			}
 			cliCtx, _ := cli.NewCliContext(&apr, dEnv.Config, dEnv.FS, latebind)
 
 			result := InitCmd{}.Exec(ctx, "dolt init", test.Args, dEnv, cliCtx)

go/cmd/dolt/commands/sql.go

@@ -100,6 +100,7 @@ const (
 	outputFlag            = "output"
 	binaryAsHexFlag       = "binary-as-hex"
 	skipBinaryAsHexFlag   = "skip-binary-as-hex"
+	disableAutoGCFlag     = "disable-auto-gc"
 	// TODO: Consider simplifying to use MySQL's skip pattern with single flag definition
 	// MySQL handles both --binary-as-hex and --skip-binary-as-hex with one option definition
 	// and uses disabled_my_option to distinguish between enable/disable
@@ -155,6 +156,7 @@ func (cmd SqlCmd) ArgParser() *argparser.ArgParser {
 	ap.SupportsFlag(binaryAsHexFlag, "", "Print binary data as hex. Enabled by default for interactive terminals.")
 	// TODO: MySQL uses a skip- pattern for negating flags and doesn't show them in help
 	ap.SupportsFlag(skipBinaryAsHexFlag, "", "Disable binary data as hex output.")
+	ap.SupportsFlag(disableAutoGCFlag, "", "Disable automatically running GC.")
 	return ap
 }
 
@@ -227,7 +229,13 @@ func (cmd SqlCmd) Exec(ctx context.Context, commandStr string, args []string, dE
 		return HandleVErrAndExitCode(errhand.BuildDError("cannot use both --%s and --%s", binaryAsHexFlag, skipBinaryAsHexFlag).Build(), usage)
 	}
 
-	queryist, err := cliCtx.QueryEngine(ctx)
+	enableAutoGC := true
+	if apr.Contains(disableAutoGCFlag) {
+		enableAutoGC = false
+	}
+	queryist, err := cliCtx.QueryEngine(ctx, func(config *cli.LateBindQueryistConfig) {
+		config.EnableAutoGC = enableAutoGC
+	})
 	if err != nil {
 		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
 	}
@@ -618,6 +626,11 @@ func execBatchMode(ctx *sql.Context, qryist cli.Queryist, input io.Reader, conti
 	scanner := NewStreamScanner(input)
 	var query string
 	for scanner.Scan() {
+		// The session we get is wrapped in a command begin/end block.
+		// By ending the command and starting a new one, Auto GC is able
+		// to form safe points if/when it is enabled.
+		sql.SessionCommandEnd(ctx.Session)
+		sql.SessionCommandBegin(ctx.Session)
 		if fileReadProg != nil {
 			updateFileReadProgressOutput()
 			fileReadProg.setReadBytes(int64(len(scanner.Bytes())))
@@ -763,6 +776,12 @@ func execShell(sqlCtx *sql.Context, qryist cli.Queryist, format engine.PrintResu
 	lastSqlCmd := ""
 
 	shell.Uninterpreted(func(c *ishell.Context) {
+		// The session we get is wrapped in a command begin/end block.
+		// By ending the command and starting a new one, Auto GC is able
+		// to form safe points if/when it is enabled.
+		sql.SessionCommandEnd(sqlCtx.Session)
+		sql.SessionCommandBegin(sqlCtx.Session)
+
 		query := c.Args[0]
 		query = strings.TrimSpace(query)
 		if len(query) == 0 {

go/cmd/dolt/commands/sqlserver/queryist_utils.go

@@ -68,7 +68,7 @@ func BuildConnectionStringQueryist(ctx context.Context, cwdFS filesys.Filesys, c
 	gatherWarnings := false
 	queryist := ConnectionQueryist{connection: conn, gatherWarnings: &gatherWarnings}
 
-	var lateBind cli.LateBindQueryist = func(ctx context.Context) (res cli.LateBindQueryistResult, err error) {
+	var lateBind cli.LateBindQueryist = func(ctx context.Context, opts ...cli.LateBindQueryistOption) (res cli.LateBindQueryistResult, err error) {
 		sqlCtx := sql.NewContext(ctx)
 		sqlCtx.SetCurrentDatabase(dbRev)
 		res.Queryist = queryist

go/cmd/dolt/commands/utils.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"crypto/sha1"
 	"fmt"
+	"io"
 	"net"
 	"os"
 	"path/filepath"
@@ -31,6 +32,7 @@ import (
 	"github.com/fatih/color"
 	"github.com/gocraft/dbr/v2"
 	"github.com/gocraft/dbr/v2/dialect"
+	"github.com/sirupsen/logrus"
 
 	"github.com/dolthub/dolt/go/cmd/dolt/cli"
 	"github.com/dolthub/dolt/go/cmd/dolt/commands/engine"
@@ -39,11 +41,13 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 	"github.com/dolthub/dolt/go/libraries/doltcore/env/actions"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess"
 	"github.com/dolthub/dolt/go/libraries/utils/argparser"
 	"github.com/dolthub/dolt/go/libraries/utils/config"
 	"github.com/dolthub/dolt/go/libraries/utils/editor"
 	"github.com/dolthub/dolt/go/libraries/utils/filesys"
+	"github.com/dolthub/dolt/go/store/chunks"
 	"github.com/dolthub/dolt/go/store/datas"
 	"github.com/dolthub/dolt/go/store/util/outputpager"
 )
@@ -246,12 +250,25 @@ func newLateBindingEngine(
 		Autocommit:         true,
 	}
 
-	var lateBinder cli.LateBindQueryist = func(ctx context.Context) (res cli.LateBindQueryistResult, err error) {
+	var lateBinder cli.LateBindQueryist = func(ctx context.Context, opts ...cli.LateBindQueryistOption) (res cli.LateBindQueryistResult, err error) {
 		// We've deferred loading the database as long as we can.
 		// If we're binding the Queryist, that means that engine is actually
 		// going to be used.
 		mrEnv.ReloadDBs(ctx)
 
+		queryistConfig := &cli.LateBindQueryistConfig{}
+		for _, opt := range opts {
+			opt(queryistConfig)
+		}
+
+		if queryistConfig.EnableAutoGC {
+			// We use a null logger here, as we do not want `dolt sql` output
+			// to include auto-gc log lines.
+			nullLgr := logrus.New()
+			nullLgr.SetOutput(io.Discard)
+			config.AutoGCController = sqle.NewAutoGCController(chunks.NoArchive, nullLgr)
+		}
+
 		se, err := engine.NewSqlEngine(
 			ctx,
 			mrEnv,

go/cmd/dolt/dolt.go

@@ -664,7 +664,7 @@ If you're interested in running this command against a remote host, hit us up on
 	isValidRepositoryRequired := subcommandName != "init" && subcommandName != "sql" && subcommandName != "sql-server" && subcommandName != "sql-client"
 
 	if noValidRepository && isValidRepositoryRequired {
-		return func(ctx context.Context) (res cli.LateBindQueryistResult, err error) {
+		return func(ctx context.Context, opts ...cli.LateBindQueryistOption) (res cli.LateBindQueryistResult, err error) {
 			err = errors.New("The current directory is not a valid dolt repository.")
 			if errors.Is(rootEnv.DBLoadError, nbs.ErrUnsupportedTableFileFormat) {
 				// This is fairly targeted and specific to allow for better error messaging. We should consider

go/libraries/doltcore/sqle/auto_gc.go

@@ -178,13 +178,14 @@ func (c *AutoGCController) newCommitHook(name string, db *doltdb.DoltDB) *autoGC
 	closed := make(chan *gcWorkReport)
 	close(closed)
 	ret := &autoGCCommitHook{
-		c:      c,
-		name:   name,
-		done:   closed,
-		next:   make(chan *gcWorkReport, 1),
-		db:     db,
-		tickCh: make(chan struct{}),
-		stopCh: make(chan struct{}),
+		c:         c,
+		name:      name,
+		done:      closed,
+		next:      make(chan *gcWorkReport, 1),
+		db:        db,
+		tickCh:    make(chan struct{}),
+		stopCh:    make(chan struct{}),
+		stoppedCh: make(chan struct{}),
 	}
 	c.hooks[name] = ret
 	if c.threads != nil {
@@ -231,6 +232,11 @@ type autoGCCommitHook struct {
 	// Closed when the thread should shutdown because the database
 	// is being removed.
 	stopCh chan struct{}
+	// Closed as the background processing thread shuts down. The
+	// database hook selects on this to avoid deadlocking if it
+	// is trying to send to the worker thread after it has been
+	// shutdown.
+	stoppedCh chan struct{}
 	// An optimistic send on this channel notifies the background
 	// thread that the sizes may have changed and it can check for
 	// the GC condition.
@@ -278,6 +284,8 @@ func (h *autoGCCommitHook) Execute(ctx context.Context, _ datas.Dataset, _ *dolt
 	select {
 	case h.tickCh <- struct{}{}:
 		return nil, nil
+	case <-h.stoppedCh:
+		return nil, nil
 	case <-ctx.Done():
 		return nil, context.Cause(ctx)
 	}
@@ -357,6 +365,7 @@ func (h *autoGCCommitHook) checkForGC(ctx context.Context) error {
 
 func (h *autoGCCommitHook) thread(ctx context.Context) {
 	defer h.wg.Done()
+	defer close(h.stoppedCh)
 	timer := time.NewTimer(checkInterval)
 	defer timer.Stop()
 	for {

go/performance/microsysbench/sysbench_test.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/dolthub/go-mysql-server/server"
@@ -49,11 +50,6 @@ const (
 
 var dEnv *env.DoltEnv
 
-func init() {
-	dEnv = dtestutils.CreateTestEnv()
-	populateRepo(dEnv, readTestData(dataFile))
-}
-
 func BenchmarkOltpPointSelect(b *testing.B) {
 	benchmarkSysbenchQuery(b, func(int) string {
 		q := "SELECT c FROM sbtest1 WHERE id=%d"
@@ -119,7 +115,13 @@ func BenchmarkSelectRandomRanges(b *testing.B) {
 	})
 }
 
+var initOnce sync.Once
+
 func benchmarkSysbenchQuery(b *testing.B, getQuery func(int) string) {
+	initOnce.Do(func() {
+		dEnv = dtestutils.CreateTestEnv()
+		populateRepo(dEnv, readTestData(dataFile))
+	})
 	ctx, eng := setupBenchmark(b, dEnv)
 	for i := 0; i < b.N; i++ {
 		schema, iter, _, err := eng.Query(ctx, getQuery(i))

integration-tests/bats/gen_vals.awk

@@ -0,0 +1,23 @@
+function randint(n)
+{
+    return int(n * rand())
+}
+function valueslines(n)
+{
+    end = "),"
+    for (i = 0; i < n; i++) {
+        if (i == n-1) {
+            end = ");"
+        }
+        print "(" randint(65536) ",", randint(65536) ",", randint(65536) ",", randint(65536) end;
+    }
+}
+BEGIN {
+    print "DROP TABLE IF EXISTS vals;";
+    print "CREATE TABLE vals (c1 int, c2 int, c3 int, c4 int);";
+    for (j = 0; j < 256; j++) {
+        print "INSERT INTO vals VALUES";
+        valueslines(1024);
+    }
+}
+