Merge branch 'master' into andy/database-revisions

go/cmd/dolt/commands/cvcmds/constraint_violations.go

@@ -0,0 +1,21 @@
+// Copyright 2021 Dolthub, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cvcmds
+
+import "github.com/dolthub/dolt/go/cmd/dolt/cli"
+
+var Commands = cli.NewSubCommandHandler("constraints", "Commands for handling constraints.", []cli.Command{
+	VerifyConstraintsCmd{},
+})

go/cmd/dolt/commands/cvcmds/verify_constraints.go

@@ -0,0 +1,152 @@
+// Copyright 2020 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 cvcmds
+
+import (
+	"context"
+
+	"github.com/dolthub/go-mysql-server/sql"
+
+	"github.com/dolthub/dolt/go/cmd/dolt/cli"
+	"github.com/dolthub/dolt/go/cmd/dolt/commands"
+	"github.com/dolthub/dolt/go/cmd/dolt/errhand"
+	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
+	"github.com/dolthub/dolt/go/libraries/doltcore/env"
+	"github.com/dolthub/dolt/go/libraries/doltcore/merge"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/sqlutil"
+	"github.com/dolthub/dolt/go/libraries/utils/argparser"
+	"github.com/dolthub/dolt/go/libraries/utils/filesys"
+	"github.com/dolthub/dolt/go/libraries/utils/set"
+)
+
+const (
+	vcAllParam        = "all"
+	vcOutputOnlyParam = "output-only"
+)
+
+var verifyConstraintsDocs = cli.CommandDocumentationContent{
+	ShortDesc: `Verifies a table's constraints`,
+	LongDesc: `This command verifies that the defined constraints on the given table(s)—such as a foreign key—are correct and satisfied.
+By default, compares the working set to to the HEAD commit. Additionally, by default this updates this table's associated
+dolt_constraint_violations system table. Both of these default behaviors may be changed with the appropriate parameters.`,
+	Synopsis: []string{`[--all] [--output-only] [{{.LessThan}}table{{.GreaterThan}}...]`},
+}
+
+type VerifyConstraintsCmd struct{}
+
+var _ cli.Command = VerifyConstraintsCmd{}
+
+func (cmd VerifyConstraintsCmd) Name() string {
+	return "verify"
+}
+
+func (cmd VerifyConstraintsCmd) Description() string {
+	return "Command to verify that the constraints on the given table(s) are satisfied."
+}
+
+func (cmd VerifyConstraintsCmd) CreateMarkdown(fs filesys.Filesys, path, commandStr string) error {
+	return nil
+}
+
+func (cmd VerifyConstraintsCmd) createArgParser() *argparser.ArgParser {
+	ap := argparser.NewArgParser()
+	ap.SupportsFlag(vcAllParam, "a", "Verifies constraints against every row.")
+	ap.SupportsFlag(vcOutputOnlyParam, "o", "Disables writing the results to the constraint violations table.")
+	ap.ArgListHelp = append(ap.ArgListHelp, [2]string{"table", "The table(s) to check constraints on. If omitted, checks all tables."})
+	return ap
+}
+
+func (cmd VerifyConstraintsCmd) Exec(ctx context.Context, commandStr string, args []string, dEnv *env.DoltEnv) int {
+	ap := cmd.createArgParser()
+	help, _ := cli.HelpAndUsagePrinters(cli.GetCommandDocumentation(commandStr, verifyConstraintsDocs, ap))
+	apr := cli.ParseArgsOrDie(ap, args, help)
+
+	verifyAllRows := apr.Contains(vcAllParam)
+	outputOnly := apr.Contains(vcOutputOnlyParam)
+	working, err := dEnv.WorkingRoot(ctx)
+	if err != nil {
+		return commands.HandleVErrAndExitCode(errhand.BuildDError("Unable to get working.").AddCause(err).Build(), nil)
+	}
+	tableNames := apr.Args()
+	if len(tableNames) == 0 {
+		tableNames, err = working.GetTableNames(ctx)
+		if err != nil {
+			return commands.HandleVErrAndExitCode(errhand.BuildDError("Unable to read table names.").AddCause(err).Build(), nil)
+		}
+	}
+	tableSet := set.NewStrSet(tableNames)
+
+	comparingRoot, err := dEnv.HeadRoot(ctx)
+	if err != nil {
+		return commands.HandleVErrAndExitCode(errhand.BuildDError("Unable to get head root.").AddCause(err).Build(), nil)
+	}
+	if verifyAllRows {
+		comparingRoot, err = doltdb.EmptyRootValue(ctx, comparingRoot.VRW())
+		if err != nil {
+			return commands.HandleVErrAndExitCode(errhand.BuildDError("Unable to create an empty root.").AddCause(err).Build(), nil)
+		}
+	}
+	endRoot, tablesWithViolations, err := merge.AddConstraintViolations(ctx, working, comparingRoot, tableSet)
+	if err != nil {
+		return commands.HandleVErrAndExitCode(errhand.BuildDError("Unable to process constraint violations.").AddCause(err).Build(), nil)
+	}
+	if !outputOnly {
+		err = dEnv.UpdateWorkingRoot(ctx, endRoot)
+		if err != nil {
+			return commands.HandleVErrAndExitCode(errhand.BuildDError("Unable to update working root.").AddCause(err).Build(), nil)
+		}
+	}
+
+	if tablesWithViolations.Size() > 0 {
+		cli.PrintErrln("All constraints are not satisfied.")
+		for _, tableName := range tablesWithViolations.AsSortedSlice() {
+			table, ok, err := endRoot.GetTable(ctx, tableName)
+			if err != nil {
+				return commands.HandleVErrAndExitCode(errhand.BuildDError("Error loading table.").AddCause(err).Build(), nil)
+			}
+			if !ok {
+				return commands.HandleVErrAndExitCode(errhand.BuildDError("Unable to load table '%s'.", tableName).Build(), nil)
+			}
+			cvSch, err := table.GetConstraintViolationsSchema(ctx)
+			if err != nil {
+				return commands.HandleVErrAndExitCode(errhand.BuildDError("Error loading constraint violations schema.").AddCause(err).Build(), nil)
+			}
+			cvMap, err := table.GetConstraintViolations(ctx)
+			if err != nil {
+				return commands.HandleVErrAndExitCode(errhand.BuildDError("Error loading constraint violations data.").AddCause(err).Build(), nil)
+			}
+			sqlSchema, err := sqlutil.FromDoltSchema(tableName, cvSch)
+			if err != nil {
+				return commands.HandleVErrAndExitCode(errhand.BuildDError("Error attempting to convert schema").AddCause(err).Build(), nil)
+			}
+			rowIter, err := sqlutil.MapToSqlIter(ctx, cvSch, cvMap)
+			if err != nil {
+				return commands.HandleVErrAndExitCode(errhand.BuildDError("Error attempting to create row iterator").AddCause(err).Build(), nil)
+			}
+			cli.Println("")
+			cli.Println(doltdb.DoltConstViolTablePrefix + tableName)
+			if cvMap.Len() > 50 {
+				cli.Printf("Over 50 constraint violations were found. Please query '%s' to see them all.\n", doltdb.DoltConstViolTablePrefix+tableName)
+			} else {
+				err = commands.PrettyPrintResults(sql.NewEmptyContext(), commands.FormatTabular, sqlSchema, rowIter, false)
+				if err != nil {
+					return commands.HandleVErrAndExitCode(errhand.BuildDError("Error outputting rows").AddCause(err).Build(), nil)
+				}
+			}
+		}
+		return 1
+	}
+	return 0
+}

go/cmd/dolt/commands/push.go

@@ -199,6 +199,10 @@ func parsePushArgs(ctx context.Context, apr *argparser.ArgParseResults, dEnv *en
 		verr = errhand.BuildDError("").SetPrintUsage().Build()
 	}
 
+	if verr != nil {
+		return nil, verr
+	}
+
 	remote, remoteOK = remotes[remoteName]
 
 	if !remoteOK {

go/cmd/dolt/commands/verify_constraints.go

@@ -1,128 +0,0 @@
-// Copyright 2020 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 commands
-
-import (
-	"context"
-	"errors"
-	"strings"
-
-	"github.com/dolthub/dolt/go/cmd/dolt/cli"
-	"github.com/dolthub/dolt/go/cmd/dolt/errhand"
-	"github.com/dolthub/dolt/go/libraries/doltcore/env"
-	"github.com/dolthub/dolt/go/libraries/utils/argparser"
-	"github.com/dolthub/dolt/go/libraries/utils/filesys"
-)
-
-var verifyConstraintsDocs = cli.CommandDocumentationContent{
-	ShortDesc: `Verifies a table's constraints'`,
-	LongDesc:  `This command verifies that the defined constraints on the given table(s)—such as a foreign key—are correct and satisfied.`,
-	Synopsis:  []string{`{{.LessThan}}table{{.GreaterThan}}...`},
-}
-
-type VerifyConstraintsCmd struct{}
-
-var _ cli.Command = VerifyConstraintsCmd{}
-
-func (cmd VerifyConstraintsCmd) Name() string {
-	return "verify-constraints"
-}
-
-func (cmd VerifyConstraintsCmd) Description() string {
-	return "Command to verify that the constraints on the given table(s) are satisfied."
-}
-
-func (cmd VerifyConstraintsCmd) CreateMarkdown(fs filesys.Filesys, path, commandStr string) error {
-	return nil
-}
-
-func (cmd VerifyConstraintsCmd) createArgParser() *argparser.ArgParser {
-	ap := argparser.NewArgParser()
-	ap.ArgListHelp = append(ap.ArgListHelp, [2]string{"table", "The table to check constraints on."})
-	return ap
-}
-
-func (cmd VerifyConstraintsCmd) Exec(ctx context.Context, commandStr string, args []string, dEnv *env.DoltEnv) int {
-	ap := cmd.createArgParser()
-	help, usage := cli.HelpAndUsagePrinters(cli.GetCommandDocumentation(commandStr, verifyConstraintsDocs, ap))
-	apr := cli.ParseArgsOrDie(ap, args, help)
-
-	if apr.NArg() == 0 {
-		usage()
-		return 0
-	}
-	tableNames := apr.Args()
-	working, err := dEnv.WorkingRoot(ctx)
-	if err != nil {
-		return HandleVErrAndExitCode(errhand.BuildDError("Unable to get working.").AddCause(err).Build(), nil)
-	}
-	fkColl, err := working.GetForeignKeyCollection(ctx)
-	if err != nil {
-		return HandleVErrAndExitCode(errhand.BuildDError("Unable to get foreign keys.").AddCause(err).Build(), nil)
-	}
-
-	var accumulatedConstraintErrors []string
-	for _, givenTableName := range tableNames {
-		tbl, tableName, ok, err := working.GetTableInsensitive(ctx, givenTableName)
-		if err != nil {
-			return HandleVErrAndExitCode(errhand.BuildDError("Unable to get table %s.", givenTableName).AddCause(err).Build(), nil)
-		}
-		if !ok {
-			return HandleVErrAndExitCode(errhand.BuildDError("Table %s does not exist.", givenTableName).Build(), nil)
-		}
-		tblSch, err := tbl.GetSchema(ctx)
-		if err != nil {
-			return HandleVErrAndExitCode(errhand.BuildDError("Unable to get schema for %s.", tableName).AddCause(err).Build(), nil)
-		}
-		fks, _ := fkColl.KeysForTable(tableName)
-
-		for _, fk := range fks {
-			childIdx := tblSch.Indexes().GetByName(fk.TableIndex)
-			childIdxRowData, err := tbl.GetIndexRowData(ctx, fk.TableIndex)
-			if err != nil {
-				return HandleVErrAndExitCode(errhand.BuildDError("Unable to get index data for %s.", fk.TableIndex).AddCause(err).Build(), nil)
-			}
-
-			parentTbl, _, ok, err := working.GetTableInsensitive(ctx, fk.ReferencedTableName)
-			if err != nil {
-				return HandleVErrAndExitCode(errhand.BuildDError("Unable to get table %s.", fk.ReferencedTableName).AddCause(err).Build(), nil)
-			}
-			if !ok {
-				return HandleVErrAndExitCode(errhand.BuildDError("Table %s does not exist.", fk.ReferencedTableName).Build(), nil)
-			}
-			parentTblSch, err := parentTbl.GetSchema(ctx)
-			if err != nil {
-				return HandleVErrAndExitCode(errhand.BuildDError("Unable to get schema for %s.", fk.ReferencedTableName).AddCause(err).Build(), nil)
-			}
-			parentIdx := parentTblSch.Indexes().GetByName(fk.ReferencedTableIndex)
-			parentIdxRowData, err := parentTbl.GetIndexRowData(ctx, fk.ReferencedTableIndex)
-			if err != nil {
-				return HandleVErrAndExitCode(errhand.BuildDError("Unable to get index data for %s.", fk.ReferencedTableIndex).AddCause(err).Build(), nil)
-			}
-
-			err = fk.ValidateData(ctx, childIdxRowData, parentIdxRowData, childIdx, parentIdx)
-			if err != nil {
-				accumulatedConstraintErrors = append(accumulatedConstraintErrors, err.Error())
-			}
-		}
-	}
-
-	if len(accumulatedConstraintErrors) > 0 {
-		dErr := errhand.BuildDError("All constraints are not satisfied.")
-		dErr = dErr.AddCause(errors.New(strings.Join(accumulatedConstraintErrors, "\n")))
-		return HandleVErrAndExitCode(dErr.Build(), nil)
-	}
-	return 0
-}

go/cmd/dolt/dolt.go

@@ -36,6 +36,7 @@ import (
 	"github.com/dolthub/dolt/go/cmd/dolt/commands"
 	"github.com/dolthub/dolt/go/cmd/dolt/commands/cnfcmds"
 	"github.com/dolthub/dolt/go/cmd/dolt/commands/credcmds"
+	"github.com/dolthub/dolt/go/cmd/dolt/commands/cvcmds"
 	"github.com/dolthub/dolt/go/cmd/dolt/commands/indexcmds"
 	"github.com/dolthub/dolt/go/cmd/dolt/commands/schcmds"
 	"github.com/dolthub/dolt/go/cmd/dolt/commands/sqlserver"
@@ -83,6 +84,7 @@ var doltCommand = cli.NewSubCommandHandler("dolt", "it's git for data", []cli.Co
 	schcmds.Commands,
 	tblcmds.Commands,
 	cnfcmds.Commands,
+	cvcmds.Commands,
 	commands.SendMetricsCmd{},
 	dumpDocsCommand,
 	commands.MigrateCmd{},
@@ -90,7 +92,7 @@ var doltCommand = cli.NewSubCommandHandler("dolt", "it's git for data", []cli.Co
 	commands.ReadTablesCmd{},
 	commands.GarbageCollectionCmd{},
 	commands.FilterBranchCmd{},
-	commands.VerifyConstraintsCmd{},
+	cvcmds.VerifyConstraintsCmd{},
 	commands.MergeBaseCmd{},
 	commands.RootsCmd{},
 })

go/libraries/doltcore/doltdb/foreign_key_coll.go

@@ -34,10 +34,49 @@ import (
 	"github.com/dolthub/dolt/go/store/types"
 )
 
+// ForeignKeyCollection represents the collection of foreign keys for a root value.
 type ForeignKeyCollection struct {
 	foreignKeys map[string]ForeignKey
 }
 
+// ForeignKeyViolationError represents a set of foreign key violations for a table.
+type ForeignKeyViolationError struct {
+	ForeignKey    ForeignKey
+	Schema        schema.Schema
+	ViolationRows []row.Row
+}
+
+// Error implements the interface error.
+func (f *ForeignKeyViolationError) Error() string {
+	if len(f.ViolationRows) == 0 {
+		return "no violations were found, should not be an error"
+	}
+	sb := strings.Builder{}
+	const earlyTerminationLimit = 50
+	terminatedEarly := false
+	for i := range f.ViolationRows {
+		if i >= earlyTerminationLimit {
+			terminatedEarly = true
+			break
+		}
+		key, _ := f.ViolationRows[i].NomsMapKey(f.Schema).Value(context.Background())
+		val, _ := f.ViolationRows[i].NomsMapValue(f.Schema).Value(context.Background())
+		valSlice, _ := val.(types.Tuple).AsSlice()
+		all, _ := key.(types.Tuple).Append(valSlice...)
+		str, _ := types.EncodedValue(context.Background(), all)
+		sb.WriteRune('\n')
+		sb.WriteString(str)
+	}
+	if terminatedEarly {
+		return fmt.Sprintf("foreign key violations on `%s`.`%s`:%s\n%d more violations are not being displayed",
+			f.ForeignKey.Name, f.ForeignKey.TableName, sb.String(), len(f.ViolationRows)-earlyTerminationLimit)
+	} else {
+		return fmt.Sprintf("foreign key violations on `%s`.`%s`:%s", f.ForeignKey.Name, f.ForeignKey.TableName, sb.String())
+	}
+}
+
+var _ error = (*ForeignKeyViolationError)(nil)
+
 type ForeignKeyReferenceOption byte
 
 const (
@@ -460,8 +499,13 @@ func (fkc *ForeignKeyCollection) copy() *ForeignKeyCollection {
 }
 
 // ValidateData ensures that the foreign key is valid by comparing the index data from the given table
-// against the index data from the referenced table.
-func (fk ForeignKey) ValidateData(ctx context.Context, childIdx, parentIdx types.Map, childDef, parentDef schema.Index) error {
+// against the index data from the referenced table. Returns an error for each violation.
+func (fk ForeignKey) ValidateData(
+	ctx context.Context,
+	childSch schema.Schema,
+	childRowData, childIdxData, parentIdxData types.Map,
+	childDef, parentDef schema.Index,
+) error {
 	if fk.ReferencedTableIndex != parentDef.Name() {
 		return fmt.Errorf("cannot validate data as wrong referenced index was given: expected `%s` but received `%s`",
 			fk.ReferencedTableIndex, parentDef.Name())
@@ -484,11 +528,12 @@ func (fk ForeignKey) ValidateData(ctx context.Context, childIdx, parentIdx types
 		return err
 	}
 
-	rdr, err := noms.NewNomsMapReader(ctx, childIdx, childDef.Schema())
+	rdr, err := noms.NewNomsMapReader(ctx, childIdxData, childDef.Schema())
 	if err != nil {
 		return err
 	}
 
+	var violatingRows []row.Row
 	for {
 		childIdxRow, err := rdr.ReadRow(ctx)
 		if err == io.EOF {
@@ -527,7 +572,7 @@ func (fk ForeignKey) ValidateData(ctx context.Context, childIdx, parentIdx types
 			return err
 		}
 
-		indexIter := noms.NewNomsRangeReader(parentDef.Schema(), parentIdx,
+		indexIter := noms.NewNomsRangeReader(parentDef.Schema(), parentIdxData,
 			[]*noms.ReadRange{{Start: partial, Inclusive: true, Reverse: false, Check: func(tuple types.Tuple) (bool, error) {
 				return tuple.StartsWith(partial), nil
 			}}},
@@ -537,12 +582,39 @@ func (fk ForeignKey) ValidateData(ctx context.Context, childIdx, parentIdx types
 		case nil:
 			continue // parent table contains child key
 		case io.EOF:
-			indexKeyStr, _ := types.EncodedValue(ctx, partial)
-			return fmt.Errorf("foreign key violation on `%s`.`%s`: `%s`", fk.Name, fk.TableName, indexKeyStr)
+			childFullKey, err := childIdxRow.NomsMapKey(childDef.Schema()).Value(ctx)
+			if err != nil {
+				return err
+			}
+			childKey, err := childDef.ToTableTuple(ctx, childFullKey.(types.Tuple), childIdxRow.Format())
+			if err != nil {
+				return err
+			}
+			childVal, ok, err := childRowData.MaybeGetTuple(ctx, childKey)
+			if err != nil {
+				return err
+			}
+			if !ok {
+				childKeyStr, _ := types.EncodedValue(ctx, childKey)
+				return fmt.Errorf("could not find row value for key %s on table `%s`", childKeyStr, fk.TableName)
+			}
+			childRow, err := row.FromNoms(childSch, childKey, childVal)
+			if err != nil {
+				return err
+			}
+			violatingRows = append(violatingRows, childRow)
 		default:
 			return err
 		}
 	}
 
-	return nil
+	if len(violatingRows) == 0 {
+		return nil
+	} else {
+		return &ForeignKeyViolationError{
+			ForeignKey:    fk,
+			Schema:        childSch,
+			ViolationRows: violatingRows,
+		}
+	}
 }

go/libraries/doltcore/doltdb/table.go

@@ -20,8 +20,11 @@ import (
 	"fmt"
 	"regexp"
 
+	"github.com/dolthub/go-mysql-server/sql"
+
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema/encoding"
+	"github.com/dolthub/dolt/go/libraries/doltcore/schema/typeinfo"
 	"github.com/dolthub/dolt/go/store/hash"
 	"github.com/dolthub/dolt/go/store/types"
 )
@@ -271,6 +274,35 @@ func (t *Table) GetConflictSchemas(ctx context.Context) (base, sch, mergeSch sch
 	return nil, nil, nil, ErrNoConflicts
 }
 
+// GetConstraintViolationsSchema returns the schema for the dolt_constraint_violations system table belonging to this
+// table.
+func (t *Table) GetConstraintViolationsSchema(ctx context.Context) (schema.Schema, error) {
+	sch, err := t.GetSchema(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	typeType, err := typeinfo.FromSqlType(
+		sql.MustCreateEnumType([]string{"foreign key", "unique index", "check constraint"}, sql.Collation_Default))
+	if err != nil {
+		return nil, err
+	}
+	typeCol, err := schema.NewColumnWithTypeInfo("violation_type", schema.DoltConstraintViolationsTypeTag, typeType, true, "", false, "")
+	if err != nil {
+		return nil, err
+	}
+	infoCol, err := schema.NewColumnWithTypeInfo("violation_info", schema.DoltConstraintViolationsInfoTag, typeinfo.JSONType, false, "", false, "")
+	if err != nil {
+		return nil, err
+	}
+
+	colColl := schema.NewColCollection()
+	colColl = colColl.Append(typeCol)
+	colColl = colColl.Append(sch.GetAllCols().GetColumns()...)
+	colColl = colColl.Append(infoCol)
+	return schema.SchemaFromCols(colColl)
+}
+
 // GetConstraintViolations returns a map of all constraint violations for this table, along with a bool indicating
 // whether the table has any violations.
 func (t *Table) GetConstraintViolations(ctx context.Context) (types.Map, error) {

go/libraries/doltcore/env/actions/commit.go

@@ -16,7 +16,6 @@ package actions
 
 import (
 	"context"
-	"fmt"
 	"sort"
 	"time"
 
@@ -183,71 +182,6 @@ func CommitStaged(ctx context.Context, roots doltdb.Roots, dbData env.DbData, pr
 	return c, nil
 }
 
-func ValidateForeignKeysOnCommit(ctx context.Context, srt *doltdb.RootValue, stagedTblNames []string) (*doltdb.RootValue, error) {
-	// Validate schemas
-	srt, err := srt.ValidateForeignKeysOnSchemas(ctx)
-	if err != nil {
-		return nil, err
-	}
-	// Validate data
-	//TODO: make this more efficient, perhaps by leveraging diffs?
-	fkColl, err := srt.GetForeignKeyCollection(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	fksToCheck := make(map[string]doltdb.ForeignKey)
-	for _, tblName := range stagedTblNames {
-		declaredFk, referencedByFk := fkColl.KeysForTable(tblName)
-		for _, fk := range declaredFk {
-			fksToCheck[fk.Name] = fk
-		}
-		for _, fk := range referencedByFk {
-			fksToCheck[fk.Name] = fk
-		}
-	}
-
-	for _, fk := range fksToCheck {
-		childTbl, _, ok, err := srt.GetTableInsensitive(ctx, fk.TableName)
-		if err != nil {
-			return nil, err
-		}
-		if !ok {
-			return nil, fmt.Errorf("foreign key '%s' references missing table '%s'", fk.Name, fk.TableName)
-		}
-		childSch, err := childTbl.GetSchema(ctx)
-		if err != nil {
-			return nil, err
-		}
-		childIdx := childSch.Indexes().GetByName(fk.TableIndex)
-		childIdxRowData, err := childTbl.GetIndexRowData(ctx, fk.TableIndex)
-		if err != nil {
-			return nil, err
-		}
-		parentTbl, _, ok, err := srt.GetTableInsensitive(ctx, fk.ReferencedTableName)
-		if err != nil {
-			return nil, err
-		}
-		if !ok {
-			return nil, fmt.Errorf("foreign key '%s' references missing table '%s'", fk.Name, fk.ReferencedTableName)
-		}
-		parentTblSch, err := parentTbl.GetSchema(ctx)
-		if err != nil {
-			return nil, err
-		}
-		parentIdx := parentTblSch.Indexes().GetByName(fk.ReferencedTableIndex)
-		parentIdxRowData, err := parentTbl.GetIndexRowData(ctx, fk.ReferencedTableIndex)
-		if err != nil {
-			return nil, err
-		}
-		err = fk.ValidateData(ctx, childIdxRowData, parentIdxRowData, childIdx, parentIdx)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return srt, nil
-}
-
 // TimeSortedCommits returns a reverse-chronological (latest-first) list of the most recent `n` ancestors of `commit`.
 // Passing a negative value for `n` will result in all ancestors being returned.
 func TimeSortedCommits(ctx context.Context, ddb *doltdb.DoltDB, commit *doltdb.Commit, n int) ([]*doltdb.Commit, error) {

go/libraries/doltcore/merge/constraint_violations.go

@@ -31,6 +31,7 @@ import (
 	json2 "github.com/dolthub/dolt/go/libraries/doltcore/sqle/json"
 	"github.com/dolthub/dolt/go/libraries/doltcore/table"
 	"github.com/dolthub/dolt/go/libraries/doltcore/table/typed/noms"
+	"github.com/dolthub/dolt/go/libraries/utils/set"
 	diff2 "github.com/dolthub/dolt/go/store/diff"
 	"github.com/dolthub/dolt/go/store/types"
 )
@@ -56,80 +57,93 @@ const (
 )
 
 // AddConstraintViolations adds all constraint violations to each table.
-func AddConstraintViolations(ctx context.Context, newRoot, baseRoot *doltdb.RootValue) (*doltdb.RootValue, error) {
+func AddConstraintViolations(ctx context.Context, newRoot, baseRoot *doltdb.RootValue, tables *set.StrSet) (*doltdb.RootValue, *set.StrSet, error) {
 	fkColl, err := newRoot.GetForeignKeyCollection(ctx)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
+	foundViolationsSet := set.NewStrSet(nil)
 	for _, foreignKey := range fkColl.AllKeys() {
+		if tables.Size() != 0 && !tables.Contains(foreignKey.TableName) {
+			continue
+		}
+
 		postParent, ok, err := newConstraintViolationsLoadedTable(ctx, foreignKey.ReferencedTableName, foreignKey.ReferencedTableIndex, newRoot)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		if !ok {
-			return nil, fmt.Errorf("foreign key %s should have index %s on table %s but it cannot be found",
+			return nil, nil, fmt.Errorf("foreign key %s should have index %s on table %s but it cannot be found",
 				foreignKey.Name, foreignKey.ReferencedTableIndex, foreignKey.ReferencedTableName)
 		}
 
 		postChild, ok, err := newConstraintViolationsLoadedTable(ctx, foreignKey.TableName, foreignKey.TableIndex, newRoot)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		if !ok {
-			return nil, fmt.Errorf("foreign key %s should have index %s on table %s but it cannot be found",
+			return nil, nil, fmt.Errorf("foreign key %s should have index %s on table %s but it cannot be found",
 				foreignKey.Name, foreignKey.TableIndex, foreignKey.TableName)
 		}
 
+		foundViolations := false
 		preParent, _, err := newConstraintViolationsLoadedTable(ctx, foreignKey.ReferencedTableName, "", baseRoot)
 		if err != nil {
 			if err != doltdb.ErrTableNotFound {
-				return nil, err
+				return nil, nil, err
 			}
 			// Parent does not exist in the ancestor so we use an empty map
 			emptyMap, err := types.NewMap(ctx, postParent.Table.ValueReadWriter())
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
-			postChild.Table, err = parentFkConstraintViolations(ctx, foreignKey, postParent, postChild, postParent.Schema, emptyMap)
+			postChild.Table, foundViolations, err = parentFkConstraintViolations(ctx, foreignKey, postParent, postChild, postParent.Schema, emptyMap)
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
 		} else {
 			// Parent exists in the ancestor
-			postChild.Table, err = parentFkConstraintViolations(ctx, foreignKey, postParent, postChild, preParent.Schema, preParent.RowData)
+			postChild.Table, foundViolations, err = parentFkConstraintViolations(ctx, foreignKey, postParent, postChild, preParent.Schema, preParent.RowData)
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
 		}
 
 		preChild, _, err := newConstraintViolationsLoadedTable(ctx, foreignKey.TableName, "", baseRoot)
 		if err != nil {
 			if err != doltdb.ErrTableNotFound {
-				return nil, err
+				return nil, nil, err
 			}
+			innerFoundViolations := false
 			// Child does not exist in the ancestor so we use an empty map
 			emptyMap, err := types.NewMap(ctx, postChild.Table.ValueReadWriter())
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
-			postChild.Table, err = childFkConstraintViolations(ctx, foreignKey, postParent, postChild, postChild.Schema, emptyMap)
+			postChild.Table, innerFoundViolations, err = childFkConstraintViolations(ctx, foreignKey, postParent, postChild, postChild.Schema, emptyMap)
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
+			foundViolations = foundViolations || innerFoundViolations
 		} else {
 			// Child exists in the ancestor
-			postChild.Table, err = childFkConstraintViolations(ctx, foreignKey, postParent, postChild, preChild.Schema, preChild.RowData)
+			innerFoundViolations := false
+			postChild.Table, innerFoundViolations, err = childFkConstraintViolations(ctx, foreignKey, postParent, postChild, preChild.Schema, preChild.RowData)
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
+			foundViolations = foundViolations || innerFoundViolations
 		}
 
 		newRoot, err = newRoot.PutTable(ctx, postChild.TableName, postChild.Table)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
+		}
+		if foundViolations {
+			foundViolationsSet.Add(postChild.TableName)
 		}
 	}
-	return newRoot, nil
+	return newRoot, foundViolationsSet, nil
 }
 
 // parentFkConstraintViolations processes foreign key constraint violations for the parent in a foreign key.
@@ -139,12 +153,13 @@ func parentFkConstraintViolations(
 	postParent, postChild *constraintViolationsLoadedTable,
 	preParentSch schema.Schema,
 	preParentRowData types.Map,
-) (*doltdb.Table, error) {
+) (*doltdb.Table, bool, error) {
+	foundViolations := false
 	postParentIndexTags := postParent.Index.IndexedColumnTags()
 	postChildIndexTags := postChild.Index.IndexedColumnTags()
 	postChildCVMap, err := postChild.Table.GetConstraintViolations(ctx)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 	postChildCVMapEditor := postChildCVMap.Edit()
 
@@ -154,11 +169,11 @@ func parentFkConstraintViolations(
 	for {
 		diffSlice, hasMore, err := differ.GetDiffs(1, 10*time.Second)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 		if len(diffSlice) != 1 {
 			if hasMore {
-				return nil, fmt.Errorf("no diff returned but should have errored earlier")
+				return nil, false, fmt.Errorf("no diff returned but should have errored earlier")
 			}
 			break
 		}
@@ -167,7 +182,7 @@ func parentFkConstraintViolations(
 		case types.DiffChangeRemoved, types.DiffChangeModified:
 			postParentRow, err := row.FromNoms(postParent.Schema, rowDiff.KeyValue.(types.Tuple), rowDiff.OldValue.(types.Tuple))
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 			hasNulls := false
 			for _, tag := range postParentIndexTags {
@@ -182,7 +197,7 @@ func parentFkConstraintViolations(
 
 			postParentIndexPartialKey, err := row.ReduceToIndexPartialKey(postParent.Index, postParentRow)
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 
 			shouldContinue, err := func() (bool, error) {
@@ -205,20 +220,24 @@ func parentFkConstraintViolations(
 
 			postParentIndexPartialKeySlice, err := postParentIndexPartialKey.AsSlice()
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 			for i := 0; i < len(postChildIndexTags); i++ {
 				postParentIndexPartialKeySlice[2*i] = types.Uint(postChildIndexTags[i])
 			}
 			postChildIndexPartialKey, err := types.NewTuple(postChild.Table.Format(), postParentIndexPartialKeySlice...)
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
-			err = parentFkConstraintViolationsProcess(ctx, foreignKey, postParent, postChild, postChildIndexPartialKey, postChildCVMapEditor)
+			changeViolates, err := parentFkConstraintViolationsProcess(ctx, foreignKey, postParent, postChild, postChildIndexPartialKey, postChildCVMapEditor)
+			if err != nil {
+				return nil, false, err
+			}
+			foundViolations = foundViolations || changeViolates
 		case types.DiffChangeAdded:
 			// We don't do anything if a parent row was added
 		default:
-			return nil, fmt.Errorf("unknown diff change type")
+			return nil, false, fmt.Errorf("unknown diff change type")
 		}
 		if !hasMore {
 			break
@@ -227,9 +246,10 @@ func parentFkConstraintViolations(
 
 	postChildCVMap, err = postChildCVMapEditor.Map(ctx)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
-	return postChild.Table.SetConstraintViolations(ctx, postChildCVMap)
+	updatedTbl, err := postChild.Table.SetConstraintViolations(ctx, postChildCVMap)
+	return updatedTbl, foundViolations, err
 }
 
 // parentFkConstraintViolationsProcess handles processing the reference options on a child, or creating a violation if
@@ -240,7 +260,8 @@ func parentFkConstraintViolationsProcess(
 	postParent, postChild *constraintViolationsLoadedTable,
 	postChildIndexPartialKey types.Tuple,
 	postChildCVMapEditor *types.MapEditor,
-) error {
+) (bool, error) {
+	foundViolation := false
 	mapIter := noms.NewNomsRangeReader(postChild.IndexSchema, postChild.IndexData,
 		[]*noms.ReadRange{{Start: postChildIndexPartialKey, Inclusive: true, Reverse: false, Check: func(tuple types.Tuple) (bool, error) {
 			return tuple.StartsWith(postChildIndexPartialKey), nil
@@ -251,33 +272,34 @@ func parentFkConstraintViolationsProcess(
 	for postChildIndexRow, err = mapIter.ReadRow(ctx); err == nil; postChildIndexRow, err = mapIter.ReadRow(ctx) {
 		postChildIndexKey, err := postChildIndexRow.NomsMapKey(postChild.IndexSchema).Value(ctx)
 		if err != nil {
-			return err
+			return false, err
 		}
 		postChildRowKey, err := postChild.Index.ToTableTuple(ctx, postChildIndexKey.(types.Tuple), postChild.Table.Format())
 		if err != nil {
-			return err
+			return false, err
 		}
 		postChildRowVal, ok, err := postChild.RowData.MaybeGetTuple(ctx, postChildRowKey)
 		if err != nil {
-			return err
+			return false, err
 		}
 		if !ok {
-			return fmt.Errorf("index %s on %s contains data that table does not", foreignKey.TableIndex, foreignKey.TableName)
+			return false, fmt.Errorf("index %s on %s contains data that table does not", foreignKey.TableIndex, foreignKey.TableName)
 		}
 		vInfo, err := foreignKeyCVJson(ctx, foreignKey, postChild.Table.ValueReadWriter(), postChild.Schema, postParent.Schema)
 		if err != nil {
-			return err
+			return false, err
 		}
 		cvKey, cvVal, err := toConstraintViolationRow(ctx, cvType_ForeignKey, vInfo, postChildRowKey, postChildRowVal)
 		if err != nil {
-			return err
+			return false, err
 		}
 		postChildCVMapEditor.Set(cvKey, cvVal)
+		foundViolation = true
 	}
 	if err != io.EOF {
-		return err
+		return false, err
 	}
-	return nil
+	return foundViolation, nil
 }
 
 // childFkConstraintViolations processes foreign key constraint violations for the child in a foreign key.
@@ -287,12 +309,13 @@ func childFkConstraintViolations(
 	postParent, postChild *constraintViolationsLoadedTable,
 	preChildSch schema.Schema,
 	preChildRowData types.Map,
-) (*doltdb.Table, error) {
+) (*doltdb.Table, bool, error) {
+	foundViolations := false
 	postParentIndexTags := postParent.Index.IndexedColumnTags()
 	postChildIndexTags := postChild.Index.IndexedColumnTags()
 	postChildCVMap, err := postChild.Table.GetConstraintViolations(ctx)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 	postChildCVMapEditor := postChildCVMap.Edit()
 
@@ -302,11 +325,11 @@ func childFkConstraintViolations(
 	for {
 		diffSlice, hasMore, err := differ.GetDiffs(1, 10*time.Second)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 		if len(diffSlice) != 1 {
 			if hasMore {
-				return nil, fmt.Errorf("no diff returned but should have errored earlier")
+				return nil, false, fmt.Errorf("no diff returned but should have errored earlier")
 			}
 			break
 		}
@@ -315,7 +338,7 @@ func childFkConstraintViolations(
 		case types.DiffChangeAdded, types.DiffChangeModified:
 			postChildRow, err := row.FromNoms(postChild.Schema, rowDiff.KeyValue.(types.Tuple), rowDiff.NewValue.(types.Tuple))
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 			hasNulls := false
 			for _, tag := range postChildIndexTags {
@@ -330,27 +353,28 @@ func childFkConstraintViolations(
 
 			postChildIndexPartialKey, err := row.ReduceToIndexPartialKey(postChild.Index, postChildRow)
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 			postChildIndexPartialKeySlice, err := postChildIndexPartialKey.AsSlice()
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 			for i := 0; i < len(postParentIndexTags); i++ {
 				postChildIndexPartialKeySlice[2*i] = types.Uint(postParentIndexTags[i])
 			}
 			parentPartialKey, err := types.NewTuple(postChild.Table.Format(), postChildIndexPartialKeySlice...)
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
-			err = childFkConstraintViolationsProcess(ctx, foreignKey, postParent, postChild, rowDiff, parentPartialKey, postChildCVMapEditor)
+			diffViolates, err := childFkConstraintViolationsProcess(ctx, foreignKey, postParent, postChild, rowDiff, parentPartialKey, postChildCVMapEditor)
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
+			foundViolations = foundViolations || diffViolates
 		case types.DiffChangeRemoved:
 			// We don't do anything if a child row was removed
 		default:
-			return nil, fmt.Errorf("unknown diff change type")
+			return nil, false, fmt.Errorf("unknown diff change type")
 		}
 		if !hasMore {
 			break
@@ -358,9 +382,10 @@ func childFkConstraintViolations(
 	}
 	postChildCVMap, err = postChildCVMapEditor.Map(ctx)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
-	return postChild.Table.SetConstraintViolations(ctx, postChildCVMap)
+	updatedTbl, err := postChild.Table.SetConstraintViolations(ctx, postChildCVMap)
+	return updatedTbl, foundViolations, err
 }
 
 // childFkConstraintViolationsProcess handles processing the constraint violations for the child of a foreign key.
@@ -371,7 +396,7 @@ func childFkConstraintViolationsProcess(
 	rowDiff *diff2.Difference,
 	parentPartialKey types.Tuple,
 	postChildCVMapEditor *types.MapEditor,
-) error {
+) (bool, error) {
 	var mapIter table.TableReadCloser = noms.NewNomsRangeReader(postParent.IndexSchema, postParent.IndexData,
 		[]*noms.ReadRange{{Start: parentPartialKey, Inclusive: true, Reverse: false, Check: func(tuple types.Tuple) (bool, error) {
 			return tuple.StartsWith(parentPartialKey), nil
@@ -380,19 +405,20 @@ func childFkConstraintViolationsProcess(
 	// If the row exists in the parent, then we don't need to do anything
 	if _, err := mapIter.ReadRow(ctx); err != nil {
 		if err != io.EOF {
-			return err
+			return false, err
 		}
 		vInfo, err := foreignKeyCVJson(ctx, foreignKey, postChild.Table.ValueReadWriter(), postChild.Schema, postParent.Schema)
 		if err != nil {
-			return err
+			return false, err
 		}
 		cvKey, cvVal, err := toConstraintViolationRow(ctx, cvType_ForeignKey, vInfo, rowDiff.KeyValue.(types.Tuple), rowDiff.NewValue.(types.Tuple))
 		if err != nil {
-			return err
+			return false, err
 		}
 		postChildCVMapEditor.Set(cvKey, cvVal)
+		return true, nil
 	}
-	return nil
+	return false, nil
 }
 
 // newConstraintViolationsLoadedTable returns a *constraintViolationsLoadedTable. Returns false if the table was loaded
@@ -439,30 +465,31 @@ func newConstraintViolationsLoadedTable(ctx context.Context, tblName, idxName st
 
 // toConstraintViolationRow converts the given key and value tuples into ones suitable to add to a constraint violation map.
 func toConstraintViolationRow(ctx context.Context, vType cvType, vInfo types.JSON, k, v types.Tuple) (types.Tuple, types.Tuple, error) {
-	constraintViolationVals := []types.Value{types.Uint(schema.DoltConstraintViolationsTypeTag), types.Uint(vType)}
+	constraintViolationKeyVals := []types.Value{types.Uint(schema.DoltConstraintViolationsTypeTag), types.Uint(vType)}
 	keySlice, err := k.AsSlice()
 	if err != nil {
 		emptyTuple := types.EmptyTuple(k.Format())
 		return emptyTuple, emptyTuple, err
 	}
-	constraintViolationVals = append(constraintViolationVals, keySlice...)
-	valSlice, err := v.AsSlice()
+	constraintViolationKeyVals = append(constraintViolationKeyVals, keySlice...)
+	constraintViolationKey, err := types.NewTuple(k.Format(), constraintViolationKeyVals...)
 	if err != nil {
 		emptyTuple := types.EmptyTuple(k.Format())
 		return emptyTuple, emptyTuple, err
 	}
-	constraintViolationVals = append(constraintViolationVals, valSlice...)
-	constraintViolationKey, err := types.NewTuple(k.Format(), constraintViolationVals...)
+
+	constraintViolationValVals, err := v.AsSlice()
 	if err != nil {
 		emptyTuple := types.EmptyTuple(k.Format())
 		return emptyTuple, emptyTuple, err
 	}
-	constraintViolationVal, err := types.NewTuple(v.Format(),
-		types.Uint(schema.DoltConstraintViolationsInfoTag), vInfo)
+	constraintViolationValVals = append(constraintViolationValVals, types.Uint(schema.DoltConstraintViolationsInfoTag), vInfo)
+	constraintViolationVal, err := types.NewTuple(v.Format(), constraintViolationValVals...)
 	if err != nil {
 		emptyTuple := types.EmptyTuple(k.Format())
 		return emptyTuple, emptyTuple, err
 	}
+
 	return constraintViolationKey, constraintViolationVal, nil
 }
 

go/libraries/doltcore/merge/merge.go

@@ -921,7 +921,7 @@ func MergeRoots(ctx context.Context, ourRoot, theirRoot, ancRoot *doltdb.RootVal
 		return nil, nil, err
 	}
 
-	newRoot, err = AddConstraintViolations(ctx, newRoot, ancRoot)
+	newRoot, _, err = AddConstraintViolations(ctx, newRoot, ancRoot, nil)
 	if err != nil {
 		return nil, nil, err
 	}

go/libraries/doltcore/sqle/dfunctions/init.go

@@ -30,6 +30,8 @@ var DoltFunctions = []sql.Function{
 	sql.FunctionN{Name: DoltMergeFuncName, Fn: NewDoltMergeFunc},
 	sql.Function0{Name: ActiveBranchFuncName, Fn: NewActiveBranchFunc},
 	sql.Function2{Name: DoltMergeBaseFuncName, Fn: NewMergeBase},
+	sql.FunctionN{Name: ConstraintsVerifyFuncName, Fn: NewConstraintsVerifyFunc},
+	sql.FunctionN{Name: ConstraintsVerifyAllFuncName, Fn: NewConstraintsVerifyAllFunc},
 }
 
 // These are the DoltFunctions that get exposed to Dolthub Api.

go/libraries/doltcore/sqle/dfunctions/verify_constraints.go

@@ -0,0 +1,156 @@
+// Copyright 2021 Dolthub, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dfunctions
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/dolthub/go-mysql-server/sql"
+	"github.com/dolthub/go-mysql-server/sql/expression"
+
+	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
+	"github.com/dolthub/dolt/go/libraries/doltcore/merge"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess"
+	"github.com/dolthub/dolt/go/libraries/utils/set"
+)
+
+const (
+	ConstraintsVerifyFuncName    = "constraints_verify"
+	ConstraintsVerifyAllFuncName = "constraints_verify_all"
+)
+
+// ConstraintsVerifyFunc represents the sql functions "verify_constraints" and "verify_constraints_all".
+type ConstraintsVerifyFunc struct {
+	expression.NaryExpression
+	isAll bool
+}
+
+var _ sql.Expression = (*ConstraintsVerifyFunc)(nil)
+
+// NewConstraintsVerifyFunc creates a new ConstraintsVerifyFunc expression that verifies the diff.
+func NewConstraintsVerifyFunc(ctx *sql.Context, args ...sql.Expression) (sql.Expression, error) {
+	return &ConstraintsVerifyFunc{expression.NaryExpression{ChildExpressions: args}, false}, nil
+}
+
+// NewConstraintsVerifyAllFunc creates a new ConstraintsVerifyFunc expression that verifies all rows.
+func NewConstraintsVerifyAllFunc(ctx *sql.Context, args ...sql.Expression) (sql.Expression, error) {
+	return &ConstraintsVerifyFunc{expression.NaryExpression{ChildExpressions: args}, true}, nil
+}
+
+// Eval implements the Expression interface.
+func (vc *ConstraintsVerifyFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, error) {
+	dbName := ctx.GetCurrentDatabase()
+	dSess := dsess.DSessFromSess(ctx.Session)
+	workingSet := dSess.WorkingSet(ctx, dbName)
+	workingRoot := workingSet.WorkingRoot()
+	var comparingRoot *doltdb.RootValue
+	var err error
+	if vc.isAll {
+		comparingRoot, err = doltdb.EmptyRootValue(ctx, workingRoot.VRW())
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		headCommit, err := dSess.GetHeadCommit(ctx, dbName)
+		if err != nil {
+			return nil, err
+		}
+		comparingRoot, err = headCommit.GetRootValue()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	tableSet := set.NewStrSet(nil)
+	for i, expr := range vc.ChildExpressions {
+		evaluatedVal, err := expr.Eval(ctx, row)
+		if err != nil {
+			return nil, err
+		}
+		val, ok := evaluatedVal.(string)
+		if !ok {
+			return nil, sql.ErrUnexpectedType.New(i, reflect.TypeOf(evaluatedVal))
+		}
+		_, tableName, ok, err := workingRoot.GetTableInsensitive(ctx, val)
+		if err != nil {
+			return nil, err
+		}
+		if !ok {
+			return nil, sql.ErrTableNotFound.New(tableName)
+		}
+		tableSet.Add(tableName)
+	}
+
+	newRoot, tablesWithViolations, err := merge.AddConstraintViolations(ctx, workingRoot, comparingRoot, tableSet)
+	if err != nil {
+		return nil, err
+	}
+	if tablesWithViolations.Size() == 0 {
+		return 1, nil
+	} else {
+		err = dSess.SetRoot(ctx, dbName, newRoot)
+		if err != nil {
+			return nil, err
+		}
+		return 0, nil
+	}
+}
+
+// String implements the Stringer interface.
+func (vc *ConstraintsVerifyFunc) String() string {
+	if vc.isAll {
+		return fmt.Sprint("CONSTRAINTS_VERIFY_ALL()")
+	} else {
+		return fmt.Sprint("CONSTRAINTS_VERIFY()")
+	}
+}
+
+// IsNullable implements the Expression interface.
+func (vc *ConstraintsVerifyFunc) IsNullable() bool {
+	return false
+}
+
+// Resolved implements the Expression interface.
+func (vc *ConstraintsVerifyFunc) Resolved() bool {
+	for _, expr := range vc.ChildExpressions {
+		if !expr.Resolved() {
+			return false
+		}
+	}
+	return true
+}
+
+func (vc *ConstraintsVerifyFunc) Type() sql.Type {
+	return sql.Int8
+}
+
+// Children implements the Expression interface.
+func (vc *ConstraintsVerifyFunc) Children() []sql.Expression {
+	exprs := make([]sql.Expression, len(vc.ChildExpressions))
+	for i := range exprs {
+		exprs[i] = vc.ChildExpressions[i]
+	}
+	return exprs
+}
+
+// WithChildren implements the Expression interface.
+func (vc *ConstraintsVerifyFunc) WithChildren(ctx *sql.Context, children ...sql.Expression) (sql.Expression, error) {
+	if vc.isAll {
+		return NewConstraintsVerifyAllFunc(ctx, children...)
+	} else {
+		return NewConstraintsVerifyFunc(ctx, children...)
+	}
+}

go/libraries/doltcore/sqle/dtables/constraint_violations.go

@@ -20,7 +20,6 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
 	"github.com/dolthub/dolt/go/libraries/doltcore/row"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
-	"github.com/dolthub/dolt/go/libraries/doltcore/schema/typeinfo"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/sqlutil"
 	"github.com/dolthub/dolt/go/store/types"
 )
@@ -30,7 +29,7 @@ import (
 type ConstraintViolationsTable struct {
 	tblName string
 	root    *doltdb.RootValue
-	dSch    schema.Schema
+	cvSch   schema.Schema
 	sqlSch  sql.Schema
 	tbl     *doltdb.Table
 	rs      RootSetter
@@ -49,44 +48,18 @@ func NewConstraintViolationsTable(ctx *sql.Context, tblName string, root *doltdb
 	} else if !ok {
 		return nil, sql.ErrTableNotFound.New(tblName)
 	}
-	sch, err := tbl.GetSchema(ctx)
+	cvSch, err := tbl.GetConstraintViolationsSchema(ctx)
 	if err != nil {
 		return nil, err
 	}
-
-	typeType, err := typeinfo.FromSqlType(
-		sql.MustCreateEnumType([]string{"foreign key", "unique index", "check constraint"}, sql.Collation_Default))
-	if err != nil {
-		return nil, err
-	}
-	typeCol, err := schema.NewColumnWithTypeInfo("violation_type", schema.DoltConstraintViolationsTypeTag, typeType, true, "", false, "")
-	if err != nil {
-		return nil, err
-	}
-	infoCol, err := schema.NewColumnWithTypeInfo("violation_info", schema.DoltConstraintViolationsInfoTag, typeinfo.JSONType, false, "", false, "")
-	if err != nil {
-		return nil, err
-	}
-
-	colColl := schema.NewColCollection()
-	colColl = colColl.Append(typeCol)
-	for _, col := range sch.GetAllCols().GetColumns() {
-		col.IsPartOfPK = true
-		colColl = colColl.Append(col)
-	}
-	colColl = colColl.Append(infoCol)
-	dSch, err := schema.SchemaFromCols(colColl)
-	if err != nil {
-		return nil, err
-	}
-	sqlSch, err := sqlutil.FromDoltSchema(doltdb.DoltConstViolTablePrefix+tblName, dSch)
+	sqlSch, err := sqlutil.FromDoltSchema(doltdb.DoltConstViolTablePrefix+tblName, cvSch)
 	if err != nil {
 		return nil, err
 	}
 	return &ConstraintViolationsTable{
 		tblName: tblName,
 		root:    root,
-		dSch:    dSch,
+		cvSch:   cvSch,
 		sqlSch:  sqlSch,
 		tbl:     tbl,
 		rs:      rs,
@@ -123,7 +96,7 @@ func (cvt *ConstraintViolationsTable) PartitionRows(ctx *sql.Context, part sql.P
 	if err != nil {
 		return nil, err
 	}
-	return &constraintViolationsIter{ctx, cvt.dSch, iter}, nil
+	return &constraintViolationsIter{ctx, cvt.cvSch, iter}, nil
 }
 
 // Deleter implements the interface sql.DeletableTable.
@@ -172,11 +145,11 @@ var _ sql.RowDeleter = (*constraintViolationsDeleter)(nil)
 
 // Delete implements the interface sql.RowDeleter.
 func (cvd *constraintViolationsDeleter) Delete(ctx *sql.Context, r sql.Row) error {
-	dRow, err := sqlutil.SqlRowToDoltRow(ctx, cvd.cvt.tbl.ValueReadWriter(), r, cvd.cvt.dSch)
+	dRow, err := sqlutil.SqlRowToDoltRow(ctx, cvd.cvt.tbl.ValueReadWriter(), r, cvd.cvt.cvSch)
 	if err != nil {
 		return err
 	}
-	key, err := dRow.NomsMapKey(cvd.cvt.dSch).Value(ctx)
+	key, err := dRow.NomsMapKey(cvd.cvt.cvSch).Value(ctx)
 	if err != nil {
 		return err
 	}

go/libraries/doltcore/sqle/sqlutil/sql_row.go

@@ -25,9 +25,45 @@ import (
 
 	"github.com/dolthub/dolt/go/libraries/doltcore/row"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
+	"github.com/dolthub/dolt/go/libraries/doltcore/table/typed/noms"
 	"github.com/dolthub/dolt/go/store/types"
 )
 
+type mapSqlIter struct {
+	ctx context.Context
+	nmr *noms.NomsMapReader
+	sch schema.Schema
+}
+
+var _ sql.RowIter = (*mapSqlIter)(nil)
+
+// Next implements the interface sql.RowIter.
+func (m *mapSqlIter) Next() (sql.Row, error) {
+	dRow, err := m.nmr.ReadRow(m.ctx)
+	if err != nil {
+		return nil, err
+	}
+	return DoltRowToSqlRow(dRow, m.sch)
+}
+
+// Close implements the interface sql.RowIter.
+func (m *mapSqlIter) Close(ctx *sql.Context) error {
+	return m.nmr.Close(ctx)
+}
+
+// MapToSqlIter returns a map reader that converts all rows to sql rows, creating a sql row iterator.
+func MapToSqlIter(ctx context.Context, sch schema.Schema, data types.Map) (sql.RowIter, error) {
+	mapReader, err := noms.NewNomsMapReader(ctx, data, sch)
+	if err != nil {
+		return nil, err
+	}
+	return &mapSqlIter{
+		ctx: ctx,
+		nmr: mapReader,
+		sch: sch,
+	}, nil
+}
+
 // DoltRowToSqlRow constructs a go-mysql-server sql.Row from a Dolt row.Row.
 func DoltRowToSqlRow(doltRow row.Row, sch schema.Schema) (sql.Row, error) {
 	colVals := make(sql.Row, sch.GetAllCols().Size())

go/libraries/doltcore/sqle/tables.go

@@ -1318,6 +1318,10 @@ func (t *AlterableDoltTable) CreateForeignKey(
 		return err
 	}
 
+	tableData, err := table.GetRowData(ctx)
+	if err != nil {
+		return err
+	}
 	tableIndexData, err := table.GetIndexRowData(ctx, tableIndex.Name())
 	if err != nil {
 		return err
@@ -1326,7 +1330,7 @@ func (t *AlterableDoltTable) CreateForeignKey(
 	if err != nil {
 		return err
 	}
-	err = foreignKey.ValidateData(ctx, tableIndexData, refTableIndexData, tableIndex, refTableIndex)
+	err = foreignKey.ValidateData(ctx, t.sch, tableData, tableIndexData, refTableIndexData, tableIndex, refTableIndex)
 	if err != nil {
 		return err
 	}

go/libraries/doltcore/table/editor/keyless_table_editor.go

@@ -285,6 +285,8 @@ func (kte *keylessTableEditor) StatementFinished(ctx context.Context, errored bo
 
 // SetConstraintViolation implements TableEditor.
 func (kte *keylessTableEditor) SetConstraintViolation(ctx context.Context, k types.LesserValuable, v types.Valuable) error {
+	kte.mu.Lock()
+	defer kte.mu.Unlock()
 	if kte.cvEditor == nil {
 		cvMap, err := kte.tbl.GetConstraintViolations(ctx)
 		if err != nil {
@@ -292,8 +294,6 @@ func (kte *keylessTableEditor) SetConstraintViolation(ctx context.Context, k typ
 		}
 		kte.cvEditor = cvMap.Edit()
 	}
-	kte.mu.Lock()
-	defer kte.mu.Unlock()
 	kte.cvEditor.Set(k, v)
 	return nil
 }

go/libraries/doltcore/table/editor/pk_table_editor.go

@@ -833,6 +833,10 @@ func (te *pkTableEditor) StatementFinished(ctx context.Context, errored bool) er
 
 // SetConstraintViolation implements TableEditor.
 func (te *pkTableEditor) SetConstraintViolation(ctx context.Context, k types.LesserValuable, v types.Valuable) error {
+	te.flushMutex.RLock()
+	defer te.flushMutex.RUnlock()
+	te.writeMutex.Lock()
+	defer te.writeMutex.Unlock()
 	if te.cvEditor == nil {
 		cvMap, err := te.t.GetConstraintViolations(ctx)
 		if err != nil {
@@ -840,10 +844,6 @@ func (te *pkTableEditor) SetConstraintViolation(ctx context.Context, k types.Les
 		}
 		te.cvEditor = cvMap.Edit()
 	}
-	te.flushMutex.RLock()
-	defer te.flushMutex.RUnlock()
-	te.writeMutex.Lock()
-	defer te.writeMutex.Unlock()
 	te.cvEditor.Set(k, v)
 	return nil
 }

go/libraries/utils/set/strset.go

@@ -72,6 +72,9 @@ func (s *StrSet) Remove(items ...string) {
 
 // Contains returns true if the item being checked is already in the set.
 func (s *StrSet) Contains(item string) bool {
+	if s == nil {
+		return false
+	}
 	if !s.caseSensitive {
 		item = strings.ToLower(item)
 	}
@@ -82,6 +85,9 @@ func (s *StrSet) Contains(item string) bool {
 
 // ContainsAll returns true if all the items being checked are already in the set.
 func (s *StrSet) ContainsAll(items []string) bool {
+	if s == nil {
+		return false
+	}
 	if !s.caseSensitive {
 		items = funcitr.MapStrings(items, strings.ToLower)
 	}
@@ -117,12 +123,18 @@ func (s *StrSet) Equals(other *StrSet) bool {
 
 // Size returns the number of unique elements in the set
 func (s *StrSet) Size() int {
+	if s == nil {
+		return 0
+	}
 	return len(s.items)
 }
 
 // AsSlice converts the set to a slice of strings. If this is an insensitive set the resulting slice will be lowercase
 // regardless of the case that was used when adding the string to the set.
 func (s *StrSet) AsSlice() []string {
+	if s == nil {
+		return nil
+	}
 	size := len(s.items)
 	sl := make([]string, size)
 
@@ -138,6 +150,9 @@ func (s *StrSet) AsSlice() []string {
 // AsSortedSlice converts the set to a slice of strings. If this is an insensitive set the resulting slice will be lowercase
 // regardless of the case that was used when adding the string to the set. The slice is sorted in ascending order.
 func (s *StrSet) AsSortedSlice() []string {
+	if s == nil {
+		return nil
+	}
 	slice := s.AsSlice()
 	sort.Slice(slice, func(i, j int) bool {
 		return slice[i] < slice[j]
@@ -148,6 +163,9 @@ func (s *StrSet) AsSortedSlice() []string {
 // Iterate accepts a callback which will be called once for each element in the set until all items have been
 // exhausted or callback returns false.
 func (s *StrSet) Iterate(callBack func(string) (cont bool)) {
+	if s == nil {
+		return
+	}
 	for k := range s.items {
 		if !callBack(k) {
 			break

go/store/types/tuple.go

@@ -535,7 +535,7 @@ func (t Tuple) Set(n uint64, v Value) (Tuple, error) {
 	return Tuple{valueImpl{t.vrw, t.format(), w.data(), nil}}, nil
 }
 
-func (t Tuple) Append(v Value) (Tuple, error) {
+func (t Tuple) Append(vals ...Value) (Tuple, error) {
 	dec := t.decoder()
 	dec.skipKind()
 	prolog := dec.buff[:dec.offset]
@@ -544,12 +544,13 @@ func (t Tuple) Append(v Value) (Tuple, error) {
 
 	w := binaryNomsWriter{make([]byte, len(t.buff)), 0}
 	w.writeRaw(prolog)
-	w.writeCount(count + 1)
+	w.writeCount(count + uint64(len(vals)))
 	w.writeRaw(dec.buff[fieldsOffset:])
-	err := v.writeTo(&w, t.format())
-
-	if err != nil {
-		return EmptyTuple(t.nbf), err
+	for _, val := range vals {
+		err := val.writeTo(&w, t.format())
+		if err != nil {
+			return EmptyTuple(t.nbf), err
+		}
 	}
 
 	return Tuple{valueImpl{t.vrw, t.format(), w.data(), nil}}, nil

integration-tests/bats/remotes.bats

@@ -1026,3 +1026,9 @@ setup_ref_test() {
     [[ "$output" =~ "remotes/origin/branch-one" ]] || false
     [[ "$output" =~ "remotes/origin/branch-two" ]] || false
 }
+
+@test "remotes: not specifying a branch throws an error" {
+    run dolt push -u origin
+    [ "$status" -eq 1 ]
+    [[ "$output" =~ "error: --set-upstream requires <remote> and <refspec> params." ]] || false
+}

integration-tests/bats/verify-constraints.bats

@@ -3,6 +3,20 @@ load $BATS_TEST_DIRNAME/helper/common.bash
 
 setup() {
     setup_common
+    dolt sql <<"SQL"
+CREATE TABLE parent3 (pk BIGINT PRIMARY KEY, v1 BIGINT, INDEX (v1));
+CREATE TABLE child3 (pk BIGINT PRIMARY KEY, v1 BIGINT, CONSTRAINT fk_name1 FOREIGN KEY (v1) REFERENCES parent3 (v1));
+CREATE TABLE parent4 (pk BIGINT PRIMARY KEY, v1 BIGINT, INDEX (v1));
+CREATE TABLE child4 (pk BIGINT PRIMARY KEY, v1 BIGINT, CONSTRAINT fk_name2 FOREIGN KEY (v1) REFERENCES parent4 (v1));
+INSERT INTO parent3 VALUES (1, 1);
+INSERT INTO parent4 VALUES (2, 2);
+SET foreign_key_checks=0;
+INSERT INTO child3 VALUES (1, 1), (2, 2);
+INSERT INTO child4 VALUES (1, 1), (2, 2);
+SET foreign_key_checks=1;
+SQL
+    dolt add -A
+    dolt commit --force -m "has fk violations"
     dolt sql <<SQL
 CREATE TABLE parent1 (
   pk BIGINT PRIMARY KEY,
@@ -30,6 +44,10 @@ INSERT INTO parent1 VALUES (1,1), (2,2), (3,3);
 INSERT INTO parent2 VALUES (1,1), (2,2), (3,3);
 INSERT INTO child1 VALUES (1,1,1), (2,2,2);
 INSERT INTO child2 VALUES (2,2), (3,3);
+SET foreign_key_checks=0;
+INSERT INTO child3 VALUES (3, 3);
+INSERT INTO child4 VALUES (3, 3);
+SET foreign_key_checks=1;
 SQL
 }
 
@@ -39,7 +57,7 @@ teardown() {
 }
 
 @test "verify-constraints: Constraints verified" {
-    dolt verify-constraints child1 child2
+    dolt constraints verify child1 child2
 }
 
 @test "verify-constraints: One table fails" {
@@ -48,11 +66,11 @@ SET foreign_key_checks=0;
 DELETE FROM parent1 WHERE pk = 1;
 SET foreign_key_checks=1;
 SQL
-    run dolt verify-constraints child1
+    run dolt constraints verify child1
     [ "$status" -eq "1" ]
     [[ "$output" =~ "child1_parent1" ]] || false
-    dolt verify-constraints child2
-    run dolt verify-constraints child1 child2
+    dolt constraints verify child2
+    run dolt constraints verify child1 child2
     [ "$status" -eq "1" ]
     [[ "$output" =~ "child1_parent1" ]] || false
     [[ ! "$output" =~ "child1_parent2" ]] || false
@@ -65,14 +83,14 @@ SET foreign_key_checks=0;
 DELETE FROM parent2 WHERE pk = 2;
 SET foreign_key_checks=1;
 SQL
-    run dolt verify-constraints child1
+    run dolt constraints verify child1
     [ "$status" -eq "1" ]
     [[ "$output" =~ "child1_parent2" ]] || false
     [[ ! "$output" =~ "child1_parent1" ]] || false
-    run dolt verify-constraints child2
+    run dolt constraints verify child2
     [ "$status" -eq "1" ]
     [[ "$output" =~ "child2_parent2" ]] || fals
-    run dolt verify-constraints child1 child2
+    run dolt constraints verify child1 child2
     [ "$status" -eq "1" ]
     [[ "$output" =~ "child1_parent2" ]] || false
     [[ "$output" =~ "child2_parent2" ]] || false
@@ -96,14 +114,164 @@ CREATE TABLE child (
 INSERT INTO parent VALUES (1, 1, 1), (2, 2, 2);
 INSERT INTO child VALUES (1, 1, 1), (2, 20, NULL);
 SQL
-    dolt verify-constraints child
+    dolt constraints verify child
 
     dolt sql <<SQL
 SET foreign_key_checks=0;
 INSERT INTO child VALUES (3, 30, 30);
 SET foreign_key_checks=1;
 SQL
-    run dolt verify-constraints child
+    run dolt constraints verify child
     [ "$status" -eq "1" ]
     [[ "$output" =~ "fk_named" ]] || false
 }
+
+@test "verify-constraints: CLI missing --all and --output-only, no named tables" {
+    run dolt constraints verify
+    [ "$status" -eq "1" ]
+    [[ "$output" =~ "fk_name1" ]] || false
+    [[ "$output" =~ "fk_name2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ "$output" =~ "child3,1" ]] || false
+    [[ "$output" =~ "child4,1" ]] || false
+}
+
+@test "verify-constraints: CLI missing --all and --output-only, named tables" {
+    run dolt constraints verify child3
+    [ "$status" -eq "1" ]
+    [[ "$output" =~ "fk_name1" ]] || false
+    [[ ! "$output" =~ "fk_name2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ "$output" =~ "child3,1" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+}
+
+@test "verify-constraints: CLI --all" {
+    run dolt constraints verify --all child3 child4
+    [ "$status" -eq "1" ]
+    [[ "$output" =~ "fk_name1" ]] || false
+    [[ "$output" =~ "fk_name2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ "$output" =~ "child3,2" ]] || false
+    [[ "$output" =~ "child4,2" ]] || false
+    [[ "${#lines[@]}" = "3" ]] || false
+}
+
+@test "verify-constraints: CLI --output-only" {
+    run dolt constraints verify --output-only child3 child4
+    [ "$status" -eq "1" ]
+    [[ "$output" =~ "fk_name1" ]] || false
+    [[ "$output" =~ "fk_name2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ ! "$output" =~ "child3,2" ]] || false
+    [[ ! "$output" =~ "child4,2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+}
+
+@test "verify-constraints: CLI --all and --output-only" {
+    run dolt constraints verify --all --output-only child3 child4
+    [ "$status" -eq "1" ]
+    [[ "$output" =~ "fk_name1" ]] || false
+    [[ "$output" =~ "fk_name2" ]] || false
+    [[ "$output" =~ "| 2  | 2  |" ]] || false
+    [[ "$output" =~ "| 3  | 3  |" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ ! "$output" =~ "child3,2" ]] || false
+    [[ ! "$output" =~ "child4,2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+}
+
+@test "verify-constraints: SQL no violations" {
+    run dolt sql -q "SELECT CONSTRAINTS_VERIFY('child1')" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "CONSTRAINTS_VERIFY('child1')" ]] || false
+    [[ "$output" =~ "1" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ ! "$output" =~ "child1_parent1" ]] || false
+    [[ ! "$output" =~ "child1_parent2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+
+    run dolt sql -q "SELECT CONSTRAINTS_VERIFY_ALL('child1')" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "CONSTRAINTS_VERIFY_ALL('child1')" ]] || false
+    [[ "$output" =~ "1" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ ! "$output" =~ "child1_parent1" ]] || false
+    [[ ! "$output" =~ "child1_parent2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+}
+
+@test "verify-constraints: SQL CONSTRAINTS_VERIFY() no named tables" {
+    run dolt sql -q "SELECT CONSTRAINTS_VERIFY()" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "CONSTRAINTS_VERIFY()" ]] || false
+    [[ "$output" =~ "0" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ "$output" =~ "child3,1" ]] || false
+    [[ "$output" =~ "child4,1" ]] || false
+    [[ "${#lines[@]}" = "3" ]] || false
+}
+
+@test "verify-constraints: SQL CONSTRAINTS_VERIFY() named table" {
+    run dolt sql -q "SELECT CONSTRAINTS_VERIFY('child3')" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "CONSTRAINTS_VERIFY('child3')" ]] || false
+    [[ "$output" =~ "0" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ "$output" =~ "child3,1" ]] || false
+    [[ ! "$output" =~ "child4,1" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+}
+
+@test "verify-constraints: SQL CONSTRAINTS_VERIFY() named tables" {
+    run dolt sql -q "SELECT CONSTRAINTS_VERIFY('child3', 'child4')" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "CONSTRAINTS_VERIFY('child3', 'child4')" ]] || false
+    [[ "$output" =~ "0" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ "$output" =~ "child3,1" ]] || false
+    [[ "$output" =~ "child4,1" ]] || false
+    [[ "${#lines[@]}" = "3" ]] || false
+}
+
+@test "verify-constraints: SQL CONSTRAINTS_VERIFY_ALL() no named tables" {
+    run dolt sql -q "SELECT CONSTRAINTS_VERIFY_ALL()" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "CONSTRAINTS_VERIFY_ALL()" ]] || false
+    [[ "$output" =~ "0" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ "$output" =~ "child3,2" ]] || false
+    [[ "$output" =~ "child4,2" ]] || false
+    [[ "${#lines[@]}" = "3" ]] || false
+}
+
+@test "verify-constraints: SQL CONSTRAINTS_VERIFY_ALL() named table" {
+    run dolt sql -q "SELECT CONSTRAINTS_VERIFY_ALL('child3')" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "CONSTRAINTS_VERIFY_ALL('child3')" ]] || false
+    [[ "$output" =~ "0" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ "$output" =~ "child3,2" ]] || false
+    [[ ! "$output" =~ "child4,2" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+}
+
+@test "verify-constraints: SQL CONSTRAINTS_VERIFY_ALL() named tables" {
+    run dolt sql -q "SELECT CONSTRAINTS_VERIFY_ALL('child3', 'child4')" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "CONSTRAINTS_VERIFY_ALL('child3', 'child4')" ]] || false
+    [[ "$output" =~ "0" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    run dolt sql -q "SELECT * FROM dolt_constraint_violations" -r=csv
+    [[ "$output" =~ "child3,2" ]] || false
+    [[ "$output" =~ "child4,2" ]] || false
+    [[ "${#lines[@]}" = "3" ]] || false
+}
+