feat: Enable Stats on Resumed Sync

destination/writers.go

@@ -54,6 +54,7 @@ type (
 		batchSize      int64
 		streamArtifact *writerSchema
 		group          *utils.CxGroup
+		committedCount int64
 	}
 )
 
@@ -249,6 +250,8 @@ func (wt *WriterThread) flush(ctx context.Context, buf []types.RawRecord) (err e
 	}
 
 	logger.Infof("Thread[%s]: successfully wrote %d records", wt.threadID, len(buf))
+	wt.committedCount += int64(len(buf)) // Track successfully committed records
+
 	return nil
 }
 
@@ -274,6 +277,10 @@ func (wt *WriterThread) Close(ctx context.Context) error {
 	}
 }
 
+func (wt *WriterThread) GetCommittedCount() int64 {
+	return wt.committedCount
+}
+
 func ClearDestination(ctx context.Context, config *types.WriterConfig, dropStreams []types.StreamInterface) error {
 	newfunc, found := RegisteredWriters[config.Type]
 	if !found {

drivers/abstract/backfill.go

@@ -24,6 +24,12 @@ func (a *AbstractDriver) Backfill(ctx context.Context, backfilledStreams chan st
 		}
 		// set state chunks
 		a.state.SetChunks(stream.Self(), chunksSet)
+
+		// Persist remaining record count from pool stats to state for resume capability
+		// Note: The initial count has already been added to pool stats before logger started
+		if pool.GetStats().TotalRecordsToSync.Load() > 0 && a.state != nil {
+			a.state.SetRemainingRecordCount(stream.Self(), pool.GetStats().TotalRecordsToSync.Load())
+		}
 	}
 	chunks := chunksSet.Array()
 	if len(chunks) == 0 {
@@ -37,7 +43,8 @@ func (a *AbstractDriver) Backfill(ctx context.Context, backfilledStreams chan st
 	sort.Slice(chunks, func(i, j int) bool {
 		return typeutils.Compare(chunks[i].Min, chunks[j].Min) < 0
 	})
-	logger.Infof("Starting backfill for stream[%s] with %d chunks", stream.GetStream().Name, len(chunks))
+	logger.Infof("Processing backfill for stream[%s] with %d chunks", stream.GetStream().Name, len(chunks))
+
 	// TODO: create writer instance again on retry
 	chunkProcessor := func(ctx context.Context, chunk types.Chunk) (err error) {
 		threadID := fmt.Sprintf("%s_%s", stream.ID(), utils.ULID())
@@ -60,6 +67,12 @@ func (a *AbstractDriver) Backfill(ctx context.Context, backfilledStreams chan st
 			if err == nil {
 				logger.Infof("finished chunk min[%v] and max[%v] of stream %s", chunk.Min, chunk.Max, stream.ID())
 				chunksLeft := a.state.RemoveChunk(stream.Self(), chunk)
+
+				// Decrement remaining record count by committed records
+				committedCount := inserter.GetCommittedCount()
+				a.state.DecrementRemainingRecordCount(stream.Self(), committedCount)
+				logger.Infof("Stream %s: chunk completed with %d committed records", stream.ID(), committedCount)
+
 				if chunksLeft == 0 && backfilledStreams != nil {
 					backfilledStreams <- stream.ID()
 				}

protocol/sync.go

@@ -115,14 +115,34 @@ var syncCmd = &cobra.Command{
 			return err
 		}
 
+		// Setup state early to enable pre-loading remaining records before stats logger starts
+		connector.SetupState(state)
+
+		// Pre-load remaining record counts from state to pool stats for accurate progress tracking
+		for _, stream := range selectedStreamsMetadata.FullLoadStreams {
+			if remainingRecords := state.GetRemainingRecordCount(stream.Self()); remainingRecords > 0 {
+				pool.AddRecordsToSyncStats(remainingRecords)
+				logger.Infof("Pre-loaded remaining records for stream %s: %d", stream.ID(), remainingRecords)
+			}
+		}
+		for _, stream := range selectedStreamsMetadata.CDCStreams {
+			if remainingRecords := state.GetRemainingRecordCount(stream.Self()); remainingRecords > 0 {
+				pool.AddRecordsToSyncStats(remainingRecords)
+				logger.Infof("Pre-loaded remaining records for stream %s: %d", stream.ID(), remainingRecords)
+			}
+		}
+		for _, stream := range selectedStreamsMetadata.IncrementalStreams {
+			if remainingRecords := state.GetRemainingRecordCount(stream.Self()); remainingRecords > 0 {
+				pool.AddRecordsToSyncStats(remainingRecords)
+				logger.Infof("Pre-loaded remaining records for stream %s: %d", stream.ID(), remainingRecords)
+			}
+		}
+
 		// start monitoring stats
 		logger.StatsLogger(cmd.Context(), func() (int64, int64, int64) {
 			stats := pool.GetStats()
 			return stats.ThreadCount.Load(), stats.TotalRecordsToSync.Load(), stats.ReadCount.Load()
 		})
-
-		// Setup State for Connector
-		connector.SetupState(state)
 		// Sync Telemetry tracking
 		telemetry.TrackSyncStarted(syncID, streams, selectedStreamsMetadata.SelectedStreams, selectedStreamsMetadata.CDCStreams, connector.Type(), destinationConfig, catalog)
 		defer func() {

types/state.go

@@ -24,6 +24,8 @@ const (
 	MixedType StateType = "MIXED"
 	// constant key for chunks
 	ChunksKey = "chunks"
+	// number of remaining records to be synced in resumable sync
+	RemainingRecordsCount = "remaining_records_count"
 )
 
 type GlobalState struct {
@@ -254,6 +256,81 @@ func (s *State) RemoveChunk(stream *ConfiguredStream, chunk Chunk) int {
 	return -1
 }
 
+// GetRemainingRecordCount retrieves the remaining record count for a stream from state
+func (s *State) GetRemainingRecordCount(stream *ConfiguredStream) int64 {
+	s.RLock()
+	defer s.RUnlock()
+
+	index, contains := utils.ArrayContains(s.Streams, func(elem *StreamState) bool {
+		return elem.Namespace == stream.Namespace() && elem.Stream == stream.Name()
+	})
+	if contains {
+		if count, loaded := s.Streams[index].State.Load(RemainingRecordsCount); loaded {
+			if countInt64, ok := count.(int64); ok {
+				return countInt64
+			}
+			if countFloat64, ok := count.(float64); ok {
+				return int64(countFloat64)
+			}
+		}
+	}
+	return 0
+}
+
+// SetRemainingRecordCount stores the remaining record count for a stream in state
+func (s *State) SetRemainingRecordCount(stream *ConfiguredStream, count int64) {
+	s.Lock()
+	defer s.Unlock()
+
+	index, contains := utils.ArrayContains(s.Streams, func(elem *StreamState) bool {
+		return elem.Namespace == stream.Namespace() && elem.Stream == stream.Name()
+	})
+	if contains {
+		s.Streams[index].State.Store(RemainingRecordsCount, count)
+		s.Streams[index].HoldsValue.Store(true)
+	} else {
+		newStream := s.initStreamState(stream)
+		newStream.State.Store(RemainingRecordsCount, count)
+		newStream.HoldsValue.Store(true)
+		s.Streams = append(s.Streams, newStream)
+	}
+	s.LogState()
+}
+
+// DecrementRemainingRecordCount decrements the remaining record count for a stream in state
+func (s *State) DecrementRemainingRecordCount(stream *ConfiguredStream, count int64) {
+	s.Lock()
+	defer s.Unlock()
+
+	index, contains := utils.ArrayContains(s.Streams, func(elem *StreamState) bool {
+		return elem.Namespace == stream.Namespace() && elem.Stream == stream.Name()
+	})
+	if contains {
+		if remaining, loaded := s.Streams[index].State.Load(RemainingRecordsCount); loaded {
+			var currentRemaining int64
+
+			// Handle both int64 and float64 (when loaded from JSON)
+			if remainingInt64, ok := remaining.(int64); ok {
+				currentRemaining = remainingInt64
+			} else if remainingFloat64, ok := remaining.(float64); ok {
+				currentRemaining = int64(remainingFloat64)
+			} else {
+				// If neither type matches, skip decrement
+				s.LogState()
+				return
+			}
+
+			newRemaining := currentRemaining - count
+			if newRemaining < 0 {
+				newRemaining = 0
+			}
+			s.Streams[index].State.Store(RemainingRecordsCount, newRemaining)
+			s.Streams[index].HoldsValue.Store(true)
+		}
+	}
+	s.LogState()
+}
+
 func (s *State) MarshalJSON() ([]byte, error) {
 	type Alias State
 	p := Alias(*s)