Optimize progress report queries (#167)

Previously, when running queries for the progress reports, verifier fetched all mismatches for all failed tasks. It would only display up to 40 mismatches, though … so even if there were thousands of mismatches, we discarded all but 40.

This changeset alters the logic so that 3 queries run in parallel:
- fetch first 20 differing-content mismatches
- fetch first 20 missing/changed mismatches
- count all mismatches, dividing them into differing-content vs. missing/changed

Thus, the server only returns & parses up to 40 documents for these queries, which is much lighter than returning every single mismatch for all tasks.

Author: FGasper

internal/verifier/migration_verifier_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/10gen/migration-verifier/internal/util"
 	"github.com/10gen/migration-verifier/mbson"
 	"github.com/10gen/migration-verifier/mslices"
+	"github.com/10gen/migration-verifier/option"
 	"github.com/cespare/permute/v2"
 	"github.com/rs/zerolog"
 	"github.com/samber/lo"
@@ -1254,6 +1255,8 @@ func (suite *IntegrationTestSuite) getFailuresForTask(
 		suite.Context(),
 		verifier.verificationDatabase(),
 		mslices.Of(taskID),
+		option.None[bson.D](),
+		option.None[int64](),
 	)
 
 	require.NoError(suite.T(), err)

internal/verifier/mismatches.go

@@ -23,6 +23,15 @@ type MismatchInfo struct {
 	Detail VerificationResult
 }
 
+func getMismatchDocMissingAggExpr(docExpr any) bson.D {
+	return getResultDocMissingAggExpr(
+		bson.D{{"$getField", bson.D{
+			{"input", docExpr},
+			{"field", "detail"},
+		}}},
+	)
+}
+
 var _ bson.Marshaler = MismatchInfo{}
 
 func (mi MismatchInfo) MarshalBSON() ([]byte, error) {
@@ -72,21 +81,91 @@ func createMismatchesCollection(ctx context.Context, db *mongo.Database) error {
 	return nil
 }
 
-func getMismatchesForTasks(
+func countMismatchesForTasks(
 	ctx context.Context,
 	db *mongo.Database,
 	taskIDs []bson.ObjectID,
-) (map[bson.ObjectID][]VerificationResult, error) {
-	cursor, err := db.Collection(mismatchesCollectionName).Find(
+	filter bson.D,
+) (int64, int64, error) {
+	cursor, err := db.Collection(mismatchesCollectionName).Aggregate(
 		ctx,
-		bson.D{
-			{"task", bson.D{{"$in", taskIDs}}},
+		mongo.Pipeline{
+			{{"$match", bson.D{
+				{"task", bson.D{{"$in", taskIDs}}},
+			}}},
+			{{"$group", bson.D{
+				{"_id", nil},
+				{"total", bson.D{{"$sum", 1}}},
+				{"match", bson.D{{"$sum", bson.D{
+					{"$cond", bson.D{
+						{"if", filter},
+						{"then", 1},
+						{"else", 0},
+					}},
+				}}}},
+			}}},
 		},
-		options.Find().SetSort(
+	)
+
+	if err != nil {
+		return 0, 0, errors.Wrap(err, "sending mismatch-counting query")
+	}
+
+	var got []bson.Raw
+	if err := cursor.All(ctx, &got); err != nil {
+		return 0, 0, errors.Wrap(err, "reading mismatch counts")
+	}
+
+	if len(got) != 1 {
+		return 0, 0, fmt.Errorf("unexpected mismatch count result: %+v", got)
+	}
+
+	totalRV, err := got[0].LookupErr("total")
+	if err != nil {
+		return 0, 0, errors.Wrap(err, "getting mismatch count’s total")
+	}
+
+	matchRV, err := got[0].LookupErr("match")
+	if err != nil {
+		return 0, 0, errors.Wrap(err, "getting mismatch count’s filter-match count")
+	}
+
+	matched := matchRV.AsInt64()
+
+	return matched, totalRV.AsInt64() - matched, nil
+}
+
+func getMismatchesForTasks(
+	ctx context.Context,
+	db *mongo.Database,
+	taskIDs []bson.ObjectID,
+	filter option.Option[bson.D],
+	limit option.Option[int64],
+) (map[bson.ObjectID][]VerificationResult, error) {
+	findOpts := options.Find().
+		SetSort(
 			bson.D{
 				{"detail.id", 1},
 			},
-		),
+		)
+
+	if limit, has := limit.Get(); has {
+		findOpts.SetLimit(limit)
+	}
+
+	query := bson.D{
+		{"task", bson.D{{"$in", taskIDs}}},
+	}
+
+	if filter, has := filter.Get(); has {
+		query = bson.D{
+			{"$and", []bson.D{query, filter}},
+		}
+	}
+	cursor, err := db.Collection(mismatchesCollectionName).Find(
+		ctx,
+		query,
+		findOpts,
 	)
 
 	if err != nil {

internal/verifier/result.go

@@ -51,6 +51,27 @@ func (vr VerificationResult) DocumentIsMissing() bool {
 	return vr.Details == Missing && vr.Field == ""
 }
 
+func getResultDocMissingAggExpr(docExpr any) bson.D {
+	return bson.D{
+		{"$and", []bson.D{
+			{{"$eq", bson.A{
+				Missing,
+				bson.D{{"$getField", bson.D{
+					{"input", docExpr},
+					{"field", "details"},
+				}}},
+			}}},
+			{{"$eq", bson.A{
+				"",
+				bson.D{{"$getField", bson.D{
+					{"input", docExpr},
+					{"field", "field"},
+				}}},
+			}}},
+		}},
+	}
+}
+
 var _ bson.Marshaler = VerificationResult{}
 
 func (vr VerificationResult) MarshalBSON() ([]byte, error) {

internal/verifier/summary.go

@@ -12,8 +12,10 @@ import (
 	"strings"
 	"time"
 
+	"github.com/10gen/migration-verifier/contextplus"
 	"github.com/10gen/migration-verifier/internal/reportutils"
 	"github.com/10gen/migration-verifier/internal/types"
+	"github.com/10gen/migration-verifier/option"
 	"github.com/olekukonko/tablewriter"
 	"github.com/pkg/errors"
 	"github.com/samber/lo"
@@ -63,6 +65,8 @@ func (verifier *Verifier) reportCollectionMetadataMismatches(ctx context.Context
 					return ft.PrimaryKey
 				},
 			),
+			option.None[bson.D](),
+			option.None[int64](),
 		)
 		if err != nil {
 			return false, false, errors.Wrapf(
@@ -112,95 +116,127 @@ func (verifier *Verifier) reportDocumentMismatches(ctx context.Context, strBuild
 	strBuilder.WriteString("\n")
 
 	// First present summaries of failures based on present/missing and differing content
-	failureTypesTable := tablewriter.NewWriter(strBuilder)
-	failureTypesTable.SetHeader([]string{"Failure Type", "Count"})
+	countsTable := tablewriter.NewWriter(strBuilder)
+	countsTable.SetHeader([]string{"Failure Type", "Count"})
+
+	failedTaskIDs := lo.Map(
+		failedTasks,
+		func(ft VerificationTask, _ int) bson.ObjectID {
+			return ft.PrimaryKey
+		},
+	)
 
-	taskDiscrepancies, err := getMismatchesForTasks(
-		ctx,
-		verifier.verificationDatabase(),
-		lo.Map(
-			failedTasks,
-			func(ft VerificationTask, _ int) bson.ObjectID {
-				return ft.PrimaryKey
-			},
-		),
+	var mismatchTaskDiscrepancies, missingOrChangedDiscrepancies map[bson.ObjectID][]VerificationResult
+
+	contentMismatchCount := int64(0)
+	missingOrChangedCount := int64(0)
+
+	eg, egCtx := contextplus.ErrGroup(ctx)
+	eg.Go(
+		func() error {
+			var err error
+			mismatchTaskDiscrepancies, err = getMismatchesForTasks(
+				egCtx,
+				verifier.verificationDatabase(),
+				failedTaskIDs,
+				option.Some(
+					bson.D{{"$expr", bson.D{
+						{"$not", getMismatchDocMissingAggExpr("$$ROOT")},
+					}}},
+				),
+				option.Some(verifier.failureDisplaySize),
+			)
+
+			return errors.Wrapf(
+				err,
+				"fetching %d failed tasks’ content-mismatch discrepancies",
+				len(failedTasks),
+			)
+		},
 	)
-	if err != nil {
-		return false, false, errors.Wrapf(
-			err,
-			"fetching %d failed tasks' discrepancies",
-			len(failedTasks),
-		)
-	}
 
-	contentMismatchCount := 0
-	missingOrChangedCount := 0
-	for _, task := range failedTasks {
-		discrepancies, hasDiscrepancies := taskDiscrepancies[task.PrimaryKey]
-		if !hasDiscrepancies {
-			return false, false, errors.Wrapf(
+	eg.Go(
+		func() error {
+			var err error
+			missingOrChangedCount, contentMismatchCount, err = countMismatchesForTasks(
+				egCtx,
+				verifier.verificationDatabase(),
+				failedTaskIDs,
+				getMismatchDocMissingAggExpr("$$ROOT"),
+			)
+
+			return errors.Wrapf(
 				err,
-				"task %v is marked %#q but has no recorded discrepancies; internal error?",
-				task.PrimaryKey,
-				task.Status,
+				"counting %d failed tasks’ discrepancies",
+				len(failedTasks),
 			)
-		}
+		},
+	)
 
-		missingCount := lo.CountBy(
-			discrepancies,
-			func(d VerificationResult) bool {
-				return d.DocumentIsMissing()
-			},
-		)
+	eg.Go(
+		func() error {
+			var err error
+			missingOrChangedDiscrepancies, err = getMismatchesForTasks(
+				egCtx,
+				verifier.verificationDatabase(),
+				failedTaskIDs,
+				option.Some(
+					bson.D{{"$expr", getMismatchDocMissingAggExpr("$$ROOT")}},
+				),
+				option.Some(verifier.failureDisplaySize),
+			)
 
-		contentMismatchCount += len(discrepancies) - missingCount
-		missingOrChangedCount += missingCount
+			return errors.Wrapf(
+				err,
+				"fetching %d failed tasks' missing/changed discrepancies",
+				len(failedTasks),
+			)
+		},
+	)
+
+	if err := eg.Wait(); err != nil {
+		return false, false, errors.Wrapf(err, "gathering mismatch data")
 	}
 
-	failureTypesTable.Append([]string{
+	countsTable.Append([]string{
 		"Documents With Differing Content",
-		fmt.Sprintf("%v", reportutils.FmtReal(contentMismatchCount)),
+		reportutils.FmtReal(contentMismatchCount),
 	})
-	failureTypesTable.Append([]string{
+	countsTable.Append([]string{
 		"Missing or Changed Documents",
-		fmt.Sprintf("%v", reportutils.FmtReal(missingOrChangedCount)),
+		reportutils.FmtReal(missingOrChangedCount),
 	})
-	strBuilder.WriteString("Failure summary:\n")
-	failureTypesTable.Render()
+	countsTable.Render()
 
 	mismatchedDocsTable := tablewriter.NewWriter(strBuilder)
 	mismatchedDocsTableRows := types.ToNumericTypeOf(0, verifier.failureDisplaySize)
 	mismatchedDocsTable.SetHeader([]string{"ID", "Cluster", "Field", "Namespace", "Details"})
 
-	printAll := int64(contentMismatchCount) < (verifier.failureDisplaySize + int64(0.25*float32(verifier.failureDisplaySize)))
-OUTA:
+	printAll := int64(contentMismatchCount) <= verifier.failureDisplaySize
+
 	for _, task := range failedTasks {
-		for _, d := range taskDiscrepancies[task.PrimaryKey] {
+		for _, d := range mismatchTaskDiscrepancies[task.PrimaryKey] {
 			if d.DocumentIsMissing() {
-				continue
-			}
-
-			if !printAll && mismatchedDocsTableRows >= verifier.failureDisplaySize {
-				break OUTA
+				panic(fmt.Sprintf("found missing-type mismatch but expected content-mismatch: %+v", d))
 			}
 
 			mismatchedDocsTableRows++
 			mismatchedDocsTable.Append([]string{
 				fmt.Sprintf("%v", d.ID),
-				fmt.Sprintf("%v", d.Cluster),
-				fmt.Sprintf("%v", d.Field),
-				fmt.Sprintf("%v", d.NameSpace),
-				fmt.Sprintf("%v", d.Details),
+				d.Cluster,
+				d.Field,
+				d.NameSpace,
+				d.Details,
 			})
 		}
 	}
 
 	if mismatchedDocsTableRows > 0 {
 		strBuilder.WriteString("\n")
 		if printAll {
-			strBuilder.WriteString("All documents in tasks in failed status due to differing content:\n")
+			strBuilder.WriteString("All documents found with differing content:\n")
 		} else {
-			fmt.Fprintf(strBuilder, "First %d documents in tasks in failed status due to differing content:\n", verifier.failureDisplaySize)
+			fmt.Fprintf(strBuilder, "First %d documents found with differing content:\n", verifier.failureDisplaySize)
 		}
 		mismatchedDocsTable.Render()
 	}
@@ -209,23 +245,18 @@ OUTA:
 	missingOrChangedDocsTableRows := types.ToNumericTypeOf(0, verifier.failureDisplaySize)
 	missingOrChangedDocsTable.SetHeader([]string{"Document ID", "Source Namespace", "Destination Namespace"})
 
-	printAll = int64(missingOrChangedCount) < (verifier.failureDisplaySize + int64(0.25*float32(verifier.failureDisplaySize)))
-OUTB:
+	printAll = int64(missingOrChangedCount) <= verifier.failureDisplaySize
 	for _, task := range failedTasks {
-		for _, d := range taskDiscrepancies[task.PrimaryKey] {
+		for _, d := range missingOrChangedDiscrepancies[task.PrimaryKey] {
 			if !d.DocumentIsMissing() {
-				continue
-			}
-
-			if !printAll && missingOrChangedDocsTableRows >= verifier.failureDisplaySize {
-				break OUTB
+				panic(fmt.Sprintf("found content-mismatch mismatch but expected missing/changed: %+v", d))
 			}
 
 			missingOrChangedDocsTableRows++
 			missingOrChangedDocsTable.Append([]string{
 				fmt.Sprintf("%v", d.ID),
-				fmt.Sprintf("%v", task.QueryFilter.Namespace),
-				fmt.Sprintf("%v", task.QueryFilter.To),
+				task.QueryFilter.Namespace,
+				task.QueryFilter.To,
 			})
 		}
 	}