First pass on support for dolt_undrop() stored procedure to "undrop" a database after it has been dropped.

go/libraries/doltcore/sqle/database_provider.go

@@ -18,9 +18,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"path/filepath"
 	"sort"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/dolthub/go-mysql-server/sql"
 
@@ -42,6 +44,10 @@ import (
 	"github.com/dolthub/dolt/go/store/types"
 )
 
+// deletedDatabaseDirectoryName is the subdirectory within the data folder where Dolt moves databases after they are
+// dropped. The dolt_undrop() stored procedure is then able to restore them from this location.
+const deletedDatabaseDirectoryName = "dolt_deleted_databases"
+
 type DoltDatabaseProvider struct {
 	// dbLocations maps a database name to its file system root
 	dbLocations        map[string]filesys.Filesys
@@ -405,46 +411,7 @@ func (p DoltDatabaseProvider) CreateCollatedDatabase(ctx *sql.Context, name stri
 		}
 	}
 
-	// If we're running in a sql-server context, ensure the new database is locked so that it can't
-	// be edited from the CLI. We can't rely on looking for an existing lock file, since this could
-	// be the first db creation if sql-server was started from a bare directory.
-	_, lckDeets := sqlserver.GetRunningServer()
-	if lckDeets != nil {
-		err = newEnv.Lock(lckDeets)
-		if err != nil {
-			ctx.GetLogger().Warnf("Failed to lock newly created database: %s", err.Error())
-		}
-	}
-
-	fkChecks, err := ctx.GetSessionVariable(ctx, "foreign_key_checks")
-	if err != nil {
-		return err
-	}
-
-	opts := editor.Options{
-		Deaf: newEnv.DbEaFactory(),
-		// TODO: this doesn't seem right, why is this getting set in the constructor to the DB
-		ForeignKeyChecksDisabled: fkChecks.(int8) == 0,
-	}
-
-	db, err := NewDatabase(ctx, name, newEnv.DbData(), opts)
-	if err != nil {
-		return err
-	}
-
-	// If we have an initialization hook, invoke it.  By default, this will
-	// be ConfigureReplicationDatabaseHook, which will setup replication
-	// for the new database if a remote url template is set.
-	err = p.InitDatabaseHook(ctx, p, name, newEnv)
-	if err != nil {
-		return err
-	}
-
-	formattedName := formatDbMapKeyName(db.Name())
-	p.databases[formattedName] = db
-	p.dbLocations[formattedName] = newEnv.FS
-
-	return nil
+	return p.registerNewDatabase(ctx, name, newEnv)
 }
 
 type InitDatabaseHook func(ctx *sql.Context, pro DoltDatabaseProvider, name string, env *env.DoltEnv) error
@@ -598,6 +565,23 @@ func (p DoltDatabaseProvider) cloneDatabaseFromRemote(
 	return dEnv, nil
 }
 
+// initializeDeletedDatabaseDirectory initializes the special directory Dolt uses to store dropped databases until
+// they are fully removed. If the directory is already created and set up correctly, then this method is a no-op.
+// If the directory doesn't exist yet, it will be created. If there are any problems initializing the directory, an
+// error is returned.
+func (p DoltDatabaseProvider) initializeDeletedDatabaseDirectory() error {
+	exists, isDir := p.fs.Exists(deletedDatabaseDirectoryName)
+	if exists && !isDir {
+		return fmt.Errorf("%s exists, but is not a directory", deletedDatabaseDirectoryName)
+	}
+
+	if exists {
+		return nil
+	}
+
+	return p.fs.MkDirs(deletedDatabaseDirectoryName)
+}
+
 // DropDatabase implements the sql.MutableDatabaseProvider interface
 func (p DoltDatabaseProvider) DropDatabase(ctx *sql.Context, name string) error {
 	_, revision := dsess.SplitRevisionDbName(name)
@@ -648,7 +632,9 @@ func (p DoltDatabaseProvider) DropDatabase(ctx *sql.Context, name string) error
 	if err != nil {
 		return err
 	}
-	dirToDelete := ""
+
+	isRootDatabase := false
+	//dirToDelete := ""
 	// if the database is in the directory itself, we remove '.dolt' directory rather than
 	// the whole directory itself because it can have other databases that are nested.
 	if rootDbLoc == dropDbLoc {
@@ -656,8 +642,11 @@ func (p DoltDatabaseProvider) DropDatabase(ctx *sql.Context, name string) error
 		if !doltDirExists {
 			return sql.ErrDatabaseNotFound.New(db.Name())
 		}
-		dirToDelete = dbfactory.DoltDir
+		dropDbLoc = filepath.Join(dropDbLoc, dbfactory.DoltDir)
+		isRootDatabase = true
 	} else {
+		// TODO: Do we really need the code in this block?
+		//       Seems like a few places are checking this.
 		exists, isDir := p.fs.Exists(dropDbLoc)
 		// Get the DB's directory
 		if !exists {
@@ -666,16 +655,43 @@ func (p DoltDatabaseProvider) DropDatabase(ctx *sql.Context, name string) error
 		} else if !isDir {
 			return fmt.Errorf("unexpected error: %s exists but is not a directory", dbKey)
 		}
-		dirToDelete = dropDbLoc
 	}
 
-	err = p.fs.Delete(dirToDelete, true)
-	if err != nil {
+	if err = p.initializeDeletedDatabaseDirectory(); err != nil {
+		return fmt.Errorf("unable to drop database %s: %w", name, err.Error())
+	}
+
+	// Move the dropped database to the Dolt deleted database directory so it can be restored if needed
+	_, file := filepath.Split(dropDbLoc)
+	var destinationDirectory string
+	if isRootDatabase {
+		// NOTE: This won't work without first creating the new subdirectory
+		newSubdirectory := filepath.Join(deletedDatabaseDirectoryName, name)
+		// TODO: If newSubdirectory exists already... then we're in trouble! (need to handle that)
+		// TODO: Maybe we should have this talk to the DroppedDatabaseVault API instead?
+		if err := p.fs.MkDirs(newSubdirectory); err != nil {
+			return err
+		}
+		destinationDirectory = filepath.Join(newSubdirectory, file)
+	} else {
+		destinationDirectory = filepath.Join(deletedDatabaseDirectoryName, file)
+	}
+
+	// Add the final directory segment and convert all hyphens to underscores in the database directory name
+	dir, file := filepath.Split(destinationDirectory)
+	if strings.Contains(file, "-") {
+		destinationDirectory = filepath.Join(dir, strings.ReplaceAll(file, "-", "_"))
+	}
+
+	if err := p.prepareToMoveDroppedDatabase(ctx, destinationDirectory); err != nil {
+		return err
+	}
+	if err = p.fs.MoveDir(dropDbLoc, destinationDirectory); err != nil {
 		return err
 	}
 
-	// We not only have to delete this database, but any derivative ones that we've stored as a result of USE or
-	// connection strings
+	// We not only have to delete tracking metadata for this database, but also for any derivative ones we've stored
+	// as a result of USE or connection strings
 	derivativeNamePrefix := strings.ToLower(dbKey + dsess.DbRevisionDelimiter)
 	for dbName := range p.databases {
 		if strings.HasPrefix(strings.ToLower(dbName), derivativeNamePrefix) {
@@ -688,6 +704,170 @@ func (p DoltDatabaseProvider) DropDatabase(ctx *sql.Context, name string) error
 	return p.invalidateDbStateInAllSessions(ctx, name)
 }
 
+// TODO: Might be helpful to group some of these DoltDeletedDatabaseDirectory helper functions into their own
+//
+//	type. For example... maybe there's a new type for DroppedDatabaseVault that we can interact with?
+func (p DoltDatabaseProvider) prepareToMoveDroppedDatabase(_ *sql.Context, targetPath string) error {
+	if exists, _ := p.fs.Exists(targetPath); !exists {
+		// If there's nothing at the desired targetPath, we're all set
+		return nil
+	}
+
+	// If there is something already there, pick a new path to move it to
+	newPath := fmt.Sprintf("%s.backup.%d", targetPath, time.Now().Unix())
+	if exists, _ := p.fs.Exists(newPath); exists {
+		return fmt.Errorf("unable to move existing dropped database out of the way: "+
+			"tried to move it to %s", newPath)
+	}
+	if err := p.fs.MoveFile(targetPath, newPath); err != nil {
+		return fmt.Errorf("unable to move existing dropped database out of the way: %w", err)
+	}
+
+	return nil
+}
+
+func (p DoltDatabaseProvider) ListUndroppableDatabases(_ *sql.Context) ([]string, error) {
+	if err := p.initializeDeletedDatabaseDirectory(); err != nil {
+		return nil, fmt.Errorf("unable to list undroppable database: %w", err.Error())
+	}
+
+	databaseNames := make([]string, 0, 5)
+	callback := func(path string, size int64, isDir bool) (stop bool) {
+		_, lastPathSegment := filepath.Split(path)
+		// TODO: Is there a common util we use for this somewhere?
+		lastPathSegment = strings.ReplaceAll(lastPathSegment, "-", "_")
+		databaseNames = append(databaseNames, lastPathSegment)
+		return false
+	}
+
+	if err := p.fs.Iter(deletedDatabaseDirectoryName, false, callback); err != nil {
+		return nil, err
+	}
+
+	return databaseNames, nil
+}
+
+// validateUndropDatabase validates that the database |name| is available to be "undropped" and that no existing
+// database is already being managed that has the same (case-insensitive) name. If any problems are encountered,
+// an error is returned.
+func (p DoltDatabaseProvider) validateUndropDatabase(ctx *sql.Context, name string) (sourcePath, destinationPath, exactCaseName string, err error) {
+	// TODO: rename to ListDatabasesThatCanBeUndropped(ctx)?
+	availableDatabases, err := p.ListUndroppableDatabases(ctx)
+	if err != nil {
+		return "", "", "", err
+	}
+
+	found := false
+	exactCaseName = name
+	lowercaseName := strings.ToLower(name)
+	for _, s := range availableDatabases {
+		if lowercaseName == strings.ToLower(s) {
+			exactCaseName = s
+			found = true
+			break
+		}
+	}
+
+	// TODO: this error creation information could be extracted to a common function that dolt_undrop function could use
+	if !found {
+		extraInformation := "there are no databases currently available to be undropped"
+		if len(availableDatabases) > 0 {
+			extraInformation = fmt.Sprintf("available databases that can be undropped: %s", strings.Join(availableDatabases, ", "))
+		}
+		return "", "", "", fmt.Errorf("no database named '%s' found to undrop. %s", name, extraInformation)
+	}
+
+	// Check to see if the destination directory for restoring the database already exists (case insensitive match)
+	destinationPath, err = p.fs.Abs(exactCaseName)
+	if err != nil {
+		return "", "", "", err
+	}
+
+	sourcePath = filepath.Join(deletedDatabaseDirectoryName, exactCaseName)
+
+	// TODO: is this always a case insensitive check??? It seems like it must be since our test is working?
+	if exists, _ := p.fs.Exists(destinationPath); exists {
+		return "", "", "", fmt.Errorf("unable to undrop database '%s'; "+
+			"another database already exists with the same case-insensitive name", exactCaseName)
+	}
+
+	return sourcePath, destinationPath, exactCaseName, nil
+}
+
+func (p DoltDatabaseProvider) UndropDatabase(ctx *sql.Context, name string) (err error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	// TODO: not sure I like sourcePath and destinationPath being returned here, but seems like they're needed in this function
+	sourcePath, destinationPath, exactCaseName, err := p.validateUndropDatabase(ctx, name)
+	if err != nil {
+		return err
+	}
+
+	// TODO: Need to account for database directory renaming (i.e. converting '-' to '_')
+
+	err = p.fs.MoveDir(sourcePath, destinationPath)
+	if err != nil {
+		return err
+	}
+
+	newFs, err := p.fs.WithWorkingDir(exactCaseName)
+	if err != nil {
+		return err
+	}
+	newEnv := env.Load(ctx, env.GetCurrentUserHomeDir, newFs, p.dbFactoryUrl, "TODO")
+
+	return p.registerNewDatabase(ctx, exactCaseName, newEnv)
+}
+
+// registerNewDatabase registers the specified DoltEnv, |newEnv|, as a new database named |name|. This
+// function is responsible for instantiating the new Database instance and updating the tracking metadata
+// in this provider. If any problems are encountered while registering the new database, an error is returned.
+func (p DoltDatabaseProvider) registerNewDatabase(ctx *sql.Context, name string, newEnv *env.DoltEnv) (err error) {
+	// If we're running in a sql-server context, ensure the new database is locked so that it can't
+	// be edited from the CLI. We can't rely on looking for an existing lock file, since this could
+	// be the first db creation if sql-server was started from a bare directory.
+	_, lckDeets := sqlserver.GetRunningServer()
+	if lckDeets != nil {
+		err = newEnv.Lock(lckDeets)
+		if err != nil {
+			ctx.GetLogger().Warnf("Failed to lock newly created database: %s", err.Error())
+		}
+	}
+
+	fkChecks, err := ctx.GetSessionVariable(ctx, "foreign_key_checks")
+	if err != nil {
+		return err
+	}
+
+	opts := editor.Options{
+		Deaf: newEnv.DbEaFactory(),
+		// TODO: this doesn't seem right, why is this getting set in the constructor to the DB
+		ForeignKeyChecksDisabled: fkChecks.(int8) == 0,
+	}
+
+	db, err := NewDatabase(ctx, name, newEnv.DbData(), opts)
+	if err != nil {
+		return err
+	}
+
+	// If we have an initialization hook, invoke it.  By default, this will
+	// be ConfigureReplicationDatabaseHook, which will setup replication
+	// for the new database if a remote url template is set.
+	err = p.InitDatabaseHook(ctx, p, name, newEnv)
+	if err != nil {
+		return err
+	}
+
+	// TODO: accessing p.databases requires locking!!!
+	//       But right now we're just assuming this function is called from another function
+	//       that has grabbed the right lock, but that's eventually going to cause a problem.
+	formattedName := formatDbMapKeyName(db.Name())
+	p.databases[formattedName] = db
+	p.dbLocations[formattedName] = newEnv.FS
+	return nil
+}
+
 // invalidateDbStateInAllSessions removes the db state for this database from every session. This is necessary when a
 // database is dropped, so that other sessions don't use stale db state.
 func (p DoltDatabaseProvider) invalidateDbStateInAllSessions(ctx *sql.Context, name string) error {

go/libraries/doltcore/sqle/dprocedures/dolt_undrop.go

@@ -0,0 +1,54 @@
+// Copyright 2023 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 dprocedures
+
+import (
+	"fmt"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess"
+	"github.com/dolthub/go-mysql-server/sql"
+	"strings"
+)
+
+// doltClean is the stored procedure version for the CLI command `dolt clean`.
+func doltUndrop(ctx *sql.Context, args ...string) (sql.RowIter, error) {
+	doltSession := dsess.DSessFromSess(ctx.Session)
+	provider := doltSession.Provider()
+
+	switch len(args) {
+	case 0:
+		// TODO: Are there any permission issues for undrop? probably the same as drop?
+		undroppableDatabases, err := provider.ListUndroppableDatabases(ctx)
+		if err != nil {
+			return nil, err
+		}
+
+		extraInformation := "there are no databases that can currently be undropped."
+		if len(undroppableDatabases) > 0 {
+			extraInformation = fmt.Sprintf("the following dropped databases are availble to be undropped: %s",
+				strings.Join(undroppableDatabases, ", "))
+		}
+		return nil, fmt.Errorf("no database name specified. %s", extraInformation)
+
+	case 1:
+		if err := provider.UndropDatabase(ctx, args[0]); err != nil {
+			return nil, err
+		}
+		return rowToIter(int64(0)), nil
+
+	default:
+		return nil, fmt.Errorf("dolt_undrop called with too many arguments: " +
+			"dolt_undrop only accepts one argument - the name of the dropped database to restore")
+	}
+}

go/libraries/doltcore/sqle/dprocedures/init.go

@@ -32,6 +32,7 @@ var DoltProcedures = []sql.ExternalStoredProcedureDetails{
 	{Name: "dolt_conflicts_resolve", Schema: int64Schema("status"), Function: doltConflictsResolve},
 	{Name: "dolt_count_commits", Schema: int64Schema("ahead", "behind"), Function: doltCountCommits, ReadOnly: true},
 	{Name: "dolt_fetch", Schema: int64Schema("status"), Function: doltFetch},
+	{Name: "dolt_undrop", Schema: int64Schema("status"), Function: doltUndrop},
 
 	// dolt_gc is enabled behind a feature flag for now, see dolt_gc.go
 	{Name: "dolt_gc", Schema: int64Schema("status"), Function: doltGC, ReadOnly: true},
@@ -55,6 +56,7 @@ var DoltProcedures = []sql.ExternalStoredProcedureDetails{
 	{Name: "dclone", Schema: int64Schema("status"), Function: doltClone},
 	{Name: "dcommit", Schema: stringSchema("hash"), Function: doltCommit},
 	{Name: "dfetch", Schema: int64Schema("status"), Function: doltFetch},
+	{Name: "dundrop", Schema: int64Schema("status"), Function: doltUndrop},
 
 	//	{Name: "dgc", Schema: int64Schema("status"), Function: doltGC},
 

go/libraries/doltcore/sqle/dsess/session_db_provider.go

@@ -96,6 +96,17 @@ type DoltDatabaseProvider interface {
 	BaseDatabase(ctx *sql.Context, dbName string) (SqlDatabase, bool)
 	// DoltDatabases returns all databases known to this provider.
 	DoltDatabases() []SqlDatabase
+	// UndropDatabase attempts to restore the database |dbName| that was previously dropped.
+	// The restored database will appear identically when accessed through the SQL
+	// interface, but may be stored in a slightly different location on disk
+	// (e.g. a root database will be restored as a regular/non-root database,
+	// databases original stored with hyphens in their directory name will be rewritten
+	// to underscores to match their SQL database name).
+	// TODO: Is this a problem for anything on hosted?
+	// If the database is unable to be restored, an error is returned explaining why.
+	UndropDatabase(ctx *sql.Context, dbName string) error
+	// ListUndroppableDatabases returns a list of database names for dropped databases that are available to be restored.
+	ListUndroppableDatabases(ctx *sql.Context) ([]string, error)
 }
 
 type SessionDatabaseBranchSpec struct {

go/libraries/utils/filesys/fs.go

@@ -72,6 +72,11 @@ type WritableFS interface {
 	// MoveFile will move a file from the srcPath in the filesystem to the destPath
 	MoveFile(srcPath, destPath string) error
 
+	// MoveDir will move a directory from the srcPath in the filesystem to the destPath. For example,
+	// MoveDir("foo", "bar/baz") will move the "foo" directory to "bar/baz", meaning the contents of "foo" are now
+	// directly under the "baz" directory.
+	MoveDir(srcPath, destPath string) error
+
 	// TempDir returns the path of a new temporary directory.
 	TempDir() string
 }

go/libraries/utils/filesys/inmemfs.go

@@ -537,6 +537,10 @@ func (fs *InMemFS) MoveFile(srcPath, destPath string) error {
 	return os.ErrNotExist
 }
 
+func (fs InMemFS) MoveDir(srcPath, destPath string) error {
+	panic("not implemented!")
+}
+
 func (fs *InMemFS) CopyFile(srcPath, destPath string) error {
 	fs.rwLock.Lock()
 	defer fs.rwLock.Unlock()

go/libraries/utils/filesys/localfs.go

@@ -271,8 +271,7 @@ func (fs *localFS) Delete(path string, force bool) error {
 }
 
 // MoveFile will move a file from the srcPath in the filesystem to the destPath
-func (fs *localFS) MoveFile(srcPath, destPath string) error {
-	var err error
+func (fs *localFS) MoveFile(srcPath, destPath string) (err error) {
 	srcPath, err = fs.Abs(srcPath)
 
 	if err != nil {
@@ -288,6 +287,24 @@ func (fs *localFS) MoveFile(srcPath, destPath string) error {
 	return file.Rename(srcPath, destPath)
 }
 
+func (fs *localFS) MoveDir(srcPath, destPath string) (err error) {
+	// TODO: This is the exact same implementation as MoveFile
+	//       Should probably at least add assertions that |srcPath| is really a dir?
+	// TODO: Or should we just try to make MoveFile work with dirs? It seems like the filesystem
+	//       based implementation already does, it's just the in-memory implementation that doesn't.
+	srcPath, err = fs.Abs(srcPath)
+	if err != nil {
+		return err
+	}
+
+	destPath, err = fs.Abs(destPath)
+	if err != nil {
+		return err
+	}
+
+	return file.Rename(srcPath, destPath)
+}
+
 // converts a path to an absolute path.  If it's already an absolute path the input path will be returned unaltered
 func (fs *localFS) Abs(path string) (string, error) {
 	if filepath.IsAbs(path) {

integration-tests/bats/undrop.bats

@@ -0,0 +1,192 @@
+#!/usr/bin/env bats
+load $BATS_TEST_DIRNAME/helper/common.bash
+
+setup() {
+    setup_common
+}
+
+teardown() {
+    assert_feature_version
+    teardown_common
+}
+
+@test "undrop: GC deletes dropped databases" {
+	# TODO: Garbage collection should remove deleted databases; implement in second milestone
+  skip "not supported yet"
+}
+
+@test "undrop: error messages" {
+  # When called without any argument, dolt_undrop() returns an error
+  # that includes the database names that can be undropped.
+  run dolt sql -q "CALL dolt_undrop();"
+  [ $status -eq 1 ]
+  [[ $output =~ "no database name specified." ]] || false
+  [[ $output =~ "there are no databases that can currently be undropped" ]] || false
+
+  # When called without an invalid database name, dolt_undrop() returns
+  # an error that includes the database names that can be undropped.
+  run dolt sql -q "CALL dolt_undrop('doesnotexist')"
+  [ $status -eq 1 ]
+  [[ $output =~ "no database named 'doesnotexist' found to undrop" ]] || false
+  [[ $output =~ "there are no databases currently available to be undropped" ]] || false
+
+  # When called with multiple arguments, dolt_undrop() returns an error
+  # explaining that only one argument may be specified.
+  run dolt sql -q "CALL dolt_undrop('one', 'two', 'three')"
+  [ $status -eq 1 ]
+  [[ $output =~ "dolt_undrop called with too many arguments" ]] || false
+  [[ $output =~ "dolt_undrop only accepts one argument - the name of the dropped database to restore" ]] || false
+}
+
+@test "undrop: undrop root database" {
+  # Create a new Dolt database directory to use as a root database
+  # NOTE: We use hyphens here to test how db dirs are renamed.
+  mkdir test-db-1 && cd test-db-1
+  dolt init
+
+  # Create some data and a commit in the database
+  dolt sql << EOF
+create table t1 (pk int primary key, c1 varchar(200));
+insert into t1 values (1, "one");
+call dolt_commit('-Am', 'creating table t1');
+EOF
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ $output =~ "test_db_1" ]] || false
+
+  # Drop the root database
+  dolt sql -q "drop database test_db_1;"
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ ! $output =~ "test_db_1" ]] || false
+
+  # Undrop the test_db_1 database
+  # NOTE: After being undropped, the database is no longer the root database,
+  #       but contained in a subdirectory like a non-root database.
+  dolt sql -q "call dolt_undrop('test_db_1');"
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ $output =~ "test_db_1" ]] || false
+
+  # Sanity check querying some data
+  run dolt sql -r csv -q "select * from test_db_1.t1;"
+  [ $status -eq 0 ]
+  [[ $output =~ "1,one" ]] || false
+}
+
+# Asserts that a non-root database can be dropped and then restored with dolt_undrop(), even when
+# the case of the database name given to dolt_undrop() doesn't match match the original case.
+@test "undrop: undrop non-root database" {
+  # We manually create a database directory with hyphens in it to test the drop/undrop logic
+  # that handles translating database directory names to logical database names.
+  mkdir drop-me && cd drop-me
+  dolt init && cd ..
+
+  dolt sql << EOF
+use drop_me;
+create table t1 (pk int primary key, c1 varchar(200));
+insert into t1 values (1, "one");
+call dolt_commit('-Am', 'creating table t1');
+EOF
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ $output =~ "drop_me" ]] || false
+
+  dolt sql -q "drop database drop_me;"
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ ! $output =~ "drop_me" ]] || false
+
+  # Call dolt_undrop() with non-matching case for the database name to
+  # ensure dolt_undrop() works with case-insensitive database names.
+  dolt sql -q "call dolt_undrop('DrOp_mE');"
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ $output =~ "drop_me" ]] || false
+
+  run dolt sql -r csv -q "select * from drop_me.t1;"
+  [ $status -eq 0 ]
+  [[ $output =~ "1,one" ]] || false
+}
+
+# When a database is dropped, and then a new database is recreated
+# with the same name and dropped, dolt_undrop will undrop the most
+# recent database with that name.
+@test "undrop: drop database, recreate, and drop again" {
+  # Create a database named test123
+  dolt sql << EOF
+create database test123;
+use test123;
+create table t1 (pk int primary key, c1 varchar(100));
+insert into t1 values (1, "one");
+call dolt_commit('-Am', 'adding table t1 to test123 database');
+EOF
+
+  # Drop database test123 and make sure it's gone
+  dolt sql -q "drop database test123;"
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ ! $output =~ "test123" ]] || false
+
+  # Create a new database named test123
+  dolt sql << EOF
+create database test123;
+use test123;
+create table t2 (pk int primary key, c2 varchar(100));
+insert into t2 values (100, "one hundy");
+call dolt_commit('-Am', 'adding table t2 to new test123 database');
+EOF
+
+  # Drop the new test123 database and make sure it's gone
+  dolt sql -q "drop database test123;"
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ ! $output =~ "test123" ]] || false
+
+  # Undrop the database
+  dolt sql -q "call dolt_undrop('test123');"
+  run dolt sql -r csv -q "select * from test123.t2;"
+  [ $status -eq 0 ]
+  [[ $output =~ "100,one hundy" ]] || false
+}
+
+# Asserts that when there is already an existing database with the same name, a dropped database
+# cannot be undropped.
+# TODO: In the future, it might be useful to allow dolt_undrop() to rename the dropped database to
+#       a new name, but for now, keep it simple and just disallow restoring in this case.
+@test "undrop: undrop conflict" {
+  dolt sql << EOF
+create database dAtAbAsE1;
+use dAtAbAsE1;
+create table t1 (pk int primary key, c1 varchar(200));
+insert into t1 values (1, "one");
+call dolt_commit('-Am', 'creating table t1');
+EOF
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ $output =~ "dAtAbAsE1" ]] || false
+
+  # Drop dAtAbAsE1
+  dolt sql -q "drop database dAtAbAsE1;"
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ ! $output =~ "dAtAbAsE1" ]] || false
+
+  # Create a new database named dAtAbAsE1
+  dolt sql << EOF
+create database database1;
+use database1;
+create table t2 (pk int primary key, c1 varchar(200));
+insert into t2 values (1000, "thousand");
+call dolt_commit('-Am', 'creating table t2');
+EOF
+  run dolt sql -q "show databases;"
+  [ $status -eq 0 ]
+  [[ $output =~ "database1" ]] || false
+
+  # Trying to undrop dAtAbAsE1 results in an error, since a database already exists
+  run dolt sql -q "call dolt_undrop('dAtAbAsE1');"
+  [ $status -eq 1 ]
+  [[ $output =~ "unable to undrop database 'dAtAbAsE1'" ]] || false
+  [[ $output =~ "another database already exists with the same case-insensitive name" ]] || false
+}