Merge branch 'develop' into yaro/fix-flush

Author: YaroShkvorets

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Unreleased
 
 * Fixed a bug with not flushing rows on stream completion
+* 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
 

db/cursor.go

@@ -24,7 +24,8 @@ type cursorRow struct {
 // GetAllCursors returns an unordered map given for each module's hash recorded
 // the active cursor for it.
 func (l *Loader) GetAllCursors(ctx context.Context) (out map[string]*sink.Cursor, err error) {
-	rows, err := l.DB.QueryContext(ctx, fmt.Sprintf("SELECT id, cursor, block_num, block_id from %s", l.cursorTable.identifier))
+	query := l.getDialect().GetAllCursorsQuery(l.cursorTable.identifier)
+	rows, err := l.DB.QueryContext(ctx, query)
 	if err != nil {
 		return nil, fmt.Errorf("query all cursors: %w", err)
 	}

db/dialect.go

@@ -22,6 +22,7 @@ type dialect interface {
 	ExecuteSetupScript(ctx context.Context, l *Loader, schemaSql string) error
 	DriverSupportRowsAffected() bool
 	GetUpdateCursorQuery(table, moduleHash string, cursor *sink.Cursor, block_num uint64, block_id string) string
+	GetAllCursorsQuery(table string) string
 	ParseDatetimeNormalization(value string) string
 	Flush(tx Tx, ctx context.Context, l *Loader, outputModuleHash string, lastFinalBlock uint64) (int, error)
 	Revert(tx Tx, ctx context.Context, l *Loader, lastValidFinalBlock uint64) error

db/dialect_clickhouse.go

@@ -120,6 +120,10 @@ func (d clickhouseDialect) GetUpdateCursorQuery(table, moduleHash string, cursor
 	`, table, moduleHash, cursor, block_num, block_id)
 }
 
+func (d clickhouseDialect) GetAllCursorsQuery(table string) string {
+	return fmt.Sprintf("SELECT id, cursor, block_num, block_id FROM %s FINAL", table)
+}
+
 func (d clickhouseDialect) ParseDatetimeNormalization(value string) string {
 	return fmt.Sprintf("parseDateTimeBestEffort(%s)", escapeStringValue(value))
 }

db/dialect_postgres.go

@@ -234,6 +234,10 @@ func (d postgresDialect) GetUpdateCursorQuery(table, moduleHash string, cursor *
 	`, table, cursor, block_num, block_id, moduleHash)
 }
 
+func (d postgresDialect) GetAllCursorsQuery(table string) string {
+	return fmt.Sprintf("SELECT id, cursor, block_num, block_id FROM %s", table)
+}
+
 func (d postgresDialect) ParseDatetimeNormalization(value string) string {
 	return escapeStringValue(value)
 }

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),
 	)
 }