[#19649, #19664, #19408]Fix bugs (#19675)

1.  Cherry-pick 2.0-dev

Approved by: @XuPeng-SH, @sukki37

Author: triump2020

pkg/vm/engine/disttae/logtail_consumer.go

@@ -1891,8 +1891,6 @@ func updatePartitionOfPush(
 		if len(tl.CkpLocation) > 0 {
 			t0 = time.Now()
 			ckpStart, ckpEnd = parseCkpDuration(tl)
-			state.CacheCkpDuration(ckpStart, partition)
-			state.AppendCheckpoint(tl.CkpLocation, partition)
 			v2.LogtailUpdatePartitonHandleCheckpointDurationHistogram.Observe(time.Since(t0).Seconds())
 		}
 
@@ -1925,20 +1923,26 @@ func updatePartitionOfPush(
 	}
 
 	//After consume checkpoints finished ,then update the start and end of
-	//the mo system table's partition and catalog.
-	if !lazyLoad {
+	//the table's partition and catalog cache.
+	if isSub {
 		if len(tl.CkpLocation) != 0 {
 			if !ckpStart.IsEmpty() || !ckpEnd.IsEmpty() {
 				t0 = time.Now()
 				state.UpdateDuration(ckpStart, types.MaxTs())
-				//Notice that the checkpoint duration is same among all mo system tables,
-				//such as mo_databases, mo_tables, mo_columns.
-				e.GetLatestCatalogCache().UpdateDuration(ckpStart, types.MaxTs())
+				if lazyLoad {
+					state.AppendCheckpoint(tl.CkpLocation, partition)
+				} else {
+					//Notice that the checkpoint duration is same among all mo system tables,
+					//such as mo_databases, mo_tables, mo_columns.
+					e.GetLatestCatalogCache().UpdateDuration(ckpStart, types.MaxTs())
+				}
 				v2.LogtailUpdatePartitonUpdateTimestampsDurationHistogram.Observe(time.Since(t0).Seconds())
 			}
 		} else {
 			state.UpdateDuration(types.TS{}, types.MaxTs())
-			e.GetLatestCatalogCache().UpdateDuration(types.TS{}, types.MaxTs())
+			if !lazyLoad {
+				e.GetLatestCatalogCache().UpdateDuration(types.TS{}, types.MaxTs())
+			}
 		}
 	}
 

pkg/vm/engine/disttae/logtailreplay/partition.go

@@ -158,41 +158,29 @@ func (p *Partition) ConsumeCheckpoints(
 		return nil
 	}
 
-	//curState := p.state.Load()
-	//if len(curState.checkpoints) == 0 {
-	//	p.UpdateDuration(types.TS{}, types.MaxTs())
-	//	return nil
-	//}
+	curState := p.state.Load()
+	if len(curState.checkpoints) == 0 {
+		return nil
+	}
 
 	lockErr := p.Lock(ctx)
 	if lockErr != nil {
 		return lockErr
 	}
 	defer p.Unlock()
 
-	curState := p.state.Load()
-	//if len(curState.checkpoints) == 0 {
-	//	p.UpdateDuration(types.TS{}, types.MaxTs())
-	//	return nil
-	//}
-
-	state := curState.Copy()
-
-	if len(state.checkpoints) == 0 {
-		state.UpdateDuration(types.TS{}, types.MaxTs())
-		if !p.state.CompareAndSwap(curState, state) {
-			panic("concurrent mutation")
-		}
+	curState = p.state.Load()
+	if len(curState.checkpoints) == 0 {
 		return nil
 	}
 
+	state := curState.Copy()
+
 	//consume checkpoints.
 	if err := state.consumeCheckpoints(fn); err != nil {
 		return err
 	}
 
-	state.UpdateDuration(state.start, types.MaxTs())
-
 	if !p.state.CompareAndSwap(curState, state) {
 		panic("concurrent mutation")
 	}