Merge pull request #3 from YaroShkvorets/yaro/clickhouse-cursor

fix cursor loading on clickhouse

Author: YaroShkvorets

CHANGELOG.md

@@ -7,6 +7,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

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