diff --git a/go/vt/binlog/binlogplayer/binlog_player.go b/go/vt/binlog/binlogplayer/binlog_player.go
index d7614759a2c..d662129cd17 100644
--- a/go/vt/binlog/binlogplayer/binlog_player.go
+++ b/go/vt/binlog/binlogplayer/binlog_player.go
@@ -79,6 +79,8 @@ type Stats struct {
 
 	SecondsBehindMaster sync2.AtomicInt64
 	History             *history.History
+
+	State sync2.AtomicString
 }
 
 // SetLastPosition sets the last replication position.
@@ -176,17 +178,12 @@ func NewBinlogPlayerTables(dbClient DBClient, tablet *topodatapb.Tablet, tables
 // If an error is encountered, it updates the vreplication state to "Error".
 // If a stop position was specified, and reached, the state is updated to "Stopped".
 func (blp *BinlogPlayer) ApplyBinlogEvents(ctx context.Context) error {
-	if err := SetVReplicationState(blp.dbClient, blp.uid, BlpRunning, ""); err != nil {
+	if err := blp.setVReplicationState(BlpRunning, ""); err != nil {
 		log.Errorf("Error writing Running state: %v", err)
 	}
 
 	if err := blp.applyEvents(ctx); err != nil {
-		msg := err.Error()
-		blp.blplStats.History.Add(&StatsHistoryRecord{
-			Time:    time.Now(),
-			Message: msg,
-		})
-		if err := SetVReplicationState(blp.dbClient, blp.uid, BlpError, msg); err != nil {
+		if err := blp.setVReplicationState(BlpError, err.Error()); err != nil {
 			log.Errorf("Error writing stop state: %v", err)
 		}
 		return err
@@ -241,14 +238,14 @@ func (blp *BinlogPlayer) applyEvents(ctx context.Context) error {
 		case blp.position.Equal(blp.stopPosition):
 			msg := fmt.Sprintf("not starting BinlogPlayer, we're already at the desired position %v", blp.stopPosition)
 			log.Info(msg)
-			if err := SetVReplicationState(blp.dbClient, blp.uid, BlpStopped, msg); err != nil {
+			if err := blp.setVReplicationState(BlpStopped, msg); err != nil {
 				log.Errorf("Error writing stop state: %v", err)
 			}
 			return nil
 		case blp.position.AtLeast(blp.stopPosition):
 			msg := fmt.Sprintf("starting point %v greater than stopping point %v", blp.position, blp.stopPosition)
 			log.Error(msg)
-			if err := SetVReplicationState(blp.dbClient, blp.uid, BlpStopped, msg); err != nil {
+			if err := blp.setVReplicationState(BlpStopped, msg); err != nil {
 				log.Errorf("Error writing stop state: %v", err)
 			}
 			// Don't return an error. Otherwise, it will keep retrying.
@@ -348,7 +345,7 @@ func (blp *BinlogPlayer) applyEvents(ctx context.Context) error {
 					if blp.position.AtLeast(blp.stopPosition) {
 						msg := "Reached stopping position, done playing logs"
 						log.Info(msg)
-						if err := SetVReplicationState(blp.dbClient, blp.uid, BlpStopped, msg); err != nil {
+						if err := blp.setVReplicationState(BlpStopped, msg); err != nil {
 							log.Errorf("Error writing stop state: %v", err)
 						}
 						return nil
@@ -463,6 +460,21 @@ func (blp *BinlogPlayer) writeRecoveryPosition(tx *binlogdatapb.BinlogTransactio
 	return nil
 }
 
+func (blp *BinlogPlayer) setVReplicationState(state, message string) error {
+	if message != "" {
+		blp.blplStats.History.Add(&StatsHistoryRecord{
+			Time:    time.Now(),
+			Message: message,
+		})
+	}
+	blp.blplStats.State.Set(state)
+	query := fmt.Sprintf("update _vt.vreplication set state='%v', message=%v where id=%v", state, encodeString(MessageTruncate(message)), blp.uid)
+	if _, err := blp.dbClient.ExecuteFetch(query, 1); err != nil {
+		return fmt.Errorf("could not set state: %v: %v", query, err)
+	}
+	return nil
+}
+
 // CreateVReplicationTable returns the statements required to create
 // the _vt.vreplication table.
 // id: is an auto-increment column that identifies the stream.
@@ -507,15 +519,6 @@ func AlterVReplicationTable() []string {
 	return []string{"ALTER TABLE _vt.vreplication ADD COLUMN db_name VARBINARY(255) NOT NULL"}
 }
 
-// SetVReplicationState updates the state in the _vt.vreplication table.
-func SetVReplicationState(dbClient DBClient, uid uint32, state, message string) error {
-	query := fmt.Sprintf("update _vt.vreplication set state='%v', message=%v where id=%v", state, encodeString(MessageTruncate(message)), uid)
-	if _, err := dbClient.ExecuteFetch(query, 1); err != nil {
-		return fmt.Errorf("could not set state: %v: %v", query, err)
-	}
-	return nil
-}
-
 // VRSettings contains the settings of a vreplication table.
 type VRSettings struct {
 	StartPos          mysql.Position
diff --git a/go/vt/vttablet/tabletmanager/vreplication/controller.go b/go/vt/vttablet/tabletmanager/vreplication/controller.go
index 5e9a9b75eae..b215edd2771 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/controller.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/controller.go
@@ -90,6 +90,7 @@ func newController(ctx context.Context, params map[string]string, dbClientFactor
 	ct.id = uint32(id)
 	ct.workflow = params["workflow"]
 
+	blpStats.State.Set(params["state"])
 	// Nothing to do if replication is stopped.
 	if params["state"] == binlogplayer.BlpStopped {
 		ct.cancel = func() {}
diff --git a/go/vt/vttablet/tabletmanager/vreplication/engine.go b/go/vt/vttablet/tabletmanager/vreplication/engine.go
index 00ac39d04aa..2470d0a5a70 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/engine.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/engine.go
@@ -405,6 +405,17 @@ func (vre *Engine) fetchIDs(dbClient binlogplayer.DBClient, selector string) (id
 	return ids, bv, nil
 }
 
+// registerJournal is invoked if any of the vreplication streams encounters a journal event.
+// Multiple registerJournal functions collaborate to converge on the final action.
+// The first invocation creates an entry in vre.journaler. The entry is initialized
+// with the list of participants that also need to converge.
+// The middle invocation happens on the first and subsequent calls: the current participant
+// marks itself as having joined the wait.
+// The final invocation happens for the last participant that joins. Having confirmed
+// that all the participants have joined, transitionJournal is invoked, which deletes
+// all current participant streams and creates new ones to replace them.
+// A unified journal event is identified by the workflow name and journal id.
+// Multiple independent journal events can go through this cycle concurrently.
 func (vre *Engine) registerJournal(journal *binlogdatapb.Journal, id int) error {
 	vre.mu.Lock()
 	defer vre.mu.Unlock()
@@ -417,6 +428,7 @@ func (vre *Engine) registerJournal(journal *binlogdatapb.Journal, id int) error
 	key := fmt.Sprintf("%s:%d", workflow, journal.Id)
 	je, ok := vre.journaler[key]
 	if !ok {
+		// First invocation. Create the entry.
 		log.Infof("Journal encountered: %v", journal)
 		controllerSources := make(map[string]bool)
 		for _, ct := range vre.controllers {
@@ -441,19 +453,25 @@ func (vre *Engine) registerJournal(journal *binlogdatapb.Journal, id int) error
 		vre.journaler[key] = je
 	}
 
+	// Middle invocation. Register yourself
 	ks := fmt.Sprintf("%s:%s", vre.controllers[id].source.Keyspace, vre.controllers[id].source.Shard)
 	log.Infof("Registering id %v against %v", id, ks)
 	je.participants[ks] = id
+	// Check if all participants have joined.
 	for _, pid := range je.participants {
 		if pid == 0 {
 			// Still need to wait.
 			return nil
 		}
 	}
+
+	// Final invocation. Perform the transition.
 	go vre.transitionJournal(key)
 	return nil
 }
 
+// transitionJournal stops all existing participants, deletes their vreplication
+// entries, and creates new ones as instructed by the journal metadata.
 func (vre *Engine) transitionJournal(key string) {
 	vre.mu.Lock()
 	defer vre.mu.Unlock()
@@ -484,6 +502,7 @@ func (vre *Engine) transitionJournal(key string) {
 		return
 	}
 
+	// Use the reference row to copy other fields like cell, tablet_types, etc.
 	params, err := readRow(dbClient, refid)
 	if err != nil {
 		log.Errorf("transitionJournal: %v", err)
diff --git a/go/vt/vttablet/tabletmanager/vreplication/framework_test.go b/go/vt/vttablet/tabletmanager/vreplication/framework_test.go
index 9d5efb00d0a..72b9e2c3b82 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/framework_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/framework_test.go
@@ -216,12 +216,18 @@ func (ftc *fakeTabletConn) StreamHealth(ctx context.Context, callback func(*quer
 	})
 }
 
+// vstreamHook allows you to do work just before calling VStream.
+var vstreamHook func(ctx context.Context)
+
 // VStream directly calls into the pre-initialized engine.
 func (ftc *fakeTabletConn) VStream(ctx context.Context, target *querypb.Target, startPos string, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error {
 	if target.Keyspace != "vttest" {
 		<-ctx.Done()
 		return io.EOF
 	}
+	if vstreamHook != nil {
+		vstreamHook(ctx)
+	}
 	return streamerEngine.Stream(ctx, startPos, filter, send)
 }
 
diff --git a/go/vt/vttablet/tabletmanager/vreplication/replicator_plan.go b/go/vt/vttablet/tabletmanager/vreplication/replicator_plan.go
index a89f887f92b..3726b36b3eb 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/replicator_plan.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/replicator_plan.go
@@ -29,12 +29,19 @@ import (
 	querypb "vitess.io/vitess/go/vt/proto/query"
 )
 
-// ReplicatorPlan is the execution plan for the replicator.
-// The constructor for this is buildReplicatorPlan in table_plan_builder.go
-// The initial build identifies the tables that need to be replicated,
-// and builds partial TablePlan objects for them. The partial plan is used
-// to send streaming requests. As the responses return field info, this
-// information is used to build the final execution plan (buildExecutionPlan).
+// ReplicatorPlan is the execution plan for the replicator. It contains
+// plans for all the tables it's replicating. Every phase of vreplication
+// builds its own instance of the ReplicatorPlan. This is because the plan
+// depends on copyState, which changes on every iteration.
+// The table plans within ReplicatorPlan will not be fully populated because
+// all the information is not available initially.
+// For simplicity, the ReplicatorPlan is immutable.
+// Once we get the field info for a table from the stream response,
+// we'll have all the necessary info to build the final plan.
+// At that time, buildExecutionPlan is invoked, which will make a copy
+// of the TablePlan from ReplicatorPlan, and fill the rest
+// of the members, leaving the original plan unchanged.
+// The constructor is buildReplicatorPlan in table_plan_builder.go
 type ReplicatorPlan struct {
 	VStreamFilter *binlogdatapb.Filter
 	TargetTables  map[string]*TablePlan
@@ -50,6 +57,8 @@ func (rp *ReplicatorPlan) buildExecutionPlan(fieldEvent *binlogdatapb.FieldEvent
 		// Unreachable code.
 		return nil, fmt.Errorf("plan not found for %s", fieldEvent.TableName)
 	}
+	// If Insert is initialized, then it means that we knew the column
+	// names and have already built most of the plan.
 	if prelim.Insert != nil {
 		tplanv := *prelim
 		// We know that we sent only column names, but they may be backticked.
@@ -93,6 +102,7 @@ func (rp *ReplicatorPlan) buildFromFields(tableName string, lastpk *sqltypes.Res
 		}
 		tpb.colExprs = append(tpb.colExprs, cexpr)
 	}
+	// The following actions are a subset of buildTablePlan.
 	if err := tpb.analyzePK(rp.tableKeys); err != nil {
 		return nil, err
 	}
@@ -118,19 +128,35 @@ func (rp *ReplicatorPlan) MarshalJSON() ([]byte, error) {
 	return json.Marshal(&v)
 }
 
-// TablePlan is the execution plan for a table within a player stream.
+// TablePlan is the execution plan for a table within a replicator.
+// If the column names are not known at the time of plan building (like
+// select *), then only TargetName, SendRule and Lastpk are initialized.
+// When the stream returns the field info, those are used as column
+// names to build the final plan.
+// Lastpk comes from copyState. If it's set, then the generated plans
+// are significantly different because any events that fall beyond
+// Lastpk must be excluded.
+// If column names were known upfront, then all fields of TablePlan
+// are built except for Fields. This member is populated only after
+// the field info is received from the stream.
 // The ParsedQuery objects assume that a map of before and after values
 // will be built based on the streaming rows. Before image values will
 // be prefixed with a "b_", and after image values will be prefixed
-// with a "a_".
+// with a "a_". The TablePlan structure is used during all the phases
+// of vreplication: catchup, copy, fastforward, or regular replication.
 type TablePlan struct {
-	TargetName   string
-	SendRule     *binlogdatapb.Rule
-	PKReferences []string
-	// Lastpk is used for delayed generation of replication queries.
+	// TargetName, SendRule will always be initialized.
+	TargetName string
+	SendRule   *binlogdatapb.Rule
+	// Lastpk will be initialized if it was specified, and
+	// will be used for building the final plan after field info
+	// is received.
 	Lastpk *sqltypes.Result
 	// BulkInsertFront, BulkInsertValues and BulkInsertOnDup are used
-	// by vcopier.
+	// by vcopier. These three parts are combined to build bulk insert
+	// statements. This is functionally equivalent to generating
+	// multiple statements using the "Insert" construct, but much more
+	// efficient for the copy phase.
 	BulkInsertFront  *sqlparser.ParsedQuery
 	BulkInsertValues *sqlparser.ParsedQuery
 	BulkInsertOnDup  *sqlparser.ParsedQuery
@@ -142,6 +168,9 @@ type TablePlan struct {
 	Update *sqlparser.ParsedQuery
 	Delete *sqlparser.ParsedQuery
 	Fields []*querypb.Field
+	// PKReferences is used to check if an event changed
+	// a primary key column (row move).
+	PKReferences []string
 }
 
 // MarshalJSON performs a custom JSON Marshalling.
@@ -149,23 +178,23 @@ func (tp *TablePlan) MarshalJSON() ([]byte, error) {
 	v := struct {
 		TargetName   string
 		SendRule     string
-		PKReferences []string               `json:",omitempty"`
 		InsertFront  *sqlparser.ParsedQuery `json:",omitempty"`
 		InsertValues *sqlparser.ParsedQuery `json:",omitempty"`
 		InsertOnDup  *sqlparser.ParsedQuery `json:",omitempty"`
 		Insert       *sqlparser.ParsedQuery `json:",omitempty"`
 		Update       *sqlparser.ParsedQuery `json:",omitempty"`
 		Delete       *sqlparser.ParsedQuery `json:",omitempty"`
+		PKReferences []string               `json:",omitempty"`
 	}{
 		TargetName:   tp.TargetName,
 		SendRule:     tp.SendRule.Match,
-		PKReferences: tp.PKReferences,
 		InsertFront:  tp.BulkInsertFront,
 		InsertValues: tp.BulkInsertValues,
 		InsertOnDup:  tp.BulkInsertOnDup,
 		Insert:       tp.Insert,
 		Update:       tp.Update,
 		Delete:       tp.Delete,
+		PKReferences: tp.PKReferences,
 	}
 	return json.Marshal(&v)
 }
diff --git a/go/vt/vttablet/tabletmanager/vreplication/replicator_plan_test.go b/go/vt/vttablet/tabletmanager/vreplication/replicator_plan_test.go
index 488f325ef0f..a7aec43d436 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/replicator_plan_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/replicator_plan_test.go
@@ -35,13 +35,13 @@ type TestReplicatorPlan struct {
 type TestTablePlan struct {
 	TargetName   string
 	SendRule     string
-	PKReferences []string `json:",omitempty"`
 	InsertFront  string   `json:",omitempty"`
 	InsertValues string   `json:",omitempty"`
 	InsertOnDup  string   `json:",omitempty"`
 	Insert       string   `json:",omitempty"`
 	Update       string   `json:",omitempty"`
 	Delete       string   `json:",omitempty"`
+	PKReferences []string `json:",omitempty"`
 }
 
 func TestBuildPlayerPlan(t *testing.T) {
diff --git a/go/vt/vttablet/tabletmanager/vreplication/stats.go b/go/vt/vttablet/tabletmanager/vreplication/stats.go
index ea688fa4774..a13b1e2e549 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/stats.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/stats.go
@@ -148,12 +148,6 @@ func (st *vrStats) status() *EngineStatus {
 	status.Controllers = make([]*ControllerStatus, len(st.controllers))
 	i := 0
 	for _, ct := range st.controllers {
-		state := "Running"
-		select {
-		case <-ct.done:
-			state = "Stopped"
-		default:
-		}
 		status.Controllers[i] = &ControllerStatus{
 			Index:               ct.id,
 			Source:              ct.source.String(),
@@ -162,7 +156,7 @@ func (st *vrStats) status() *EngineStatus {
 			SecondsBehindMaster: ct.blpStats.SecondsBehindMaster.Get(),
 			Counts:              ct.blpStats.Timings.Counts(),
 			Rates:               ct.blpStats.Rates.Get(),
-			State:               state,
+			State:               ct.blpStats.State.Get(),
 			SourceTablet:        ct.sourceTablet.Get(),
 			Messages:            ct.blpStats.MessageHistory(),
 		}
diff --git a/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go b/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go
index 9cf14cee39c..904833b7348 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/table_plan_builder.go
@@ -28,13 +28,21 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 )
 
+// This file contains just the builders for ReplicatorPlan and TablePlan.
+// ReplicatorPlan and TablePlan are in replicator_plan.go.
+// TODO(sougou): reorganize this in a better fashion.
+
 // ExcludeStr is the filter value for excluding tables that match a rule.
 // TODO(sougou): support this on vstreamer side also.
 const ExcludeStr = "exclude"
 
+// tablePlanBuilder contains the metadata needed for building a TablePlan.
 type tablePlanBuilder struct {
 	name       sqlparser.TableIdent
 	sendSelect *sqlparser.Select
+	// selColumns keeps track of the columns we want to pull from source.
+	// If Lastpk is set, we compare this list against the table's pk and
+	// add missing references.
 	selColumns map[string]bool
 	colExprs   []*colExpr
 	onInsert   insertType
@@ -43,7 +51,7 @@ type tablePlanBuilder struct {
 }
 
 // colExpr describes the processing to be performed to
-// compute the value of the target table column.
+// compute the value of one column of the target table.
 type colExpr struct {
 	colName sqlparser.ColIdent
 	// operation==opExpr: full expression is set
@@ -71,12 +79,22 @@ const (
 )
 
 // insertType describes the type of insert statement to generate.
+// Please refer to TestBuildPlayerPlan for examples.
 type insertType int
 
 // The following values are the various insert types.
 const (
+	// insertNormal is for normal selects without a group by, like
+	// "select a+b as c from t".
 	insertNormal = insertType(iota)
+	// insertOnDup is for the more traditional grouped expressions, like
+	// "select a, b, count(*) as c from t group by a". For statements
+	// like these, "insert.. on duplicate key" statements will be generated
+	// causing "b" to be updated to the latest value (last value wins).
 	insertOnDup
+	// insertIgnore is for special grouped expressions where all columns are
+	// in the group by, like "select a, b, c from t group by a, b, c".
+	// This generates "insert ignore" statements (first value wins).
 	insertIgnore
 )
 
@@ -85,6 +103,14 @@ const (
 // a table-specific rule is built to be sent to the source. We don't send the
 // original rule to the source because it may not match the same tables as the
 // target.
+// tableKeys specifies the list of primary key columns for each table.
+// copyState is a map of tables that have not been fully copied yet.
+// If a table is not present in copyState, then it has been fully copied. If so,
+// all replication events are applied. The table still has to match a Filter.Rule.
+// If it has a non-nil entry, then the value is the last primary key (lastpk)
+// that was copied.  If so, only replication events < lastpk are applied.
+// If the entry is nil, then copying of the table has not started yet. If so,
+// no events are applied.
 // The TablePlan built is a partial plan. The full plan for a table is built
 // when we receive field information from events or rows sent by the source.
 // buildExecutionPlan is the function that builds the full plan.
@@ -149,6 +175,7 @@ func MatchTable(tableName string, filter *binlogdatapb.Filter) (*binlogdatapb.Ru
 
 func buildTablePlan(tableName, filter string, tableKeys map[string][]string, lastpk *sqltypes.Result) (*TablePlan, error) {
 	query := filter
+	// generate equivalent select statement if filter is empty or a keyrange.
 	switch {
 	case filter == "":
 		buf := sqlparser.NewTrackedBuffer(nil)
@@ -170,6 +197,8 @@ func buildTablePlan(tableName, filter string, tableKeys map[string][]string, las
 	}
 
 	if expr, ok := sel.SelectExprs[0].(*sqlparser.StarExpr); ok {
+		// If it's a "select *", we return a partial plan, and complete
+		// it when we get back field info from the stream.
 		if len(sel.SelectExprs) != 1 {
 			return nil, fmt.Errorf("unexpected: %v", sqlparser.String(sel))
 		}
@@ -198,6 +227,13 @@ func buildTablePlan(tableName, filter string, tableKeys map[string][]string, las
 	if err := tpb.analyzeExprs(sel.SelectExprs); err != nil {
 		return nil, err
 	}
+	// It's possible that the target table does not materialize all
+	// the primary keys of the source table. In such situations,
+	// we still have to be able to validate the incoming event
+	// against the current lastpk. For this, we have to request
+	// the missing columns so we can compare against those values.
+	// If there is no lastpk to validate against, then we don't
+	// care.
 	if tpb.lastpk != nil {
 		for _, f := range tpb.lastpk.Fields {
 			tpb.addCol(sqlparser.NewColIdent(f.Name))
@@ -239,13 +275,13 @@ func (tpb *tablePlanBuilder) generate(tableKeys map[string][]string) *TablePlan
 	return &TablePlan{
 		TargetName:       tpb.name.String(),
 		Lastpk:           tpb.lastpk,
-		PKReferences:     pkrefs,
 		BulkInsertFront:  tpb.generateInsertPart(sqlparser.NewTrackedBuffer(bvf.formatter)),
 		BulkInsertValues: tpb.generateValuesPart(sqlparser.NewTrackedBuffer(bvf.formatter), bvf),
 		BulkInsertOnDup:  tpb.generateOnDupPart(sqlparser.NewTrackedBuffer(bvf.formatter)),
 		Insert:           tpb.generateInsertStatement(),
 		Update:           tpb.generateUpdateStatement(),
 		Delete:           tpb.generateDeleteStatement(),
+		PKReferences:     pkrefs,
 	}
 }
 
@@ -356,6 +392,7 @@ func (tpb *tablePlanBuilder) analyzeExpr(selExpr sqlparser.SelectExpr) (*colExpr
 		case *sqlparser.Subquery:
 			return false, fmt.Errorf("unsupported subquery: %v", sqlparser.String(node))
 		case *sqlparser.FuncExpr:
+			// Other aggregates are not supported.
 			if node.IsAggregate() {
 				return false, fmt.Errorf("unexpected: %v", sqlparser.String(node))
 			}
@@ -369,6 +406,8 @@ func (tpb *tablePlanBuilder) analyzeExpr(selExpr sqlparser.SelectExpr) (*colExpr
 	return cexpr, nil
 }
 
+// addCol adds the specified column to the send query
+// if it's not already present.
 func (tpb *tablePlanBuilder) addCol(ident sqlparser.ColIdent) {
 	if tpb.selColumns[ident.Lowered()] {
 		return
@@ -381,6 +420,7 @@ func (tpb *tablePlanBuilder) addCol(ident sqlparser.ColIdent) {
 
 func (tpb *tablePlanBuilder) analyzeGroupBy(groupBy sqlparser.GroupBy) error {
 	if groupBy == nil {
+		// If there's no grouping, the it's an insertNormal.
 		return nil
 	}
 	for _, expr := range groupBy {
@@ -397,9 +437,11 @@ func (tpb *tablePlanBuilder) analyzeGroupBy(groupBy sqlparser.GroupBy) error {
 		}
 		cexpr.isGrouped = true
 	}
+	// If all colExprs are grouped, then it's an insertIgnore.
 	tpb.onInsert = insertIgnore
 	for _, cExpr := range tpb.colExprs {
 		if !cExpr.isGrouped {
+			// If some colExprs are not grouped, then it's an insertOnDup.
 			tpb.onInsert = insertOnDup
 			break
 		}
@@ -407,6 +449,7 @@ func (tpb *tablePlanBuilder) analyzeGroupBy(groupBy sqlparser.GroupBy) error {
 	return nil
 }
 
+// analyzePK builds tpb.pkCols.
 func (tpb *tablePlanBuilder) analyzePK(tableKeys map[string][]string) error {
 	pkcols, ok := tableKeys[tpb.name.String()]
 	if !ok {
@@ -441,9 +484,12 @@ func (tpb *tablePlanBuilder) generateInsertStatement() *sqlparser.ParsedQuery {
 
 	tpb.generateInsertPart(buf)
 	if tpb.lastpk == nil {
+		// If there's no lastpk, generate straight values.
 		buf.Myprintf(" values ", tpb.name)
 		tpb.generateValuesPart(buf, bvf)
 	} else {
+		// If there is a lastpk, generate values as a select from dual
+		// where the pks < lastpk
 		tpb.generateSelectPart(buf, bvf)
 	}
 	tpb.generateOnDupPart(buf)
@@ -478,6 +524,7 @@ func (tpb *tablePlanBuilder) generateValuesPart(buf *sqlparser.TrackedBuffer, bv
 		case opCount:
 			buf.WriteString("1")
 		case opSum:
+			// NULL values must be treated as 0 for SUM.
 			buf.Myprintf("ifnull(%v, 0)", cexpr.expr)
 		}
 	}
@@ -513,12 +560,18 @@ func (tpb *tablePlanBuilder) generateOnDupPart(buf *sqlparser.TrackedBuffer) *sq
 	buf.Myprintf(" on duplicate key update ")
 	separator := ""
 	for _, cexpr := range tpb.colExprs {
+		// We don't know of a use case where the group by columns
+		// don't match the pk of a table. But we'll allow this,
+		// and won't update the pk column with the new value if
+		// this does happen. This can be revisited if there's
+		// a legitimate use case in the future that demands
+		// a different behavior. This rule is applied uniformly
+		// for updates and deletes also.
 		if cexpr.isGrouped || cexpr.isPK {
 			continue
 		}
 		buf.Myprintf("%s%v=", separator, cexpr.colName)
 		separator = ", "
-		// TODO: What to do here?
 		switch cexpr.operation {
 		case opExpr:
 			buf.Myprintf("values(%v)", cexpr.colName)
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vcopier.go b/go/vt/vttablet/tabletmanager/vreplication/vcopier.go
index 7e3b0fe4fc9..c0f9cc693b6 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vcopier.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vcopier.go
@@ -48,6 +48,10 @@ func newVCopier(vr *vreplicator) *vcopier {
 	}
 }
 
+// initTablesForCopy (phase 1) identifies the list of tables to be copied and inserts
+// them into copy_state. If there are no tables to copy, it explicitly stops
+// the stream. Otherwise, the copy phase (phase 2) may think that all tables are copied.
+// This will cause us to go into the replication phase (phase 3) without a starting position.
 func (vc *vcopier) initTablesForCopy(ctx context.Context) error {
 	defer vc.vr.dbClient.Rollback()
 
@@ -81,6 +85,23 @@ func (vc *vcopier) initTablesForCopy(ctx context.Context) error {
 	return vc.vr.dbClient.Commit()
 }
 
+// copyNext performs a multi-step process on each iteration.
+// Step 1: catchup: During this step, it replicates from the source from the last position.
+// This is a partial replication: events are applied only to tables or subsets of tables
+// that have already been copied. This goes on until replication catches up.
+// Step 2: Start streaming. This returns the initial field info along with the GTID
+// as of which the snapshot is being streamed.
+// Step 3: fastForward: The target is fast-forwarded to the GTID obtained. This should
+// be quick because we were mostly caught up as of step 1. This ensures that the
+// snapshot of the rows are consistent with the position where the target stopped.
+// Step 4: copy rows: Copy the next set of rows from the stream that was started in Step 2.
+// This goes on until all rows are copied, or a timeout. In both cases, copyNext
+// returns, and the replicator decides whether to invoke copyNext again, or to
+// go to the next phase if all the copying is done.
+// Steps 2, 3 and 4 are performed by copyTable.
+// copyNext also builds the copyState metadata that contains the tables and their last
+// primary key that was copied. A nil Result means that nothing has been copied.
+// A table that was fully copied is removed from copyState.
 func (vc *vcopier) copyNext(ctx context.Context, settings binlogplayer.VRSettings) error {
 	qr, err := vc.vr.dbClient.Execute(fmt.Sprintf("select table_name, lastpk from _vt.copy_state where vrepl_id=%d", vc.vr.id))
 	if err != nil {
@@ -112,6 +133,9 @@ func (vc *vcopier) copyNext(ctx context.Context, settings binlogplayer.VRSetting
 	return vc.copyTable(ctx, tableToCopy, copyState)
 }
 
+// catchup replays events to the subset of the tables that have been copied
+// until replication is caught up. In order to stop, the seconds behind master has
+// to fall below replicationLagTolerance.
 func (vc *vcopier) catchup(ctx context.Context, copyState map[string]*sqltypes.Result) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
@@ -159,6 +183,9 @@ func (vc *vcopier) catchup(ctx context.Context, copyState map[string]*sqltypes.R
 	}
 }
 
+// copyTable performs the synchronized copy of the next set of rows from
+// the current table being copied. Each packet received is transactionally
+// committed with the lastpk. This allows for consistent resumability.
 func (vc *vcopier) copyTable(ctx context.Context, tableName string, copyState map[string]*sqltypes.Result) error {
 	defer vc.vr.dbClient.Rollback()
 
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vcopier_flaky_test.go b/go/vt/vttablet/tabletmanager/vreplication/vcopier_flaky_test.go
index 837054ed1d4..9ee4648750b 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vcopier_flaky_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vcopier_flaky_test.go
@@ -156,12 +156,12 @@ func TestPlayerCopyBigTable(t *testing.T) {
 		vstreamRowsSendHook = nil
 	}
 
-	vstreamRowsHook = func(context.Context) {
+	vstreamHook = func(context.Context) {
 		// Sleeping 50ms guarantees that the catchup wait loop executes multiple times.
 		// This is because waitRetryTime is set to 10ms.
 		time.Sleep(50 * time.Millisecond)
 		// Do this no more than once.
-		vstreamRowsHook = nil
+		vstreamHook = nil
 	}
 
 	filter := &binlogdatapb.Filter{
@@ -283,12 +283,12 @@ func TestPlayerCopyWildcardRule(t *testing.T) {
 		vstreamRowsSendHook = nil
 	}
 
-	vstreamRowsHook = func(context.Context) {
+	vstreamHook = func(context.Context) {
 		// Sleeping 50ms guarantees that the catchup wait loop executes multiple times.
 		// This is because waitRetryTime is set to 10ms.
 		time.Sleep(50 * time.Millisecond)
 		// Do this no more than once.
-		vstreamRowsHook = nil
+		vstreamHook = nil
 	}
 
 	filter := &binlogdatapb.Filter{
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vplayer.go b/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
index 85159a8bfb8..a6b85772662 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
@@ -33,6 +33,7 @@ import (
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 )
 
+// vplayer replays binlog events by pulling them from a vstreamer.
 type vplayer struct {
 	vr        *vreplicator
 	startPos  mysql.Position
@@ -44,8 +45,10 @@ type vplayer struct {
 	tablePlans     map[string]*TablePlan
 
 	pos mysql.Position
-	// unsavedEvent is saved any time we skip an event without
-	// saving: This can be an empty commit or a skipped DDL.
+	// unsavedEvent is set any time we skip an event without
+	// saving, which is on an empty commit.
+	// If nothing else happens for idleTimeout since timeLastSaved,
+	// the position of the unsavedEvent gets saved.
 	unsavedEvent *binlogdatapb.VEvent
 	// timeLastSaved is set every time a GTID is saved.
 	timeLastSaved time.Time
@@ -53,10 +56,19 @@ type vplayer struct {
 	lastTimestampNs int64
 	// timeOffsetNs keeps track of the clock difference with respect to source tablet.
 	timeOffsetNs int64
-	// canAcceptStmtEvents set to true if the current player can accept events in statement mode. Only true for filters that are match all.
+	// canAcceptStmtEvents is set to true if the current player can accept events in statement mode. Only true for filters that are match all.
 	canAcceptStmtEvents bool
 }
 
+// newVPlayer creates a new vplayer. Parameters:
+// vreplicator: the outer replicator. It's used for common functions like setState.
+//   Also used to access the engine for registering journal events.
+// settings: current settings read from _vt.vreplication.
+// copyState: if set, contains the list of tables yet to be copied, or in the process
+//   of being copied. If copyState is non-nil, the plans generated make sure that
+//   replication is only applied to parts that have been copied so far.
+// pausePos: if set, replication will stop at that position without updating the state to "Stopped".
+//   This is used by the fastForward function during copying.
 func newVPlayer(vr *vreplicator, settings binlogplayer.VRSettings, copyState map[string]*sqltypes.Result, pausePos mysql.Position) *vplayer {
 	saveStop := true
 	if !pausePos.IsZero() {
@@ -75,7 +87,7 @@ func newVPlayer(vr *vreplicator, settings binlogplayer.VRSettings, copyState map
 	}
 }
 
-// play is not resumable. If pausePos is set, play returns without updating the vreplication state.
+// play is the entry point for playing binlogs.
 func (vp *vplayer) play(ctx context.Context) error {
 	if !vp.stopPos.IsZero() && vp.startPos.AtLeast(vp.stopPos) {
 		log.Infof("Stop position %v already reached: %v", vp.startPos, vp.stopPos)
@@ -100,20 +112,19 @@ func (vp *vplayer) play(ctx context.Context) error {
 		}
 	}
 
-	if err := vp.fetchAndApply(ctx); err != nil {
-		msg := err.Error()
-		vp.vr.stats.History.Add(&binlogplayer.StatsHistoryRecord{
-			Time:    time.Now(),
-			Message: msg,
-		})
-		if err := vp.vr.setMessage(msg); err != nil {
-			log.Errorf("Failed to set error state: %v", err)
-		}
-		return err
-	}
-	return nil
+	return vp.fetchAndApply(ctx)
 }
 
+// fetchAndApply performs the fetching and application of the binlogs.
+// This is done by two different threads. The fetcher thread pulls
+// events from the vstreamer and adds them to the relayLog.
+// The applyEvents thread pulls accumulated events from the relayLog
+// to apply them to mysql. The reason for this separation is because
+// commits are slow during apply. So, more events can accumulate in
+// the relay log during a commit. In such situations, the next iteration
+// of apply combines all the transactions in the relay log into a single
+// one. This allows for the apply thread to catch up more quickly if
+// a backlog builds up.
 func (vp *vplayer) fetchAndApply(ctx context.Context) (err error) {
 	log.Infof("Starting VReplication player id: %v, startPos: %v, stop: %v, filter: %v", vp.vr.id, vp.startPos, vp.stopPos, vp.vr.source)
 
@@ -224,6 +235,54 @@ func (vp *vplayer) updatePos(ts int64) (posReached bool, err error) {
 	return posReached, nil
 }
 
+// applyEvents is the main thread that applies the events. It has the following use
+// cases to take into account:
+// * Normal transaction that has row mutations. In this case, the transaction
+//   is committed along with an update of the position.
+// * DDL event: the action depends on the OnDDL setting.
+// * OTHER event: the current position of the event is saved.
+// * JOURNAL event: if the event is relevant to the current stream, invoke registerJournal
+//   of the engine, and terminate.
+// * HEARTBEAT: update SecondsBehindMaster.
+// * Empty transaction: The event is remembered as an unsavedEvent. If no commits
+//   happen for idleTimeout since timeLastSaved, the current position of the unsavedEvent
+//   is committed (updatePos).
+// * An empty transaction: Empty transactions are necessary because the current
+//   position of that transaction may be the stop position. If so, we have to record it.
+//   If not significant, we should avoid saving these empty transactions individually
+//   because they can cause unnecessary churn and binlog bloat. We should
+//   also not go for too long without saving because we should not fall way behind
+//   on the current replication position. Additionally, WaitForPos or other external
+//   agents could be waiting on that specific position by watching the vreplication
+//   record.
+// * A group of transactions: Combine them into a single transaction.
+// * Partial transaction: Replay the events received so far and refetch from relay log
+//   for more.
+// * A combination of any of the above: The trickier case is the one where a group
+//   of transactions come in, with the last one being partial. In this case, all transactions
+//   up to the last one have to be committed, and the final one must be partially applied.
+//
+// Of the above events, the saveable ones are COMMIT, DDL, and OTHER. Eventhough
+// A GTID comes as a separate event, it's not saveable until a subsequent saveable
+// event occurs. VStreamer currently sequences the GTID to be sent just before
+// a saveable event, but we do not rely on this. To handle this, we only remember
+// the position when a GTID is encountered. The next saveable event causes the
+// current position to be saved.
+//
+// In order to handle the above use cases, we use an implicit transaction scheme:
+// A BEGIN does not really start a transaction. Ony a ROW event does. With this
+// approach, no transaction gets started if an empty one arrives. If a we receive
+// a commit, and a we are not in a transaction, we infer that the transaction was
+// empty, and remember it as an unsaved event. The next GTID event will reset the
+// unsaved event. If the next commit is also an empty transaction, then the latest
+// one gets remembered as unsaved. A non-empty transaction, a must-save event,
+// or a timeout will eventually cause the next save.
+// The timeout (1s) plays another significant role: If the source and target shards of
+// the replication are the same, then a commit of an unsaved event will generate
+// another empty event. This is an infinite loop, and the timeout prevents
+// this from becoming a tight loop.
+// TODO(sougou): we can look at recognizing self-generated events and find a better
+// way to handle them.
 func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 	defer vp.vr.dbClient.Rollback()
 
@@ -242,19 +301,12 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 			behind := time.Now().UnixNano() - vp.lastTimestampNs - vp.timeOffsetNs
 			vp.vr.stats.SecondsBehindMaster.Set(behind / 1e9)
 		}
-		// Filtered replication often ends up receiving a large number of empty transactions.
-		// This is required because the player needs to know the latest position of the source.
-		// This allows it to stop at that position if requested.
-		// This position also needs to be saved, which will allow an external request
-		// to check if a required position has been reached.
-		// However, this leads to a large number of empty commits which not only slow
-		// down the replay, but also generate binlog bloat on the target.
-		// In order to mitigate this problem, empty transactions are saved at most
-		// once every idleTimeout.
+		// Empty transactions are saved at most once every idleTimeout.
 		// This covers two situations:
 		// 1. Fetch was idle for idleTimeout.
 		// 2. We've been receiving empty events for longer than idleTimeout.
-		// In both cases, now > timeLastSaved. If so, any unsaved GTID should be saved.
+		// In both cases, now > timeLastSaved. If so, the GTID of the last unsavedEvent
+		// must be saved.
 		if time.Since(vp.timeLastSaved) >= idleTimeout && vp.unsavedEvent != nil {
 			posReached, err := vp.updatePos(vp.unsavedEvent.Timestamp)
 			if err != nil {
@@ -275,10 +327,18 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 				mustSave := false
 				switch event.Type {
 				case binlogdatapb.VEventType_COMMIT:
+					// If we've reached the stop position, we must save the current commit
+					// even if it's empty. So, the next applyEvent is invoked with the
+					// mustSave flag.
 					if !vp.stopPos.IsZero() && vp.pos.AtLeast(vp.stopPos) {
 						mustSave = true
 						break
 					}
+					// In order to group multiple commits into a single one, we look ahead for
+					// the next commit. If there is one, we skip the current commit, which ends up
+					// applying the next set of events as part of the current transaction. This approach
+					// also handles the case where the last transaction is partial. In that case,
+					// we only group the transactions with commits we've seen so far.
 					if hasAnotherCommit(items, i, j+1) {
 						continue
 					}
@@ -294,7 +354,7 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 func hasAnotherCommit(items [][]*binlogdatapb.VEvent, i, j int) bool {
 	for i < len(items) {
 		for j < len(items[i]) {
-			// We ignore GTID, BEGIN, FIELD and ROW.
+			// We skip GTID, BEGIN, FIELD, ROW and DMLs.
 			switch items[i][j].Type {
 			case binlogdatapb.VEventType_COMMIT:
 				return true
@@ -318,7 +378,7 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 			return err
 		}
 		vp.pos = pos
-		// A new position should not be saved until a commit or DDL.
+		// A new position should not be saved until a saveable event occurs.
 		vp.unsavedEvent = nil
 		if vp.stopPos.IsZero() {
 			return nil
@@ -423,6 +483,10 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 			}
 			return io.EOF
 		case binlogdatapb.OnDDLAction_EXEC:
+			// It's impossible to save the position transactionally with the statement.
+			// So, we apply the DDL first, and then save the position.
+			// Manual intervention may be needed if there is a partial
+			// failure here.
 			if _, err := vp.vr.dbClient.ExecuteWithRetry(ctx, event.Ddl); err != nil {
 				return err
 			}
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vreplicator.go b/go/vt/vttablet/tabletmanager/vreplication/vreplicator.go
index 0e16bccdcce..e22dc9638a1 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vreplicator.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vreplicator.go
@@ -59,7 +59,26 @@ type vreplicator struct {
 	tableKeys map[string][]string
 }
 
-// newVReplicator creates a new vreplicator
+// newVReplicator creates a new vreplicator. The valid fields from the source are:
+// Keyspce, Shard, Filter, OnDdl, ExternalMySql and StopAfterCopy.
+// The Filter consists of Rules. Each Rule has a Match and an (inner) Filter field.
+// The Match can be a table name or, if it begins with a "/", a wildcard.
+// The Filter can be empty: get all rows and columns.
+// The Filter can be a keyrange, like "-80": get all rows that are within the keyrange.
+// The Filter can be a select expression. Examples.
+//   "select * from t", same as an empty Filter,
+//   "select * from t where in_keyrange('-80')", same as "-80",
+//   "select * from t where in_keyrange(col1, 'hash', '-80')",
+//   "select col1, col2 from t where...",
+//   "select col1, keyspace_id() as ksid from t where...",
+//   "select id, count(*), sum(price) from t group by id".
+//   Only "in_keyrange" expressions are supported in the where clause.
+//   The select expressions can be any valid non-aggregate expressions,
+//   or count(*), or sum(col).
+//   If the target column name does not match the source expression, an
+//   alias like "a+b as targetcol" must be used.
+//   More advanced constructs can be used. Please see the table plan builder
+//   documentation for more info.
 func newVReplicator(id uint32, source *binlogdatapb.BinlogSource, sourceVStreamer VStreamerClient, stats *binlogplayer.Stats, dbClient binlogplayer.DBClient, mysqld mysqlctl.MysqlDaemon, vre *Engine) *vreplicator {
 	return &vreplicator{
 		vre:             vre,
@@ -72,8 +91,38 @@ func newVReplicator(id uint32, source *binlogdatapb.BinlogSource, sourceVStreame
 	}
 }
 
-// Replicate starts a vreplication stream.
+// Replicate starts a vreplication stream. It can be in one of three phases:
+// 1. Init: If a request is issued with no starting position, we assume that the
+// contents of the tables must be copied first. During this phase, the list of
+// tables to be copied is inserted into the copy_state table. A successful insert
+// gets us out of this phase.
+// 2. Copy: If the copy_state table has rows, then we are in this phase. During this
+// phase, we repeatedly invoke copyNext until all the tables are copied. After each
+// table is successfully copied, it's removed from the copy_state table. We exit this
+// phase when there are no rows left in copy_state.
+// 3. Replicate: In this phase, we replicate binlog events indefinitely, unless
+// a stop position was requested. This phase differs from the Init phase because
+// there is a replication position.
+// If a request had a starting position, then we go directly into phase 3.
+// During these phases, the state of vreplication is reported as 'Init', 'Copying',
+// or 'Running'. They all mean the same thing. The difference in the phases depends
+// on the criteria defined above. The different states reported are mainly
+// informational. The 'Stopped' state is, however, honored.
+// All phases share the same plan building framework. We leverage the fact the
+// row representation of a read (during copy) and a binlog event are identical.
+// However, there are some subtle differences, explained in the plan builder
+// code.
 func (vr *vreplicator) Replicate(ctx context.Context) error {
+	err := vr.replicate(ctx)
+	if err != nil {
+		if err := vr.setMessage(err.Error()); err != nil {
+			log.Errorf("Failed to set error state: %v", err)
+		}
+	}
+	return err
+}
+
+func (vr *vreplicator) replicate(ctx context.Context) error {
 	tableKeys, err := vr.buildTableKeys()
 	if err != nil {
 		return err
@@ -179,7 +228,18 @@ func (vr *vreplicator) setMessage(message string) error {
 }
 
 func (vr *vreplicator) setState(state, message string) error {
-	return binlogplayer.SetVReplicationState(vr.dbClient, vr.id, state, message)
+	if message != "" {
+		vr.stats.History.Add(&binlogplayer.StatsHistoryRecord{
+			Time:    time.Now(),
+			Message: message,
+		})
+	}
+	vr.stats.State.Set(state)
+	query := fmt.Sprintf("update _vt.vreplication set state='%v', message=%v where id=%v", state, encodeString(binlogplayer.MessageTruncate(message)), vr.id)
+	if _, err := vr.dbClient.ExecuteFetch(query, 1); err != nil {
+		return fmt.Errorf("could not set state: %v: %v", query, err)
+	}
+	return nil
 }
 
 func encodeString(in string) string {