Merge #143424 #143463

143424: raft: use term cache for leader commit term check r=pav-kv a=hakuuww

Previously, we avoided calling term(index) for leader commit term check by keeping the first entry index for the current leader term in memory, and comparing against that index. (see #137826)

Now, since we have implemented term cache
(which stores info of the current leader term's first entryID in this case), the term cache is used for that check to reduce the complexity of our code. (see #142239 )

Part of #136296
Fixes: #143362
Epic: None
Release note: None

143463: sql: refactor PlanCDCExpression r=mgartner a=mgartner

#### sql: collect CDC presentation outside of plan walker

The column presentation of the top node is collected to determine the
output columns for the CDC expression. There is no need to do this
within the plan node walker, so it has been moved outside.

Release note: None

#### sql: use InputCount and Input planNode methods to walk CDC plans

The `Input` and `InputCount` methods of the `planNode` interface are now
used to walk plan node trees of CDC expressions. This continues the
effort to deprecate and remove the plan node walkers (see #137620 for
more details on the motivation for this).

Epic: None
Release note: None

#### sql: refactor return statements in PlanCDCExpression

The `return` statements for error cases now explicitly return an empty
`CDCExpressionPlan` for clarity.

Release note: None


Co-authored-by: Anthony Xu <[email protected]>
Co-authored-by: Marcus Gartner <[email protected]>

Author: 3 people

pkg/raft/raft.go

@@ -335,16 +335,6 @@ type raft struct {
 
 	state pb.StateType
 
-	// idxPreLeading is the last log index as of when this node became the
-	// leader. Separates entries proposed by previous leaders from the entries
-	// proposed by the current leader. Used only in StateLeader, and updated
-	// when entering StateLeader (in becomeLeader()).
-	//
-	// Invariants (when in StateLeader at raft.Term):
-	//	- entries at indices <= idxPreLeading have term < raft.Term
-	//	- entries at indices > idxPreLeading have term == raft.Term
-	idxPreLeading uint64
-
 	// isLearner is true if the local raft node is a learner.
 	isLearner bool
 
@@ -1047,7 +1037,9 @@ func (r *raft) maybeCommit() bool {
 	// This comparison is equivalent in output to:
 	// if !r.raftLog.matchTerm(entryID{term: r.Term, index: index})
 	// But avoids (potentially) loading the entry term from storage.
-	if index <= r.idxPreLeading {
+	// termCache.last() stores the first entryID added to raftLog in the
+	// current leader term by invariants.
+	if index < r.raftLog.termCache.last().index {
 		return false
 	}
 
@@ -1386,11 +1378,6 @@ func (r *raft) becomeLeader() {
 	// could be expensive.
 	r.pendingConfIndex = r.raftLog.lastIndex()
 
-	// Remember the last log index before the term advances to
-	// our current (leader) term.
-	// See the idxPreLeading comment for more details.
-	r.idxPreLeading = r.raftLog.lastIndex()
-
 	emptyEnt := pb.Entry{Data: nil}
 	if !r.appendEntry(emptyEnt) {
 		// This won't happen because we just called reset() above.

pkg/raft/raft_test.go

@@ -2592,26 +2592,34 @@ func TestLeaderAppResp(t *testing.T) {
 		// Follower 2 responds to leader, indicating log index 2 is replicated.
 		// Leader tries to commit, but commit index doesn't advance since the index
 		// is from a previous term.
-		// We hit maybeCommit() and do term check comparison by using the invariant
-		// raft.idxPreLeading.
+		// We hit maybeCommit() and do term check comparison by using the
+		// last "term flip" entryID stored in the termCache.
 		// There is no storage access for term in the maybeCommit() code path
 		{2, false, 2, 7, 1, 2, 0, 2},
 
+		// Follower 2 responds to leader, indicating log index 3 is replicated.
+		// Leader tries to commit, but commit index doesn't advance since the index
+		// is from a previous term. Same as above.
+		{3, false, 3, 7, 1, 3, 0, 1},
+
 		// NB: For the following tests, we are skipping the MsgAppResp for the first
 		// 3 entries, by directly processing MsgAppResp for later entries.
 		//
 		// Follower 2 is StateProbing at 4, it sends MsgAppResp for 4, and is moved
 		// to StateReplicate and as many entries as possible are sent to it (5, 6).
 		// Correspondingly the Next is then 7 (entry 7 does not exist, indicating
 		// the follower will be up to date should it process the emitted MsgApp).
-		// accept resp; leader commits; respond with commit index
+		// accept resp; leader commits; respond with commit index.
+		// maybeCommit() is successful.
 		{4, false, 4, 7, 1, 4, 4, 1},
 
 		// Follower 2 says term2, index5 is already replicated.
 		// The leader responds with the updated commit index to follower 2.
+		// maybeCommit() is successful.
 		{5, false, 5, 7, 1, 5, 5, 1},
 		// Follower 2 says term2, index6 is already replicated.
 		// The leader responds with the updated commit index to follower 2.
+		// maybeCommit() is successful.
 		{6, false, 6, 7, 1, 6, 6, 1},
 	} {
 		t.Run("", func(t *testing.T) {

pkg/sql/distsql_plan_changefeed.go

@@ -94,7 +94,7 @@ func PlanCDCExpression(
 
 	familyID, err := extractFamilyID(cdcExpr)
 	if err != nil {
-		return cdcPlan, err
+		return CDCExpressionPlan{}, err
 	}
 
 	cdcCat := &cdcOptCatalog{
@@ -110,7 +110,7 @@ func PlanCDCExpression(
 
 	memo, err := opc.buildExecMemo(ctx)
 	if err != nil {
-		return cdcPlan, err
+		return CDCExpressionPlan{}, err
 	}
 	if log.V(2) {
 		log.Infof(ctx, "Optimized CDC expression: %s", memo)
@@ -122,55 +122,61 @@ func PlanCDCExpression(
 		ctx, &p.curPlan, &p.stmt, newExecFactory(ctx, p), memo, p.SemaCtx(),
 		p.EvalContext(), allowAutoCommit, disableTelemetryAndPlanGists,
 	); err != nil {
-		return cdcPlan, err
+		return CDCExpressionPlan{}, err
+	}
+
+	// The top node contains the list of columns to return.
+	presentation := planColumns(p.curPlan.main.planNode)
+	if len(presentation) == 0 {
+		return CDCExpressionPlan{}, errors.AssertionFailedf("unable to determine result columns")
 	}
 
 	// Walk the plan, perform sanity checks and extract information we need.
 	var spans roachpb.Spans
-	var presentation colinfo.ResultColumns
-
-	if err := walkPlan(ctx, p.curPlan.main.planNode, planObserver{
-		enterNode: func(ctx context.Context, nodeName string, plan planNode) (bool, error) {
-			switch n := plan.(type) {
-			case *scanNode:
-				// Collect spans we wanted to scan.  The select statement used for this
-				// plan should result in a single table scan of primary index span.
-				if len(spans) > 0 {
-					return false, errors.AssertionFailedf("unexpected multiple primary index scan operations")
+	var validatePlanAndCollectSpans func(p planNode) error
+	validatePlanAndCollectSpans = func(p planNode) error {
+		switch n := p.(type) {
+		case *scanNode:
+			// Collect spans we wanted to scan. The select statement used for
+			// this plan should result in a single table scan of primary index
+			// span.
+			if len(spans) > 0 {
+				return errors.AssertionFailedf("unexpected multiple primary index scan operations")
+			}
+			if n.index.GetID() != n.desc.GetPrimaryIndexID() {
+				return errors.AssertionFailedf(
+					"expect scan of primary index, found scan of %d", n.index.GetID())
+			}
+			spans = n.spans
+		case *zeroNode:
+			return errors.Newf(
+				"changefeed expression %s does not match any rows", tree.AsString(cdcExpr))
+		default:
+			// Recurse into input nodes.
+			for i, n := 0, p.InputCount(); i < n; i++ {
+				input, err := p.Input(i)
+				if err != nil {
+					return err
 				}
-				if n.index.GetID() != n.desc.GetPrimaryIndexID() {
-					return false, errors.AssertionFailedf(
-						"expect scan of primary index, found scan of %d", n.index.GetID())
+				if err = validatePlanAndCollectSpans(input); err != nil {
+					return err
 				}
-				spans = n.spans
-			case *zeroNode:
-				return false, errors.Newf(
-					"changefeed expression %s does not match any rows", tree.AsString(cdcExpr))
 			}
-
-			// Because the walk is top down, the top node is the node containing the
-			// list of columns to return.
-			if len(presentation) == 0 {
-				presentation = planColumns(plan)
-			}
-			return true, nil
-		},
-	}); err != nil {
-		return cdcPlan, err
+		}
+		return nil
+	}
+	if err := validatePlanAndCollectSpans(p.curPlan.main.planNode); err != nil {
+		return CDCExpressionPlan{}, err
 	}
 
 	if len(spans) == 0 {
 		// Should have been handled by the zeroNode check above.
-		return cdcPlan, errors.AssertionFailedf("expected at least 1 span to scan")
-	}
-
-	if len(presentation) == 0 {
-		return cdcPlan, errors.AssertionFailedf("unable to determine result columns")
+		return CDCExpressionPlan{}, errors.AssertionFailedf("expected at least 1 span to scan")
 	}
 
 	if len(p.curPlan.subqueryPlans) > 0 || len(p.curPlan.cascades) > 0 ||
 		len(p.curPlan.checkPlans) > 0 || len(p.curPlan.triggers) > 0 {
-		return cdcPlan, errors.AssertionFailedf("unexpected query structure")
+		return CDCExpressionPlan{}, errors.AssertionFailedf("unexpected query structure")
 	}
 
 	planCtx := p.DistSQLPlanner().NewPlanningCtx(ctx, &p.extendedEvalCtx, p, p.txn, LocalDistribution)