diff --git a/go/cmd/dolt/commands/sql.go b/go/cmd/dolt/commands/sql.go
index 40381d35a99..c138859cbdf 100644
--- a/go/cmd/dolt/commands/sql.go
+++ b/go/cmd/dolt/commands/sql.go
@@ -105,6 +105,7 @@ const (
 	UserFlag              = "user"
 	DefaultUser           = "root"
 	DefaultHost           = "localhost"
+	UseDbFlag             = "use-db"
 
 	welcomeMsg = `# Welcome to the DoltSQL shell.
 # Statements must be terminated with ';'.
@@ -233,7 +234,7 @@ func (cmd SqlCmd) Exec(ctx context.Context, commandStr string, args []string, dE
 		fi, err := os.Stdin.Stat()
 		if err != nil {
 			if !osutil.IsWindows {
-				return HandleVErrAndExitCode(errhand.BuildDError("Couldn't stat STDIN. This is a bug.").Build(), usage)
+				return sqlHandleVErrAndExitCode(queryist, errhand.BuildDError("Couldn't stat STDIN. This is a bug.").Build(), usage)
 			}
 		} else {
 			isTty = fi.Mode()&os.ModeCharDevice != 0
@@ -246,11 +247,11 @@ func (cmd SqlCmd) Exec(ctx context.Context, commandStr string, args []string, dE
 			isTty = false
 			input, err = os.OpenFile(fileInput, os.O_RDONLY, os.ModePerm)
 			if err != nil {
-				return HandleVErrAndExitCode(errhand.BuildDError("couldn't open file %s", fileInput).Build(), usage)
+				return sqlHandleVErrAndExitCode(queryist, errhand.BuildDError("couldn't open file %s", fileInput).Build(), usage)
 			}
 			info, err := os.Stat(fileInput)
 			if err != nil {
-				return HandleVErrAndExitCode(errhand.BuildDError("couldn't get file size %s", fileInput).Build(), usage)
+				return sqlHandleVErrAndExitCode(queryist, errhand.BuildDError("couldn't get file size %s", fileInput).Build(), usage)
 			}
 
 			// initialize fileReadProg global variable if there is a file to process queries from
@@ -261,23 +262,23 @@ func (cmd SqlCmd) Exec(ctx context.Context, commandStr string, args []string, dE
 		if isTty {
 			err := execShell(sqlCtx, queryist, format)
 			if err != nil {
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+				return sqlHandleVErrAndExitCode(queryist, errhand.VerboseErrorFromError(err), usage)
 			}
 		} else if runInBatchMode {
 			se, ok := queryist.(*engine.SqlEngine)
 			if !ok {
 				misuse := fmt.Errorf("Using batch with non-local access pattern. Stop server if it is running")
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(misuse), usage)
+				return sqlHandleVErrAndExitCode(queryist, errhand.VerboseErrorFromError(misuse), usage)
 			}
 
 			verr := execBatch(sqlCtx, se, input, continueOnError, format)
 			if verr != nil {
-				return HandleVErrAndExitCode(verr, usage)
+				return sqlHandleVErrAndExitCode(queryist, verr, usage)
 			}
 		} else {
 			err := execMultiStatements(sqlCtx, queryist, input, continueOnError, format)
 			if err != nil {
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+				return sqlHandleVErrAndExitCode(queryist, errhand.VerboseErrorFromError(err), usage)
 			}
 		}
 	}
@@ -285,6 +286,35 @@ func (cmd SqlCmd) Exec(ctx context.Context, commandStr string, args []string, dE
 	return 0
 }
 
+// sqlHandleVErrAndExitCode is a helper function to print errors to the user. Currently, the Queryist interface is used to
+// determine if this is a local or remote execution. This is hacky, and too simplistic. We should possibly add an error
+// messaging interface to the CliContext.
+func sqlHandleVErrAndExitCode(queryist cli.Queryist, verr errhand.VerboseError, usage cli.UsagePrinter) int {
+	if verr != nil {
+		if msg := verr.Verbose(); strings.TrimSpace(msg) != "" {
+			if _, ok := queryist.(*engine.SqlEngine); !ok {
+				// We are in a context where we are attempting to connect to a remote database. These errors
+				// are unstructured, so we add some additional context around them.
+				tmpMsg := `You've encountered a new behavior in dolt which is not fully documented yet.
+A local dolt server is using your dolt data directory, and in an attempt to service your request, we are attempting to 
+connect to it. That has failed. You should stop the server, or reach out to @macneale on https://discord.gg/gqr7K4VNKe
+for help.`
+				cli.PrintErrln(tmpMsg)
+				msg = fmt.Sprintf("A local server is running, and dolt is failing to connect. Error connecting to remote database: \"%s\".\n", msg)
+			}
+			cli.PrintErrln(msg)
+		}
+
+		if verr.ShouldPrintUsage() {
+			usage()
+		}
+
+		return 1
+	}
+
+	return 0
+}
+
 // handleLegacyArguments is a temporary function to parse args, and print a error and explanation when the old form is provided.
 func (cmd SqlCmd) handleLegacyArguments(ap *argparser.ArgParser, commandStr string, args []string) (*argparser.ArgParseResults, error) {
 
@@ -326,19 +356,19 @@ func (cmd SqlCmd) handleLegacyArguments(ap *argparser.ArgParser, commandStr stri
 
 func listSavedQueries(ctx *sql.Context, qryist cli.Queryist, dEnv *env.DoltEnv, format engine.PrintResultFormat, usage cli.UsagePrinter) int {
 	if !dEnv.Valid() {
-		return HandleVErrAndExitCode(errhand.BuildDError("error: --%s must be used in a dolt database directory.", listSavedFlag).Build(), usage)
+		return sqlHandleVErrAndExitCode(qryist, errhand.BuildDError("error: --%s must be used in a dolt database directory.", listSavedFlag).Build(), usage)
 	}
 
 	workingRoot, err := dEnv.WorkingRoot(ctx)
 	if err != nil {
-		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+		return sqlHandleVErrAndExitCode(qryist, errhand.VerboseErrorFromError(err), usage)
 	}
 
 	hasQC, err := workingRoot.HasTable(ctx, doltdb.DoltQueryCatalogTableName)
 
 	if err != nil {
 		verr := errhand.BuildDError("error: Failed to read from repository.").AddCause(err).Build()
-		return HandleVErrAndExitCode(verr, usage)
+		return sqlHandleVErrAndExitCode(qryist, verr, usage)
 	}
 
 	if !hasQC {
@@ -346,27 +376,27 @@ func listSavedQueries(ctx *sql.Context, qryist cli.Queryist, dEnv *env.DoltEnv,
 	}
 
 	query := "SELECT * FROM " + doltdb.DoltQueryCatalogTableName
-	return HandleVErrAndExitCode(execQuery(ctx, qryist, query, format), usage)
+	return sqlHandleVErrAndExitCode(qryist, execQuery(ctx, qryist, query, format), usage)
 }
 
 func executeSavedQuery(ctx *sql.Context, qryist cli.Queryist, dEnv *env.DoltEnv, savedQueryName string, format engine.PrintResultFormat, usage cli.UsagePrinter) int {
 	if !dEnv.Valid() {
-		return HandleVErrAndExitCode(errhand.BuildDError("error: --%s must be used in a dolt database directory.", executeFlag).Build(), usage)
+		return sqlHandleVErrAndExitCode(qryist, errhand.BuildDError("error: --%s must be used in a dolt database directory.", executeFlag).Build(), usage)
 	}
 
 	workingRoot, err := dEnv.WorkingRoot(ctx)
 	if err != nil {
-		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+		return sqlHandleVErrAndExitCode(qryist, errhand.VerboseErrorFromError(err), usage)
 	}
 
 	sq, err := dtables.RetrieveFromQueryCatalog(ctx, workingRoot, savedQueryName)
 
 	if err != nil {
-		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+		return sqlHandleVErrAndExitCode(qryist, errhand.VerboseErrorFromError(err), usage)
 	}
 
 	cli.PrintErrf("Executing saved query '%s':\n%s\n", savedQueryName, sq.Query)
-	return HandleVErrAndExitCode(execQuery(ctx, qryist, sq.Query, format), usage)
+	return sqlHandleVErrAndExitCode(qryist, execQuery(ctx, qryist, sq.Query, format), usage)
 }
 
 func queryMode(
@@ -388,19 +418,19 @@ func queryMode(
 		se, ok := qryist.(*engine.SqlEngine)
 		if !ok {
 			misuse := fmt.Errorf("Using batch with non-local access pattern. Stop server if it is running")
-			return HandleVErrAndExitCode(errhand.VerboseErrorFromError(misuse), usage)
+			return sqlHandleVErrAndExitCode(se, errhand.VerboseErrorFromError(misuse), usage)
 		}
 
 		batchInput := strings.NewReader(query)
 		verr := execBatch(ctx, se, batchInput, continueOnError, format)
 		if verr != nil {
-			return HandleVErrAndExitCode(verr, usage)
+			return sqlHandleVErrAndExitCode(qryist, verr, usage)
 		}
 	} else {
 		input := strings.NewReader(query)
 		err := execMultiStatements(ctx, qryist, input, continueOnError, format)
 		if err != nil {
-			return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+			return sqlHandleVErrAndExitCode(qryist, errhand.VerboseErrorFromError(err), usage)
 		}
 	}
 
@@ -409,58 +439,35 @@ func queryMode(
 
 func execSaveQuery(ctx *sql.Context, dEnv *env.DoltEnv, qryist cli.Queryist, apr *argparser.ArgParseResults, query string, format engine.PrintResultFormat, usage cli.UsagePrinter) int {
 	if !dEnv.Valid() {
-		return HandleVErrAndExitCode(errhand.BuildDError("error: --%s must be used in a dolt database directory.", saveFlag).Build(), usage)
+		return sqlHandleVErrAndExitCode(qryist, errhand.BuildDError("error: --%s must be used in a dolt database directory.", saveFlag).Build(), usage)
 	}
 
 	saveName := apr.GetValueOrDefault(saveFlag, "")
 
 	verr := execQuery(ctx, qryist, query, format)
 	if verr != nil {
-		return HandleVErrAndExitCode(verr, usage)
+		return sqlHandleVErrAndExitCode(qryist, verr, usage)
 	}
 
 	workingRoot, err := dEnv.WorkingRoot(ctx)
 	if err != nil {
-		return HandleVErrAndExitCode(errhand.BuildDError("error: failed to get working root").AddCause(err).Build(), usage)
+		return sqlHandleVErrAndExitCode(qryist, errhand.BuildDError("error: failed to get working root").AddCause(err).Build(), usage)
 	}
 
 	saveMessage := apr.GetValueOrDefault(messageFlag, "")
 	newRoot, verr := saveQuery(ctx, workingRoot, query, saveName, saveMessage)
 	if verr != nil {
-		return HandleVErrAndExitCode(verr, usage)
+		return sqlHandleVErrAndExitCode(qryist, verr, usage)
 	}
 
 	err = dEnv.UpdateWorkingRoot(ctx, newRoot)
 	if err != nil {
-		return HandleVErrAndExitCode(errhand.BuildDError("error: failed to update working root").AddCause(err).Build(), usage)
+		return sqlHandleVErrAndExitCode(qryist, errhand.BuildDError("error: failed to update working root").AddCause(err).Build(), usage)
 	}
 
 	return 0
 }
 
-// getMultiRepoEnv returns an appropriate MultiRepoEnv for this invocation of the command
-func getMultiRepoEnv(ctx context.Context, workingDir string, dEnv *env.DoltEnv) (mrEnv *env.MultiRepoEnv, resolvedDir string, verr errhand.VerboseError) {
-	var err error
-	fs := dEnv.FS
-	if len(workingDir) > 0 {
-		fs, err = fs.WithWorkingDir(workingDir)
-	}
-	if err != nil {
-		return nil, "", errhand.VerboseErrorFromError(err)
-	}
-	resolvedDir, err = fs.Abs("")
-	if err != nil {
-		return nil, "", errhand.VerboseErrorFromError(err)
-	}
-
-	mrEnv, err = env.MultiEnvForDirectory(ctx, dEnv.Config.WriteableConfig(), fs, dEnv.Version, dEnv.IgnoreLockFile, dEnv)
-	if err != nil {
-		return nil, "", errhand.VerboseErrorFromError(err)
-	}
-
-	return mrEnv, resolvedDir, nil
-}
-
 func execBatch(
 	sqlCtx *sql.Context,
 	se *engine.SqlEngine,
diff --git a/go/cmd/dolt/commands/sqlserver/sqlclient.go b/go/cmd/dolt/commands/sqlserver/sqlclient.go
index af55ab7a0bc..57a109e50c9 100644
--- a/go/cmd/dolt/commands/sqlserver/sqlclient.go
+++ b/go/cmd/dolt/commands/sqlserver/sqlclient.go
@@ -38,13 +38,13 @@ import (
 	"github.com/dolthub/dolt/go/cmd/dolt/commands/engine"
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 	"github.com/dolthub/dolt/go/libraries/utils/argparser"
+	"github.com/dolthub/dolt/go/libraries/utils/filesys"
 	"github.com/dolthub/dolt/go/libraries/utils/iohelp"
 )
 
 const (
 	sqlClientDualFlag     = "dual"
 	SqlClientQueryFlag    = "query"
-	SqlClientUseDbFlag    = "use-db"
 	sqlClientResultFormat = "result-format"
 )
 
@@ -85,7 +85,7 @@ func (cmd SqlClientCmd) ArgParser() *argparser.ArgParser {
 	ap := SqlServerCmd{}.ArgParserWithName(cmd.Name())
 	ap.SupportsFlag(sqlClientDualFlag, "d", "Causes this command to spawn a dolt server that is automatically connected to.")
 	ap.SupportsString(SqlClientQueryFlag, "q", "string", "Sends the given query to the server and immediately exits.")
-	ap.SupportsString(SqlClientUseDbFlag, "", "db_name", fmt.Sprintf("Selects the given database before executing a query. "+
+	ap.SupportsString(commands.UseDbFlag, "", "db_name", fmt.Sprintf("Selects the given database before executing a query. "+
 		"By default, uses the current folder's name. Must be used with the --%s flag.", SqlClientQueryFlag))
 	ap.SupportsString(sqlClientResultFormat, "", "format", fmt.Sprintf("Returns the results in the given format. Must be used with the --%s flag.", SqlClientQueryFlag))
 	return ap
@@ -127,8 +127,8 @@ func (cmd SqlClientCmd) Exec(ctx context.Context, commandStr string, args []stri
 			cli.PrintErrln(color.RedString(fmt.Sprintf("--%s flag may not be used with --%s", sqlClientDualFlag, SqlClientQueryFlag)))
 			return 1
 		}
-		if apr.Contains(SqlClientUseDbFlag) {
-			cli.PrintErrln(color.RedString(fmt.Sprintf("--%s flag may not be used with --%s", sqlClientDualFlag, SqlClientUseDbFlag)))
+		if apr.Contains(commands.UseDbFlag) {
+			cli.PrintErrln(color.RedString(fmt.Sprintf("--%s flag may not be used with --%s", sqlClientDualFlag, commands.UseDbFlag)))
 			return 1
 		}
 		if apr.Contains(sqlClientResultFormat) {
@@ -136,7 +136,7 @@ func (cmd SqlClientCmd) Exec(ctx context.Context, commandStr string, args []stri
 			return 1
 		}
 
-		serverConfig, err = GetServerConfig(dEnv, apr)
+		serverConfig, err = GetServerConfig(dEnv.FS, apr)
 		if err != nil {
 			cli.PrintErrln(color.RedString("Bad Configuration"))
 			cli.PrintErrln(err.Error())
@@ -159,7 +159,7 @@ func (cmd SqlClientCmd) Exec(ctx context.Context, commandStr string, args []stri
 			return 1
 		}
 	} else {
-		serverConfig, err = GetServerConfig(dEnv, apr)
+		serverConfig, err = GetServerConfig(dEnv.FS, apr)
 		if err != nil {
 			cli.PrintErrln(color.RedString("Bad Configuration"))
 			cli.PrintErrln(err.Error())
@@ -168,13 +168,13 @@ func (cmd SqlClientCmd) Exec(ctx context.Context, commandStr string, args []stri
 	}
 
 	query, hasQuery := apr.GetValue(SqlClientQueryFlag)
-	dbToUse, hasUseDb := apr.GetValue(SqlClientUseDbFlag)
+	dbToUse, hasUseDb := apr.GetValue(commands.UseDbFlag)
 	resultFormat, hasResultFormat := apr.GetValue(sqlClientResultFormat)
 	if !hasQuery && hasUseDb {
-		cli.PrintErrln(color.RedString(fmt.Sprintf("--%s may only be used with --%s", SqlClientUseDbFlag, SqlClientQueryFlag)))
+		cli.PrintErrln(color.RedString(fmt.Sprintf("--%s may only be used with --%s", commands.UseDbFlag, SqlClientQueryFlag)))
 		return 1
 	} else if !hasQuery && hasResultFormat {
-		cli.PrintErrln(color.RedString(fmt.Sprintf("--%s may only be used with --%s", SqlClientUseDbFlag, sqlClientResultFormat)))
+		cli.PrintErrln(color.RedString(fmt.Sprintf("--%s may only be used with --%s", commands.UseDbFlag, sqlClientResultFormat)))
 		return 1
 	}
 	if !hasUseDb && hasQuery {
@@ -453,3 +453,52 @@ func secondsSince(start time.Time, end time.Time) float64 {
 	timeDisplay := float64(seconds) + float64(milliRemainder)*.001
 	return timeDisplay
 }
+
+type ConnectionQueryist struct {
+	connection *dbr.Connection
+}
+
+func (c ConnectionQueryist) Query(ctx *sql.Context, query string) (sql.Schema, sql.RowIter, error) {
+	rows, err := c.connection.QueryContext(ctx, query)
+	if err != nil {
+		return nil, nil, err
+	}
+	rowIter, err := NewMysqlRowWrapper(rows)
+	if err != nil {
+		return nil, nil, err
+	}
+	return rowIter.Schema(), rowIter, nil
+}
+
+var _ cli.Queryist = ConnectionQueryist{}
+
+// BuildConnectionStringQueryist returns a Queryist that connects to the server specified by the given server config. Presence in this
+// module isn't ideal, but it's the only way to get the server config into the queryist.
+func BuildConnectionStringQueryist(ctx context.Context, cwdFS filesys.Filesys, apr *argparser.ArgParseResults, port int, database string) (cli.LateBindQueryist, error) {
+	serverConfig, err := GetServerConfig(cwdFS, apr)
+	if err != nil {
+		return nil, err
+	}
+
+	parsedMySQLConfig, err := mysqlDriver.ParseDSN(ConnectionString(serverConfig, database))
+	if err != nil {
+		return nil, err
+	}
+
+	parsedMySQLConfig.Addr = fmt.Sprintf("localhost:%d", port)
+
+	mysqlConnector, err := mysqlDriver.NewConnector(parsedMySQLConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	conn := &dbr.Connection{DB: mysql.OpenDB(mysqlConnector), EventReceiver: nil, Dialect: dialect.MySQL}
+
+	queryist := ConnectionQueryist{connection: conn}
+
+	var lateBind cli.LateBindQueryist = func(ctx context.Context) (cli.Queryist, *sql.Context, func(), error) {
+		return queryist, sql.NewContext(ctx), func() { conn.Conn(ctx) }, nil
+	}
+
+	return lateBind, nil
+}
diff --git a/go/cmd/dolt/commands/sqlserver/sqlserver.go b/go/cmd/dolt/commands/sqlserver/sqlserver.go
index a9cf18d8d7f..e064fa6dbd2 100644
--- a/go/cmd/dolt/commands/sqlserver/sqlserver.go
+++ b/go/cmd/dolt/commands/sqlserver/sqlserver.go
@@ -207,7 +207,7 @@ func startServer(ctx context.Context, versionStr, commandStr string, args []stri
 	if err := validateSqlServerArgs(apr); err != nil {
 		return 1
 	}
-	serverConfig, err := GetServerConfig(dEnv, apr)
+	serverConfig, err := GetServerConfig(dEnv.FS, apr)
 	if err != nil {
 		if serverController != nil {
 			serverController.StopServer()
@@ -246,16 +246,16 @@ func startServer(ctx context.Context, versionStr, commandStr string, args []stri
 
 // GetServerConfig returns ServerConfig that is set either from yaml file if given, if not it is set with values defined
 // on command line. Server config variables not defined are set to default values.
-func GetServerConfig(dEnv *env.DoltEnv, apr *argparser.ArgParseResults) (ServerConfig, error) {
+func GetServerConfig(cwdFS filesys.Filesys, apr *argparser.ArgParseResults) (ServerConfig, error) {
 	var yamlCfg YAMLConfig
 	if cfgFile, ok := apr.GetValue(configFileFlag); ok {
-		cfg, err := getYAMLServerConfig(dEnv.FS, cfgFile)
+		cfg, err := getYAMLServerConfig(cwdFS, cfgFile)
 		if err != nil {
 			return nil, err
 		}
 		yamlCfg = cfg.(YAMLConfig)
 	} else {
-		return getCommandLineServerConfig(dEnv, apr)
+		return getCommandLineServerConfig(apr)
 	}
 
 	// if command line user argument was given, replace yaml's user and password
@@ -350,7 +350,7 @@ func SetupDoltConfig(dEnv *env.DoltEnv, apr *argparser.ArgParseResults, config S
 
 // getCommandLineServerConfig sets server config variables and persisted global variables with values defined on command line.
 // If not defined, it sets variables to default values.
-func getCommandLineServerConfig(dEnv *env.DoltEnv, apr *argparser.ArgParseResults) (ServerConfig, error) {
+func getCommandLineServerConfig(apr *argparser.ArgParseResults) (ServerConfig, error) {
 	serverConfig := DefaultServerConfig()
 
 	if sock, ok := apr.GetValue(socketFlag); ok {
diff --git a/go/cmd/dolt/commands/utils.go b/go/cmd/dolt/commands/utils.go
index 74f641c6ccd..a22c8bf5a51 100644
--- a/go/cmd/dolt/commands/utils.go
+++ b/go/cmd/dolt/commands/utils.go
@@ -28,6 +28,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 	"github.com/dolthub/dolt/go/libraries/doltcore/env/actions"
 	"github.com/dolthub/dolt/go/libraries/utils/argparser"
+	"github.com/dolthub/dolt/go/libraries/utils/filesys"
 )
 
 var fwtStageName = "fwt"
@@ -78,17 +79,22 @@ func MaybeGetCommitWithVErr(dEnv *env.DoltEnv, maybeCommit string) (*doltdb.Comm
 
 // NewArgFreeCliContext creates a new CliContext instance with no arguments using a local SqlEngine. This is useful for testing primarily
 func NewArgFreeCliContext(ctx context.Context, dEnv *env.DoltEnv) (cli.CliContext, errhand.VerboseError) {
-	lateBind, err := BuildSqlEngineQueryist(ctx, dEnv, argparser.NewEmptyResults())
+	mrEnv, err := env.MultiEnvForSingleEnv(ctx, dEnv)
 	if err != nil {
-		return nil, err
+		return nil, errhand.VerboseErrorFromError(err)
+	}
+
+	lateBind, verr := BuildSqlEngineQueryist(ctx, dEnv.FS, mrEnv, argparser.NewEmptyResults())
+	if err != nil {
+		return nil, verr
 	}
 	return cli.NewCliContext(argparser.NewEmptyResults(), lateBind)
 }
 
 // BuildSqlEngineQueryist Utility function to build a local SQLEngine for use interacting with data on disk using
-// SQL queries. ctx and dEnv must be non-nil. apr can be nil.
-func BuildSqlEngineQueryist(ctx context.Context, dEnv *env.DoltEnv, apr *argparser.ArgParseResults) (cli.LateBindQueryist, errhand.VerboseError) {
-	if ctx == nil || dEnv == nil || apr == nil {
+// SQL queries. ctx, cwdFS, mrEnv, and apr must all be non-nil.
+func BuildSqlEngineQueryist(ctx context.Context, cwdFS filesys.Filesys, mrEnv *env.MultiRepoEnv, apr *argparser.ArgParseResults) (cli.LateBindQueryist, errhand.VerboseError) {
+	if ctx == nil || cwdFS == nil || mrEnv == nil || apr == nil {
 		errhand.VerboseErrorFromError(fmt.Errorf("Invariant violated. Nil argument provided to BuildSqlEngineQueryist"))
 	}
 
@@ -98,44 +104,42 @@ func BuildSqlEngineQueryist(ctx context.Context, dEnv *env.DoltEnv, apr *argpars
 		username = user
 	}
 
-	// data-dir args come either from the global args or the subcommand args.  We need to check both.
-	var dataDir string
-	dataDirGiven := false
-	if dataDirPath, ok := apr.GetValue(DataDirFlag); ok {
-		dataDir = dataDirPath
-		dataDirGiven = true
-	}
-
-	mrEnv, dataDir, verr := getMultiRepoEnv(ctx, dataDir, dEnv)
-	if verr != nil {
-		return nil, verr
+	// We want to know if the user provided us the data-dir flag, but we want to use the abs value used to
+	// create the DoltEnv. This is a little messy.
+	dataDir, dataDirGiven := apr.GetValue(DataDirFlag)
+	dataDir, err := cwdFS.Abs(dataDir)
+	if err != nil {
+		return nil, errhand.VerboseErrorFromError(err)
 	}
 
 	// need to return cfgdirpath and error
 	var cfgDirPath string
 	cfgDir, cfgDirSpecified := apr.GetValue(CfgDirFlag)
 	if cfgDirSpecified {
-		cfgDirPath = cfgDir
+		cfgDirPath, err = cwdFS.Abs(cfgDir)
+		if err != nil {
+			return nil, errhand.VerboseErrorFromError(err)
+		}
 	} else if dataDirGiven {
 		cfgDirPath = filepath.Join(dataDir, DefaultCfgDirName)
 	} else {
-		// Look in parent directory for doltcfg
+		// Look in CWD parent directory for doltcfg
 		parentDirCfg := filepath.Join("..", DefaultCfgDirName)
-		parentExists, isDir := dEnv.FS.Exists(parentDirCfg)
+		parentExists, isDir := cwdFS.Exists(parentDirCfg)
 		parentDirExists := parentExists && isDir
 
-		// Look in data directory (which is necessarily current directory) for doltcfg
-		currDirCfg := filepath.Join(dataDir, DefaultCfgDirName)
-		currExists, isDir := dEnv.FS.Exists(currDirCfg)
-		currDirExists := currExists && isDir
+		// Look in data directory for doltcfg
+		dataDirCfg := filepath.Join(dataDir, DefaultCfgDirName)
+		dataDirCfgExists, isDir := cwdFS.Exists(dataDirCfg)
+		currDirExists := dataDirCfgExists && isDir
 
-		// Error if both current and parent exist
+		// Error if both CWD/../.doltfcfg and dataDir/.doltcfg exist because it's unclear which to use.
 		if currDirExists && parentDirExists {
-			p1, err := dEnv.FS.Abs(cfgDirPath)
+			p1, err := cwdFS.Abs(cfgDirPath)
 			if err != nil {
 				return nil, errhand.VerboseErrorFromError(err)
 			}
-			p2, err := dEnv.FS.Abs(parentDirCfg)
+			p2, err := cwdFS.Abs(parentDirCfg)
 			if err != nil {
 				return nil, errhand.VerboseErrorFromError(err)
 			}
@@ -146,15 +150,19 @@ func BuildSqlEngineQueryist(ctx context.Context, dEnv *env.DoltEnv, apr *argpars
 		if parentDirExists {
 			cfgDirPath = parentDirCfg
 		} else {
-			cfgDirPath = currDirCfg
+			cfgDirPath = dataDirCfg
 		}
 	}
 
-	var err error
 	// If no privilege filepath specified, default to doltcfg directory
 	privsFp, hasPrivsFp := apr.GetValue(PrivsFilePathFlag)
 	if !hasPrivsFp {
-		privsFp, err = dEnv.FS.Abs(filepath.Join(cfgDirPath, DefaultPrivsName))
+		privsFp, err = cwdFS.Abs(filepath.Join(cfgDirPath, DefaultPrivsName))
+		if err != nil {
+			return nil, errhand.VerboseErrorFromError(err)
+		}
+	} else {
+		privsFp, err = cwdFS.Abs(privsFp)
 		if err != nil {
 			return nil, errhand.VerboseErrorFromError(err)
 		}
@@ -163,13 +171,25 @@ func BuildSqlEngineQueryist(ctx context.Context, dEnv *env.DoltEnv, apr *argpars
 	// If no branch control file path is specified, default to doltcfg directory
 	branchControlFilePath, hasBCFilePath := apr.GetValue(BranchCtrlPathFlag)
 	if !hasBCFilePath {
-		branchControlFilePath, err = dEnv.FS.Abs(filepath.Join(cfgDirPath, DefaultBranchCtrlName))
+		branchControlFilePath, err = cwdFS.Abs(filepath.Join(cfgDirPath, DefaultBranchCtrlName))
 		if err != nil {
 			return nil, errhand.VerboseErrorFromError(err)
 		}
+	} else {
+		branchControlFilePath, err = cwdFS.Abs(branchControlFilePath)
+		if err != nil {
+			return nil, errhand.VerboseErrorFromError(err)
+		}
+	}
+
+	// Whether we're running in shell mode or some other mode, sql commands from the command line always have a current
+	// database set when you begin using them.
+	database, hasDB := apr.GetValue(UseDbFlag)
+	if !hasDB {
+		database = mrEnv.GetFirstDatabase()
 	}
 
-	binder, err := newLateBindingEngine(ctx, apr, cfgDirPath, privsFp, branchControlFilePath, username, mrEnv)
+	binder, err := newLateBindingEngine(cfgDirPath, privsFp, branchControlFilePath, username, database, mrEnv)
 	if err != nil {
 		return nil, errhand.VerboseErrorFromError(err)
 	}
@@ -178,12 +198,11 @@ func BuildSqlEngineQueryist(ctx context.Context, dEnv *env.DoltEnv, apr *argpars
 }
 
 func newLateBindingEngine(
-	ctx context.Context,
-	apr *argparser.ArgParseResults,
 	cfgDirPath string,
 	privsFp string,
 	branchControlFilePath string,
 	username string,
+	database string,
 	mrEnv *env.MultiRepoEnv,
 ) (cli.LateBindQueryist, error) {
 
@@ -212,7 +231,7 @@ func newLateBindingEngine(
 
 		// Whether we're running in shell mode or some other mode, sql commands from the command line always have a current
 		// database set when you begin using them.
-		sqlCtx.SetCurrentDatabase(mrEnv.GetFirstDatabase())
+		sqlCtx.SetCurrentDatabase(database)
 
 		// Add specified user as new superuser, if it doesn't already exist
 		if user := se.GetUnderlyingEngine().Analyzer.Catalog.MySQLDb.GetUser(config.ServerUser, config.ServerHost, false); user == nil {
diff --git a/go/cmd/dolt/dolt.go b/go/cmd/dolt/dolt.go
index 45f5807fa7f..d6cefad0eb1 100644
--- a/go/cmd/dolt/dolt.go
+++ b/go/cmd/dolt/dolt.go
@@ -140,6 +140,7 @@ const stdOutFlag = "--stdout"
 const stdErrFlag = "--stderr"
 const stdOutAndErrFlag = "--out-and-err"
 const ignoreLocksFlag = "--ignore-lock-file"
+const verboseEngineSetupFlag = "--verbose-engine-setup"
 
 const cpuProf = "cpu"
 const memProf = "mem"
@@ -168,6 +169,7 @@ func runMain() int {
 
 	csMetrics := false
 	ignoreLockFile := false
+	verboseEngineSetup := false
 	if len(args) > 0 {
 		var doneDebugFlags bool
 		for !doneDebugFlags && len(args) > 0 {
@@ -324,6 +326,9 @@ func runMain() int {
 
 				args = args[2:]
 
+			case verboseEngineSetupFlag:
+				verboseEngineSetup = true
+				args = args[1:]
 			default:
 				doneDebugFlags = true
 			}
@@ -342,7 +347,42 @@ func runMain() int {
 		return exit
 	}
 
-	dEnv := env.Load(ctx, env.GetCurrentUserHomeDir, filesys.LocalFS, doltdb.LocalDirDoltDB, Version)
+	globalArgs, args, initCliContext, printUsage, err := splitArgsOnSubCommand(args)
+	_, usage := cli.HelpAndUsagePrinters(cli.CommandDocsForCommandString("dolt", doc, globalArgParser))
+	if printUsage {
+		doltCommand.PrintUsage("dolt")
+
+		specialMsg := `
+Dolt subcommands are in transition to using the flags listed below as global flags.
+The sql subcommand is currently the only command that uses these flags. All other commands will ignore them.
+`
+		cli.Println(specialMsg)
+		usage()
+
+		return 0
+	}
+	if err != nil {
+		cli.PrintErrln(color.RedString("Failure to parse arguments: %v", err))
+		return 1
+	}
+	apr := cli.ParseArgsOrDie(globalArgParser, globalArgs, usage)
+
+	var fs filesys.Filesys
+	fs = filesys.LocalFS
+	dataDir, hasDataDir := apr.GetValue(commands.DataDirFlag)
+	if hasDataDir {
+		// If a relative path was provided, this ensures we have an absolute path everywhere.
+		dataDir, err = fs.Abs(dataDir)
+		if err != nil {
+			cli.PrintErrln(color.RedString("Failed to get absolute path for %s: %v", dataDir, err))
+			return 1
+		}
+		if ok, dir := fs.Exists(dataDir); !ok || !dir {
+			cli.Println(color.RedString("Provided data directory does not exist: %s", dataDir))
+			return 1
+		}
+	}
+	dEnv := env.Load(ctx, env.GetCurrentUserHomeDir, fs, doltdb.LocalDirDoltDB, Version)
 	dEnv.IgnoreLockFile = ignoreLockFile
 
 	root, err := env.GetCurrentUserHomeDir()
@@ -409,7 +449,12 @@ func runMain() int {
 	// variables like `${db_name}_default_branch` (maybe these should not be
 	// part of Dolt config in the first place!).
 
-	mrEnv, err := env.MultiEnvForDirectory(ctx, dEnv.Config.WriteableConfig(), dEnv.FS, dEnv.Version, dEnv.IgnoreLockFile, dEnv)
+	dataDirFS, err := dEnv.FS.WithWorkingDir(dataDir)
+	if err != nil {
+		cli.PrintErrln(color.RedString("Failed to set the data directory. %v", err))
+		return 1
+	}
+	mrEnv, err := env.MultiEnvForDirectory(ctx, dEnv.Config.WriteableConfig(), dataDirFS, dEnv.Version, dEnv.IgnoreLockFile, dEnv)
 	if err != nil {
 		cli.PrintErrln("failed to load database names")
 		return 1
@@ -424,31 +469,9 @@ func runMain() int {
 		cli.Printf("error: failed to load persisted global variables: %s\n", err.Error())
 	}
 
-	globalArgs, args, initCliContext, printUsage, err := splitArgsOnSubCommand(args)
-	if printUsage {
-		doltCommand.PrintUsage("dolt")
-		_, usage := cli.HelpAndUsagePrinters(cli.CommandDocsForCommandString("dolt", doc, globalArgParser))
-
-		specialMsg := `
-Dolt subcommands are in transition to using the flags listed below as global flags.
-The sql subcommand is currently the only command that uses these flags. All other commands will ignore them.
-`
-		cli.Println(specialMsg)
-		usage()
-
-		return 0
-	}
-	if err != nil {
-		cli.PrintErrln(color.RedString("Failure to parse arguments: %v", err))
-		return 1
-	}
-
 	var cliCtx cli.CliContext = nil
 	if initCliContext {
-		_, usage := cli.HelpAndUsagePrinters(cli.CommandDocsForCommandString("dolt", doc, globalArgParser))
-		apr := cli.ParseArgsOrDie(globalArgParser, globalArgs, usage)
-
-		lateBind, err := commands.BuildSqlEngineQueryist(ctx, dEnv, apr)
+		lateBind, err := buildLateBinder(ctx, dEnv.FS, mrEnv, apr, verboseEngineSetup)
 		if err != nil {
 			cli.PrintErrln(color.RedString("Failure to Load SQL Engine: %v", err))
 			return 1
@@ -459,7 +482,6 @@ The sql subcommand is currently the only command that uses these flags. All othe
 			cli.PrintErrln(color.RedString("Unexpected Error: %v", err))
 			return 1
 		}
-
 	}
 
 	ctx, stop := context.WithCancel(ctx)
@@ -482,6 +504,44 @@ The sql subcommand is currently the only command that uses these flags. All othe
 	return res
 }
 
+func buildLateBinder(ctx context.Context, cwdFS filesys.Filesys, mrEnv *env.MultiRepoEnv, apr *argparser.ArgParseResults, verbose bool) (cli.LateBindQueryist, error) {
+
+	var targetEnv *env.DoltEnv = nil
+
+	useDb, hasUseDb := apr.GetValue(commands.UseDbFlag)
+	if hasUseDb {
+		targetEnv = mrEnv.GetEnv(useDb)
+		if targetEnv == nil {
+			return nil, fmt.Errorf("The provided --use-db %s does not exist or is not a directory.", useDb)
+		}
+	} else {
+		useDb = mrEnv.GetFirstDatabase()
+	}
+
+	if targetEnv == nil && useDb != "" {
+		targetEnv = mrEnv.GetEnv(useDb)
+	}
+
+	// nil targetEnv will happen if the user ran a command in an empty directory - which we support in some cases.
+	if targetEnv != nil {
+		isLocked, lock, err := targetEnv.GetLock()
+		if err != nil {
+			return nil, err
+		}
+		if isLocked {
+			if verbose {
+				cli.Println("verbose: starting remote mode")
+			}
+			return sqlserver.BuildConnectionStringQueryist(ctx, cwdFS, apr, lock.Port, useDb)
+		}
+	}
+
+	if verbose {
+		cli.Println("verbose: starting local mode")
+	}
+	return commands.BuildSqlEngineQueryist(ctx, cwdFS, mrEnv, apr)
+}
+
 // splitArgsOnSubCommand splits the args into two slices, the first containing all args before the first subcommand,
 // and the second containing all args after the first subcommand. The second slice will start with the subcommand name.
 func splitArgsOnSubCommand(args []string) (globalArgs, subArgs []string, initCliContext, printUsages bool, err error) {
@@ -568,12 +628,12 @@ func interceptSendMetrics(ctx context.Context, args []string) (bool, int) {
 func buildGlobalArgs() *argparser.ArgParser {
 	ap := argparser.NewArgParserWithVariableArgs("dolt")
 
-	// Pulling this argument forward first to pave the way. Others will follow.
 	ap.SupportsString(commands.UserFlag, "u", "user", fmt.Sprintf("Defines the local superuser (defaults to `%v`). If the specified user exists, will take on permissions of that user.", commands.DefaultUser))
 	ap.SupportsString(commands.DataDirFlag, "", "directory", "Defines a directory whose subdirectories should all be dolt data repositories accessible as independent databases within. Defaults to the current directory.")
 	ap.SupportsString(commands.CfgDirFlag, "", "directory", "Defines a directory that contains configuration files for dolt. Defaults to `$data-dir/.doltcfg`. Will only be created if there is a change to configuration settings.")
 	ap.SupportsString(commands.PrivsFilePathFlag, "", "privilege file", "Path to a file to load and store users and grants. Defaults to `$doltcfg-dir/privileges.db`. Will only be created if there is a change to privileges.")
 	ap.SupportsString(commands.BranchCtrlPathFlag, "", "branch control file", "Path to a file to load and store branch control permissions. Defaults to `$doltcfg-dir/branch_control.db`. Will only be created if there is a change to branch control permissions.")
+	ap.SupportsString(commands.UseDbFlag, "", "database", "The name of the database to use when executing SQL queries. Defaults the database of the root directory, if it exists, and the first alphabetically if not.")
 
 	return ap
 }
diff --git a/go/libraries/doltcore/env/environment.go b/go/libraries/doltcore/env/environment.go
index 5a420e749e0..dc74362dfc7 100644
--- a/go/libraries/doltcore/env/environment.go
+++ b/go/libraries/doltcore/env/environment.go
@@ -1149,7 +1149,18 @@ func (dEnv *DoltEnv) IsLocked() bool {
 	if dEnv.IgnoreLockFile {
 		return false
 	}
-	return FsIsLocked(dEnv.FS)
+
+	ans, _, _ := fsIsLocked(dEnv.FS)
+	return ans
+}
+
+// GetLock returns the lockfile for this database or nil if the database is not locked
+func (dEnv *DoltEnv) GetLock() (bool, *DBLock, error) {
+	if dEnv.IgnoreLockFile {
+		return false, nil, nil
+	}
+
+	return fsIsLocked(dEnv.FS)
 }
 
 // DBLock is a struct that contains the pid of the process that created the lockfile and the port that the server is running on
@@ -1259,24 +1270,27 @@ func WriteLockfile(fs filesys.Filesys, lock DBLock) error {
 
 // FsIsLocked returns true if a lockFile exists with the same pid as
 // any live process.
-func FsIsLocked(fs filesys.Filesys) bool {
+func fsIsLocked(fs filesys.Filesys) (bool, *DBLock, error) {
 	lockFile, _ := fs.Abs(filepath.Join(dbfactory.DoltDir, ServerLockFile))
 
 	ok, _ := fs.Exists(lockFile)
 	if !ok {
-		return false
+		return false, nil, nil
 	}
 
 	loadedLock, err := LoadDBLockFile(fs, lockFile)
 	if err != nil { // if there's any error assume that env is locked since the file exists
-		return true
+		return true, nil, err
 	}
 
 	// Check whether the pid that spawned the lock file is still running. Ignore it if not.
 	p, err := ps.FindProcess(loadedLock.Pid)
 	if err != nil {
-		return false
+		return false, nil, nil
 	}
 
-	return p != nil
+	if p != nil {
+		return true, loadedLock, nil
+	}
+	return false, nil, nil
 }
diff --git a/go/libraries/doltcore/env/multi_repo_env.go b/go/libraries/doltcore/env/multi_repo_env.go
index 02a84eba2c7..d01120b9af6 100644
--- a/go/libraries/doltcore/env/multi_repo_env.go
+++ b/go/libraries/doltcore/env/multi_repo_env.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sort"
 	"strings"
 	"unicode"
 
@@ -58,52 +59,61 @@ type MultiRepoEnv struct {
 	ignoreLockFile bool
 }
 
+// NewMultiEnv returns a new MultiRepoEnv instance dirived from a root DoltEnv instance.
+func MultiEnvForSingleEnv(ctx context.Context, env *DoltEnv) (*MultiRepoEnv, error) {
+	return MultiEnvForDirectory(ctx, env.Config.WriteableConfig(), env.FS, env.Version, env.IgnoreLockFile, env)
+}
+
 // MultiEnvForDirectory returns a MultiRepoEnv for the directory rooted at the file system given. The doltEnv from the
 // invoking context is included. If it's non-nil and valid, it will be included in the returned MultiRepoEnv, and will
 // be the first database in all iterations.
 func MultiEnvForDirectory(
 	ctx context.Context,
 	config config.ReadWriteConfig,
-	fs filesys.Filesys,
+	dataDirFS filesys.Filesys,
 	version string,
 	ignoreLockFile bool,
 	dEnv *DoltEnv,
 ) (*MultiRepoEnv, error) {
-	mrEnv := &MultiRepoEnv{
-		envs:           make([]NamedEnv, 0),
-		fs:             fs,
-		cfg:            config,
-		dialProvider:   NewGRPCDialProviderFromDoltEnv(dEnv),
-		ignoreLockFile: ignoreLockFile,
-	}
-
-	// Load current fs and put into mr env
-	var dbName string
-	if _, ok := fs.(*filesys.InMemFS); ok {
-		dbName = "dolt"
-	} else {
-		path, err := fs.Abs("")
+	// Load current dataDirFS and put into mr env
+	var dbName string = "dolt"
+	var newDEnv *DoltEnv = dEnv
+
+	// InMemFS is used only for testing.
+	// All other FS Types should get a newly created Environment which will serve as the primary env in the MultiRepoEnv
+	if _, ok := dataDirFS.(*filesys.InMemFS); !ok {
+		path, err := dataDirFS.Abs("")
 		if err != nil {
 			return nil, err
 		}
 		envName := getRepoRootDir(path, string(os.PathSeparator))
 		dbName = dirToDBName(envName)
+
+		newDEnv = Load(ctx, GetCurrentUserHomeDir, dataDirFS, doltdb.LocalDirDoltDB, version)
+	}
+
+	mrEnv := &MultiRepoEnv{
+		envs:           make([]NamedEnv, 0),
+		fs:             dataDirFS,
+		cfg:            config,
+		dialProvider:   NewGRPCDialProviderFromDoltEnv(newDEnv),
+		ignoreLockFile: ignoreLockFile,
 	}
 
 	envSet := map[string]*DoltEnv{}
-	if dEnv.Valid() {
-		envSet[dbName] = dEnv
+	if newDEnv.Valid() {
+		envSet[dbName] = newDEnv
 	}
 
 	// If there are other directories in the directory, try to load them as additional databases
-	fs.Iter(".", false, func(path string, size int64, isDir bool) (stop bool) {
+	dataDirFS.Iter(".", false, func(path string, size int64, isDir bool) (stop bool) {
 		if !isDir {
 			return false
 		}
 
 		dir := filepath.Base(path)
 
-		newFs, err := fs.WithWorkingDir(dir)
+		newFs, err := dataDirFS.WithWorkingDir(dir)
 		if err != nil {
 			return false
 		}
@@ -124,14 +134,19 @@ func MultiEnvForDirectory(
 	enforceSingleFormat(envSet)
 
 	// if the current directory database is in our set, add it first so it will be the current database
-	var ok bool
-	if dEnv, ok = envSet[dbName]; ok && dEnv.Valid() {
-		mrEnv.addEnv(dbName, dEnv)
+	if env, ok := envSet[dbName]; ok && env.Valid() {
+		mrEnv.addEnv(dbName, env)
 		delete(envSet, dbName)
 	}
 
-	for dbName, dEnv = range envSet {
-		mrEnv.addEnv(dbName, dEnv)
+	// get the keys from the envSet keys as a sorted list
+	sortedKeys := make([]string, 0, len(envSet))
+	for k := range envSet {
+		sortedKeys = append(sortedKeys, k)
+	}
+	sort.Strings(sortedKeys)
+	for _, dbName := range sortedKeys {
+		mrEnv.addEnv(dbName, envSet[dbName])
 	}
 
 	return mrEnv, nil
diff --git a/integration-tests/bats/replication-multidb.bats b/integration-tests/bats/replication-multidb.bats
index 5fcf8c0fc38..888c8a7dec9 100644
--- a/integration-tests/bats/replication-multidb.bats
+++ b/integration-tests/bats/replication-multidb.bats
@@ -249,6 +249,7 @@ call dolt_commit('-am', 'new table');
 SQL
 
     cd $TMPDIRS
+    mkdir -p "dbs2"
 
     # this is a hack: we have to change our persisted global server
     # vars for the sql command to work on the replica TODO: fix this
diff --git a/integration-tests/bats/sql-create-database.bats b/integration-tests/bats/sql-create-database.bats
index 88434bd5e19..444e971d23c 100644
--- a/integration-tests/bats/sql-create-database.bats
+++ b/integration-tests/bats/sql-create-database.bats
@@ -1,3 +1,4 @@
+#!/usr/bin/env bats
 load $BATS_TEST_DIRNAME/helper/common.bash
 
 setup() {
diff --git a/integration-tests/bats/sql-local-remote.bats b/integration-tests/bats/sql-local-remote.bats
new file mode 100644
index 00000000000..98461b36fc4
--- /dev/null
+++ b/integration-tests/bats/sql-local-remote.bats
@@ -0,0 +1,110 @@
+#! /usr/bin/env bats
+load $BATS_TEST_DIRNAME/helper/common.bash
+load $BATS_TEST_DIRNAME/helper/query-server-common.bash
+
+make_repo() {
+  mkdir "$1"
+  cd "$1"
+  dolt init
+  dolt sql -q "create table $1_tbl (id int)"
+  cd ..
+}
+
+setup() {
+    setup_no_dolt_init
+    make_repo defaultDB
+    make_repo altDB
+}
+
+teardown() {
+    stop_sql_server 1
+    teardown_common
+}
+
+@test "sql-local-remote: test switch between server/no server" {
+    start_sql_server defaultDB
+
+    run dolt --verbose-engine-setup --user dolt sql -q "show databases" 
+    [ "$status" -eq 0 ] || false
+    [[ "$output" =~ "starting remote mode" ]] || false
+    [[ "$output" =~ "defaultDB" ]] || false
+    [[ "$output" =~ "altDB" ]] || false
+
+    stop_sql_server 1
+
+    run dolt --verbose-engine-setup sql -q "show databases" 
+    [ "$status" -eq 0 ] || false
+    [[ "$output" =~ "starting local mode" ]] || false
+    [[ "$output" =~ "defaultDB" ]] || false
+    [[ "$output" =~ "altDB" ]] || false
+}
+
+@test "sql-local-remote: check --data-dir pointing to a server root can be used when in different directory." {
+    start_sql_server altDb
+    ROOT_DIR=$(pwd)
+
+    mkdir someplace_else
+    cd someplace_else
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR" --user dolt --use-db altDB sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting remote mode" ]] || false
+    [[ "$output" =~ "altDB_tbl" ]] || false
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR" --user dolt --use-db defaultDB sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting remote mode" ]] || false
+    [[ "$output" =~ "defaultDB_tbl" ]] || false
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR" --user dolt sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting remote mode" ]] || false
+    [[ "$output" =~ "altDB_tbl" ]] || false
+
+    stop_sql_server 1
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR" --user dolt --use-db altDB sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting local mode" ]] || false
+    [[ "$output" =~ "altDB_tbl" ]] || false
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR" --user dolt --use-db defaultDB sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting local mode" ]] || false
+    [[ "$output" =~ "defaultDB_tbl" ]] || false
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR" --user dolt sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting local mode" ]] || false
+    [[ "$output" =~ "altDB_tbl" ]] || false
+}
+
+
+@test "sql-local-remote: check --data-dir pointing to a database root can be used when in different directory." {
+    start_sql_server altDb
+    ROOT_DIR=$(pwd)
+
+    mkdir -p someplace_new/fun
+    cd someplace_new/fun
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR/altDB" --user dolt sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting remote mode" ]] || false
+    [[ "$output" =~ "altDB_tbl" ]] || false
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR/altDB" --user dolt --use-db defaultDB sql -q "show tables"
+    [ "$status" -eq 1 ]
+    [[ "$output" =~ "defaultDB does not exist" ]] || false
+
+    stop_sql_server 1
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR/altDB" --user dolt sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting local mode" ]] || false
+    [[ "$output" =~ "altDB_tbl" ]] || false
+
+    run dolt --verbose-engine-setup --data-dir="$ROOT_DIR/altDB" --user dolt --use-db defaultDB sql -q "show tables"
+    [ "$status" -eq 1 ]
+    [[ "$output" =~ "defaultDB does not exist" ]] || false
+}
+
diff --git a/integration-tests/bats/sql-server.bats b/integration-tests/bats/sql-server.bats
index d51257fa200..3e1823abf5f 100644
--- a/integration-tests/bats/sql-server.bats
+++ b/integration-tests/bats/sql-server.bats
@@ -74,7 +74,7 @@ EOF
 
     # verify that dolt_clone works
     dolt sql-client -P $PORT -u dolt --use-db '' -q "create database test01" ""
-    dolt sql-client -P $PORT -u dolt --use-db 'test01' -q"call dolt_clone('file:///$tempDir/remote')" 
+    dolt sql-client -P $PORT -u dolt --use-db 'test01' -q"call dolt_clone('file:///$tempDir/remote')"
 }
 
 @test "sql-server: loglevels are case insensitive" {
@@ -200,19 +200,19 @@ SQL
     [[ "$output" =~ "one_pk" ]] || false
 
     # Add rows on the command line
-    run dolt --user=dolt sql -q "insert into one_pk values (1,1,1)"
-    [ "$status" -eq 1 ]
-
+    run dolt --verbose-engine-setup --user=dolt sql -q "insert into one_pk values (1,1,1)"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting remote mode" ]] || false
     run dolt sql-client -P $PORT -u dolt -q "SELECT * FROM one_pk"
     [ $status -eq 0 ]
-    ! [[ $output =~ " 1 " ]] || false
+    [[ $output =~ " 1 " ]] || false
 
     # Test import as well (used by doltpy)
     echo 'pk,c1,c2' > import.csv
     echo '2,2,2' >> import.csv
     run dolt table import -u one_pk import.csv
     [ "$status" -eq 1 ]
-    
+
     run dolt sql-client -P $PORT -u dolt -q "SELECT * FROM one_pk"
     [ $status -eq 0 ]
     ! [[ $output =~ " 2 " ]] || false
@@ -235,7 +235,7 @@ SQL
         c1 BIGINT,
         c2 BIGINT,
         PRIMARY KEY (pk))"
-    
+
     run dolt ls
     [ "$status" -eq 0 ]
     [[ "$output" =~ "No tables in working set" ]] || false
@@ -394,7 +394,7 @@ SQL
     [[ $output =~ "1,1," ]] || false
     [[ $output =~ "2,2,2" ]] || false
     [[ $output =~ "3,3,3" ]] || false
-    
+
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "
     UPDATE r1_one_pk SET c2=1 WHERE pk=1;
     USE repo2;
@@ -445,10 +445,10 @@ SQL
     [[ $output =~ "r2_one_pk" ]] || false
 
     # put data in both using database scoped inserts
-    dolt sql-client -P $PORT -u dolt --use-db repo1 -q "INSERT INTO repo1.r1_one_pk (pk) VALUES (0)" 
+    dolt sql-client -P $PORT -u dolt --use-db repo1 -q "INSERT INTO repo1.r1_one_pk (pk) VALUES (0)"
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "INSERT INTO repo1.r1_one_pk (pk,c1) VALUES (1,1)"
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "INSERT INTO repo1.r1_one_pk (pk,c1,c2) VALUES (2,2,2),(3,3,3)"
-    
+
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "INSERT INTO repo2.r2_one_pk (pk) VALUES (0)"
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "INSERT INTO repo2.r2_one_pk (pk,c3) VALUES (1,1)"
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "INSERT INTO repo2.r2_one_pk (pk,c3,c4) VALUES (2,2,2),(3,3,3)"
@@ -466,10 +466,10 @@ SQL
     [[ $output =~ "1,1," ]] || false
     [[ $output =~ "2,2,2" ]] || false
     [[ $output =~ "3,3,3" ]] || false
-    
+
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "DELETE FROM repo1.r1_one_pk where pk=0"
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "DELETE FROM repo2.r2_one_pk where pk=0"
-    
+
     run dolt sql-client -P $PORT -u dolt --use-db repo1 --result-format csv -q "SELECT * FROM repo1.r1_one_pk ORDER BY pk"
     [ $status -eq 0 ]
     ! [[ $output =~ "0,," ]] || false
@@ -740,15 +740,15 @@ SQL
     INSERT INTO one_pk (pk,c1,c2) VALUES (2,2,2),(3,3,3);
     CALL DOLT_ADD('.');
     CALL dolt_commit('-am', 'test commit message', '--author', 'John Doe <john@example.com>');"
-    
+
     run dolt ls
     [ "$status" -eq 0 ]
     [[ "$output" =~ "one_pk" ]] || false
 
-    run dolt --user=dolt sql -q "drop table one_pk"
-    [ "$status" -eq 1 ]
+    run dolt --verbose-engine-setup  --user=dolt --use-db repo1 sql -q "drop table one_pk"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting remote mode" ]] || false
 
-    dolt sql-client -P $PORT -u dolt --use-db repo1 -q "drop table one_pk"
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "call dolt_add('.')"
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "call dolt_commit('-am', 'Dropped table one_pk')"
 
@@ -891,7 +891,7 @@ EOF
     dolt sql-server --config ./config.yml --socket "dolt.$PORT.sock" &
     SERVER_PID=$!
     sleep 1
-    
+
     # We do things manually here because we need to control
     # CLIENT_MULTI_STATEMENTS.
     python3 -c '
@@ -1005,7 +1005,7 @@ END""")
     [[ $output =~ "2,2" ]] || false
     [[ $output =~ "4,4" ]] || false
     ! [[ $output =~ "3,3" ]] || false
-    
+
     # drop the table on main, should keep counting from 4
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "drop table t1"
     dolt sql-client -P $PORT -u dolt --use-db repo1 -q "CREATE TABLE t1(pk bigint primary key auto_increment, val int)" ""
@@ -1055,7 +1055,7 @@ END""")
     cd repo3
     run dolt sql -q "select * from test" -r csv
     [ "$status" -eq 0 ]
-    [ "${lines[0]}" = "pk" ] 
+    [ "${lines[0]}" = "pk" ]
     [ "${lines[1]}" = "0" ]
     [ "${lines[2]}" = "1" ]
     [ "${lines[3]}" = "2" ]
@@ -1183,7 +1183,7 @@ END""")
     run dolt sql-client --use-db "test1/newbranch" -u dolt -P $PORT -q "select * from a"
     [ $status -ne 0 ]
     [[ "$output" =~ "database not found" ]] || false
-    
+
     # can't drop a branch-qualified database name
     run dolt sql-client -P $PORT -u dolt --use-db '' -q "drop database \`test2/newbranch\`"
     [ $status -ne 0 ]
@@ -1237,7 +1237,7 @@ END""")
     mkdir no_dolt && cd no_dolt
     mkdir db_dir
     start_sql_server_with_args --host 0.0.0.0 --user dolt --data-dir=db_dir
-    
+
     dolt sql-client -P $PORT -u dolt --use-db '' -q "create database test1"
     run dolt sql-client -P $PORT -u dolt --use-db '' -q "show databases"
     [ $status -eq 0 ]
@@ -1304,7 +1304,7 @@ END""")
     run dolt sql-client -P $PORT -u dolt --use-db '' -q "create database dir_exists"
     [ $status -ne 0 ]
     [[ $output =~ exists ]] || false
-    
+
     run dolt sql-client -P $PORT -u dolt --use-db '' -q	"create database file_exists"
     [ $status -ne 0 ]
     [[ $output =~ exists ]] || false
@@ -1496,14 +1496,17 @@ databases:
 
 @test "sql-server: sql-server locks database to writes" {
     cd repo2
-    dolt sql -q "create table a (x int primary key)" 
+    dolt sql -q "create table a (x int primary key)"
     start_sql_server
-    run dolt sql -q "create table b (x int primary key)" 
-    [ "$status" -eq 1 ]
-    [[ "$output" =~ "database is locked to writes" ]] || false
-    run dolt sql -q "insert into b values (0)"
+
+    run dolt --verbose-engine-setup sql -q "create table b (x int primary key)"
     [ "$status" -eq 1 ]
-    [[ "$output" =~ "database is locked to writes" ]] || false
+    [[ "$output" =~ "Error connecting to remote database" ]] || false
+    [[ "$output" =~ "User not found 'root'" ]] || false
+
+    run dolt --verbose-engine-setup --user dolt sql -q "create table b (x int primary key)"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "starting remote mode" ]] || false
 }
 
 @test "sql-server: start server without socket flag should set default socket path" {
diff --git a/integration-tests/bats/sql-shell.bats b/integration-tests/bats/sql-shell.bats
index 7751fb06e9f..59de2c442d8 100644
--- a/integration-tests/bats/sql-shell.bats
+++ b/integration-tests/bats/sql-shell.bats
@@ -478,8 +478,7 @@ teardown() {
     ! [[ "$output" =~ ".doltcfg" ]] || false
 
     # create new user
-    run dolt --data-dir=db_dir --privilege-file=privs.db sql <<< "create user new_user"
-    [ "$status" -eq 0 ]
+    dolt --data-dir=db_dir --privilege-file=privs.db sql <<< "create user new_user"
 
     # show users, expect root user and new_user
     run dolt --data-dir=db_dir --privilege-file=privs.db sql <<< "select user from mysql.user;"
diff --git a/integration-tests/bats/sql.bats b/integration-tests/bats/sql.bats
index 2591057a945..6f5c2763fb0 100755
--- a/integration-tests/bats/sql.bats
+++ b/integration-tests/bats/sql.bats
@@ -1031,7 +1031,7 @@ SQL
     run dolt sql -r csv -q "select @@character_set_client"
     [ $status -eq 0 ]
     [[ "$output" =~ "utf8mb4" ]] || false
-    
+
     run dolt sql -r json -q "select * from test order by a"
     [ $status -eq 0 ]
     [ "$output" == '{"rows": [{"a":1,"b":1.5,"c":"1","d":"2020-01-01 00:00:00"},{"a":2,"b":2.5,"c":"2","d":"2020-02-02 00:00:00"},{"a":3,"c":"3","d":"2020-03-03 00:00:00"},{"a":4,"b":4.5,"d":"2020-04-04 00:00:00"},{"a":5,"b":5.5,"c":"5"}]}' ]
@@ -2776,3 +2776,57 @@ SQL
 [[ "$output" =~ "Query OK, 1 row affected (2".*" sec)" ]] || false
 [[ "$output" =~ "Query OK, 1 row affected (3".*" sec)" ]] || false
 }
+
+
+@test "sql: check --data-dir used from a completely different location and still resolve DB names" {
+    # remove config files
+    rm -rf .doltcfg
+    rm -rf db_dir
+
+    mkdir db_dir
+    cd db_dir
+    ROOT_DIR=$(pwd)
+
+    # create an alternate database, without the table
+    mkdir dba
+    cd dba
+    dolt init
+    cd ..
+    dolt sql -q "create table dba_tbl (id int)"
+
+    mkdir dbb
+    cd dbb
+    dolt init
+    dolt sql -q "create table dbb_tbl (id int)"
+
+    # Ensure --data-dir flag is really used by changing the cwd.
+    cd /tmp
+
+    run dolt --data-dir="$ROOT_DIR/dbb" sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "dbb_tbl" ]] || false
+
+    run dolt --data-dir="$ROOT_DIR/dba" sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "dba_tbl" ]] || false
+
+    # Default to first DB alphabetically.
+    run dolt --data-dir="$ROOT_DIR" sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "dba_tbl" ]] || false
+
+    # --use-db arg can be used to be specific.
+    run dolt --data-dir="$ROOT_DIR" --use-db=dbb sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "dbb_tbl" ]] || false
+
+    # Redundant use of the flag is OK.
+    run dolt --data-dir="$ROOT_DIR/dbb" --use-db=dbb sql -q "show tables"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "dbb_tbl" ]] || false
+
+    # Use of the use-db flag when we have a different DB specified by data-dir should error.
+    run dolt --data-dir="$ROOT_DIR/dbb" --use-db=dba sql -q "show tables"
+    [ "$status" -eq 1 ]
+    [[ "$output" =~ "provided --use-db dba does not exist or is not a directory" ]] || false
+}