Further change stream optimizations (#135)

This optimizes the change stream by:
- Removing `updateLookup`. It’s there so that we can group rechecks into more evenly-sized tasks, but it incurs the overhead of fetching the document on each change event. That overhead almost certainly outweighs the task-sizing benefit.
- Removing `wallTime`. When change events are small, this uselessly inflates the event and reduces the # of events we can parse at once.
- Replacing `documentKey` with a new `_docID` field that just contains the document ID, which is the only part of the document key that we actually need. (The shard key fields are not needed in the change event.)

This also removes the internal ParsedEvent.ID field, which nothing actually needs.

Author: FGasper

internal/verifier/change_stream.go

@@ -41,23 +41,16 @@ var supportedEventOpTypes = mapset.NewSet(
 
 // ParsedEvent contains the fields of an event that we have parsed from 'bson.Raw'.
 type ParsedEvent struct {
-	ID           any                            `bson:"_id"`
 	OpType       string                         `bson:"operationType"`
 	Ns           *Namespace                     `bson:"ns,omitempty"`
-	DocKey       DocKey                         `bson:"documentKey,omitempty"`
+	DocID        any                            `bson:"_docID,omitempty"`
 	FullDocument bson.Raw                       `bson:"fullDocument,omitempty"`
 	FullDocLen   option.Option[types.ByteCount] `bson:"_fullDocLen"`
 	ClusterTime  *primitive.Timestamp           `bson:"clusterTime,omitEmpty"`
 }
 
 func (pe *ParsedEvent) String() string {
-	return fmt.Sprintf("{OpType: %s, namespace: %s, docID: %v, clusterTime: %v}", pe.OpType, pe.Ns, pe.DocKey.ID, pe.ClusterTime)
-}
-
-// DocKey is a deserialized form for the ChangeEvent documentKey field. We currently only care about
-// the _id.
-type DocKey struct {
-	ID any `bson:"_id"`
+	return fmt.Sprintf("{OpType: %s, namespace: %s, docID: %v, clusterTime: %v}", pe.OpType, pe.Ns, pe.DocID, pe.ClusterTime)
 }
 
 const (
@@ -244,7 +237,7 @@ func (verifier *Verifier) HandleChangeStreamEvents(ctx context.Context, batch ch
 
 		dbNames[i] = srcDBName
 		collNames[i] = srcCollName
-		docIDs[i] = changeEvent.DocKey.ID
+		docIDs[i] = changeEvent.DocID
 
 		if changeEvent.FullDocLen.OrZero() > 0 {
 			dataSizes[i] = int(changeEvent.FullDocLen.OrZero())
@@ -323,8 +316,12 @@ func (csr *ChangeStreamReader) GetChangeStreamFilter() (pipeline mongo.Pipeline)
 	pipeline = append(
 		pipeline,
 		bson.D{
-			{"$unset", []string{
-				"updateDescription",
+			{"$addFields", bson.D{
+				{"_docID", "$documentKey._id"},
+
+				{"updateDescription", "$$REMOVE"},
+				{"wallTime", "$$REMOVE"},
+				{"documentKey", "$$REMOVE"},
 			}},
 		},
 	)
@@ -622,8 +619,7 @@ func (csr *ChangeStreamReader) createChangeStream(
 ) (*mongo.ChangeStream, mongo.Session, primitive.Timestamp, error) {
 	pipeline := csr.GetChangeStreamFilter()
 	opts := options.ChangeStream().
-		SetMaxAwaitTime(maxChangeStreamAwaitTime).
-		SetFullDocument(options.UpdateLookup)
+		SetMaxAwaitTime(maxChangeStreamAwaitTime)
 
 	if csr.clusterInfo.VersionArray[0] >= 6 {
 		opts = opts.SetCustomPipeline(bson.M{"showExpandedEvents": true})

internal/verifier/change_stream_test.go

@@ -108,7 +108,6 @@ func (suite *IntegrationTestSuite) TestChangeStreamFilter_BsonSize() {
 	cs, err := suite.srcMongoClient.Watch(
 		ctx,
 		filter,
-		options.ChangeStream().SetFullDocument("updateLookup"),
 	)
 	suite.Require().NoError(err)
 	defer cs.Close(ctx)
@@ -125,17 +124,18 @@ func (suite *IntegrationTestSuite) TestChangeStreamFilter_BsonSize() {
 
 	suite.Require().True(cs.Next(ctx), "should get event")
 
-	suite.Require().Equal(
-		"abc",
-		cs.Current.Lookup("documentKey", "_id").StringValue(),
-		"event should reference expected document",
-	)
 	suite.Assert().Less(len(cs.Current), 10_000, "event should not be large")
 
 	parsed := ParsedEvent{}
 	suite.Require().NoError(cs.Decode(&parsed))
 	suite.Require().Equal("insert", parsed.OpType)
 
+	suite.Require().Equal(
+		"abc",
+		parsed.DocID,
+		"event should reference expected document",
+	)
+
 	suite.Require().True(parsed.FullDocLen.IsSome(), "full doc len should be in event")
 	suite.Assert().Greater(parsed.FullDocLen.MustGet(), types.ByteCount(10_000))
 
@@ -313,7 +313,8 @@ func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
 			"docID": "heyhey",
 		},
 		recheckDocs[0]["_id"],
-		"recheck doc should have expected ID",
+		"recheck doc (%v) should have expected ID",
+		recheckDocs[0],
 	)
 }
 

internal/verifier/migration_verifier_test.go

@@ -577,9 +577,7 @@ func (suite *IntegrationTestSuite) TestGetPersistedNamespaceStatistics_Recheck()
 			events: []ParsedEvent{{
 				OpType: "insert",
 				Ns:     &Namespace{DB: "mydb", Coll: "coll2"},
-				DocKey: DocKey{
-					ID: "heyhey",
-				},
+				DocID:  "heyhey",
 				ClusterTime: &primitive.Timestamp{
 					T: uint32(time.Now().Unix()),
 				},
@@ -593,14 +591,9 @@ func (suite *IntegrationTestSuite) TestGetPersistedNamespaceStatistics_Recheck()
 		ctx,
 		changeEventBatch{
 			events: []ParsedEvent{{
-				ID: bson.M{
-					"docID": "ID/docID",
-				},
 				OpType: "insert",
 				Ns:     &Namespace{DB: "mydb", Coll: "coll1"},
-				DocKey: DocKey{
-					ID: "hoohoo",
-				},
+				DocID:  "hoohoo",
 				ClusterTime: &primitive.Timestamp{
 					T: uint32(time.Now().Unix()),
 				},
@@ -842,7 +835,7 @@ func (suite *IntegrationTestSuite) TestFailedVerificationTaskInsertions() {
 	err = verifier.InsertFailedCompareRecheckDocs(ctx, "foo.bar2", []any{42}, []int{100})
 	suite.Require().NoError(err)
 	event := ParsedEvent{
-		DocKey: DocKey{ID: int32(55)},
+		DocID:  int32(55),
 		OpType: "delete",
 		Ns: &Namespace{
 			DB:   "foo",

internal/verifier/recheck_test.go

@@ -48,9 +48,7 @@ func (suite *IntegrationTestSuite) TestFailedCompareThenReplace() {
 
 	event := ParsedEvent{
 		OpType: "insert",
-		DocKey: DocKey{
-			ID: "theDocID",
-		},
+		DocID:  "theDocID",
 		Ns: &Namespace{
 			DB:   "the",
 			Coll: "namespace",