Merge branch 'develop' into yaro/clickhouse-cursor

Author: YaroShkvorets

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Unreleased
 
 * Fixed cursor loading on Clickhouse
+* Improved batch block flush logic to flush after a certain number of blocks instead of taking by modulo
 
 ## v4.5.0
 

sinker/sinker.go

@@ -118,11 +118,12 @@ func (s *SQLSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstream
 		}
 	}
 
-	if (s.batchBlockModulo(isLive) > 0 && data.Clock.Number%s.batchBlockModulo(isLive) == 0) || s.loader.FlushNeeded() {
+	blockFlushNeeded := s.batchBlockModulo(isLive) > 0 && data.Clock.Number-s.stats.GetLastBlockNum() >= s.batchBlockModulo(isLive)
+	if blockFlushNeeded || s.loader.FlushNeeded() {
 		s.logger.Debug("flushing to database",
 			zap.Stringer("block", cursor.Block()),
 			zap.Bool("is_live", *isLive),
-			zap.Bool("block_flush_interval_reached", s.batchBlockModulo(isLive) > 0 && data.Clock.Number%s.batchBlockModulo(isLive) == 0),
+			zap.Bool("block_flush_interval_reached", blockFlushNeeded),
 			zap.Bool("row_flush_interval_reached", s.loader.FlushNeeded()),
 		)
 

sinker/stats.go

@@ -15,7 +15,8 @@ type Stats struct {
 
 	dbFlushRate        *dmetrics.AvgRatePromCounter
 	dbFlushAvgDuration *dmetrics.AvgDurationCounter
-	flusehdRows        *dmetrics.ValueFromMetric
+	flushedRows        *dmetrics.ValueFromMetric
+	dbFlushedRowsRate  *dmetrics.AvgRatePromCounter
 	lastBlock          bstream.BlockRef
 	logger             *zap.Logger
 }
@@ -26,7 +27,8 @@ func NewStats(logger *zap.Logger) *Stats {
 
 		dbFlushRate:        dmetrics.MustNewAvgRateFromPromCounter(FlushCount, 1*time.Second, 30*time.Second, "flush"),
 		dbFlushAvgDuration: dmetrics.NewAvgDurationCounter(30*time.Second, dmetrics.InferUnit, "per flush"),
-		flusehdRows:        dmetrics.NewValueFromMetric(FlushedRowsCount, "rows"),
+		flushedRows:        dmetrics.NewValueFromMetric(FlushedRowsCount, "rows"),
+		dbFlushedRowsRate:  dmetrics.MustNewAvgRateFromPromCounter(FlushedRowsCount, 1*time.Second, 30*time.Second, "flushed rows"),
 		logger:             logger,
 
 		lastBlock: unsetBlockRef{},
@@ -41,6 +43,10 @@ func (s *Stats) RecordFlushDuration(duration time.Duration) {
 	s.dbFlushAvgDuration.AddDuration(duration)
 }
 
+func (s *Stats) GetLastBlockNum() uint64 {
+	return s.lastBlock.Num()
+}
+
 func (s *Stats) Start(each time.Duration, cursor *sink.Cursor) {
 	if !cursor.IsBlank() {
 		s.lastBlock = cursor.Block()
@@ -71,7 +77,8 @@ func (s *Stats) LogNow() {
 	s.logger.Info("postgres sink stats",
 		zap.Stringer("db_flush_rate", s.dbFlushRate),
 		zap.Stringer("db_flush_duration_rate", s.dbFlushAvgDuration),
-		zap.Uint64("flushed_rows", s.flusehdRows.ValueUint()),
+		zap.Uint64("flushed_rows", s.flushedRows.ValueUint()),
+		zap.Stringer("db_flushed_rows_rate", s.dbFlushedRowsRate),
 		zap.Stringer("last_block", s.lastBlock),
 	)
 }