added validation stage for dangling chunks

go/libraries/doltcore/doltdb/durable/index.go

@@ -147,6 +147,24 @@ type nomsIndex struct {
 
 var _ Index = nomsIndex{}
 
+func IterAllIndexes(
+	ctx context.Context,
+	sch schema.Schema,
+	set IndexSet,
+	cb func(name string, idx Index) error,
+) error {
+	for _, def := range sch.Indexes().AllIndexes() {
+		idx, err := set.GetIndex(ctx, sch, def.Name())
+		if err != nil {
+			return err
+		}
+		if err = cb(def.Name(), idx); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // NomsMapFromIndex unwraps the Index and returns the underlying types.Map.
 func NomsMapFromIndex(i Index) types.Map {
 	return i.(nomsIndex).index

go/libraries/doltcore/sqle/enginetest/validation.go

@@ -15,12 +15,22 @@
 package enginetest
 
 import (
-	"github.com/dolthub/go-mysql-server/sql"
+	"context"
+	"fmt"
+
+	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
+	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb/durable"
+	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
+	"github.com/dolthub/dolt/go/store/types"
 
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle"
 )
 
-func ValidateDatabase(ctx *sql.Context, db sqle.Database) (err error) {
+func ValidateDatabase(ctx context.Context, db sqle.Database) (err error) {
+	if !types.IsFormat_DOLT_1(db.GetDoltDB().Format()) {
+		return nil
+	}
 	for _, stage := range validationStages {
 		if err = stage(ctx, db); err != nil {
 			return err
@@ -29,12 +39,71 @@ func ValidateDatabase(ctx *sql.Context, db sqle.Database) (err error) {
 	return
 }
 
-type validator func(ctx *sql.Context, db sqle.Database) error
+type validator func(ctx context.Context, db sqle.Database) error
 
 var validationStages = []validator{
-	sanityCheck,
+	validateChunkReferences,
 }
 
-func sanityCheck(ctx *sql.Context, db sqle.Database) error {
+func validateChunkReferences(ctx context.Context, db sqle.Database) error {
+	ddb := db.GetDoltDB()
+	bb, err := ddb.GetBranches(ctx)
+	if err != nil {
+		return err
+	}
+
+	for i := range bb {
+		var c *doltdb.Commit
+		var r *doltdb.RootValue
+
+		c, err = ddb.ResolveCommitRef(ctx, bb[i])
+		if err != nil {
+			return err
+		}
+		r, err = c.GetRootValue(ctx)
+		if err != nil {
+			return err
+		}
+
+		err = r.IterTables(ctx, func(_ string, t *doltdb.Table, sch schema.Schema) (stop bool, err error) {
+			if sch == nil {
+				return true, fmt.Errorf("expected non-nil schema: %v", sch)
+			}
+
+			rows, err := t.GetRowData(ctx)
+			if err != nil {
+				return true, err
+			}
+			if err = validateIndexChunkReferences(ctx, rows); err != nil {
+				return false, err
+			}
+
+			indexes, err := t.GetIndexSet(ctx)
+			if err != nil {
+				return true, err
+			}
+			err = durable.IterAllIndexes(ctx, sch, indexes, func(_ string, idx durable.Index) error {
+				return validateIndexChunkReferences(ctx, idx)
+			})
+			if err != nil {
+				return false, err
+			}
+			return
+		})
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
+
+func validateIndexChunkReferences(ctx context.Context, idx durable.Index) error {
+	pm := durable.ProllyMapFromIndex(idx)
+	return pm.WalkNodes(ctx, func(ctx context.Context, nd tree.Node) error {
+		if nd.Size() <= 0 {
+			return fmt.Errorf("encountered nil tree.Node")
+		}
+		return nil
+	})
+}