Now using zapx.HumanDuration for all our Duration logging needs

maoueh · maoueh · commit 3d8227bd51a0 · 2025-11-27T09:56:16.000-05:00
This will print the string duration a bit more nicely with the unit suffix to better show the duration to the user.
diff --git a/cmd/substreams-sink-sql/inject_csv.go b/cmd/substreams-sink-sql/inject_csv.go
@@ -19,6 +19,7 @@ import (
 	. "github.com/streamingfast/cli"
 	"github.com/streamingfast/cli/sflags"
 	"github.com/streamingfast/dstore"
+	"github.com/streamingfast/logging/zapx"
 	db2 "github.com/streamingfast/substreams-sink-sql/db_changes/db"
 	"go.uber.org/zap"
 )
@@ -79,7 +80,7 @@ func injectCSVE(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("table filler %q: %w", tableName, err)
 	}
 
-	zlog.Info("table done", zap.Duration("total", time.Since(t0)))
+	zlog.Info("table done", zapx.HumanDuration("total", time.Since(t0)))
 	return nil
 }
 
@@ -232,7 +233,7 @@ func (t *TableFiller) injectCSVFromReader(ctx context.Context, fl io.Reader, sou
 		zap.String("source", source),
 		zap.String("table_name", t.tblName),
 		zap.Int64("rows_affected", count),
-		zap.Duration("elapsed", elapsed),
+		zapx.HumanDuration("elapsed", elapsed),
 	)
 
 	return nil
diff --git a/cmd/substreams-sink-sql/main.go b/cmd/substreams-sink-sql/main.go
@@ -12,6 +12,7 @@ import (
 	. "github.com/streamingfast/cli"
 	"github.com/streamingfast/cli/sflags"
 	"github.com/streamingfast/dmetrics"
+	"github.com/streamingfast/logging/zapx"
 	"go.uber.org/zap"
 )
 
@@ -50,7 +51,7 @@ func preStart(cmd *cobra.Command, _ []string) {
 
 	delay := sflags.MustGetDuration(cmd, "delay-before-start")
 	if delay > 0 {
-		zlog.Info("sleeping to respect delay before start setting", zap.Duration("delay", delay))
+		zlog.Info("sleeping to respect delay before start setting", zapx.HumanDuration("delay", delay))
 		time.Sleep(delay)
 	}
 
diff --git a/db_changes/bundler/stats.go b/db_changes/bundler/stats.go
@@ -5,6 +5,7 @@ import (
 
 	"github.com/streamingfast/bstream"
 	"github.com/streamingfast/dmetrics"
+	"github.com/streamingfast/logging/zapx"
 	"go.uber.org/zap"
 )
 
@@ -61,14 +62,14 @@ func (s *boundaryStats) Log() []zap.Field {
 	return []zap.Field{
 		zap.Uint64("file_count", s.totalBoundaryCount),
 		zap.Stringer("boundary", s.boundary),
-		zap.Duration("boundary_process_duration", s.boundaryProcessTime),
-		zap.Duration("upload_duration", s.uploadedDuration),
-		zap.Duration("data_process_duration", s.procesingDataTime),
-		zap.Duration("avg_upload_duration", s.avgUploadDuration.Average()),
-		zap.Duration("total_upload_duration", s.avgUploadDuration.Total()),
-		zap.Duration("avg_boundary_process_duration", s.avgBoundaryProcessDuration.Average()),
-		zap.Duration("total_boundary_process_duration", s.avgBoundaryProcessDuration.Total()),
-		zap.Duration("avg_data_process_duration", s.avgDataProcessDuration.Average()),
-		zap.Duration("total_data_process_duration", s.avgDataProcessDuration.Total()),
+		zapx.HumanDuration("boundary_process_duration", s.boundaryProcessTime),
+		zapx.HumanDuration("upload_duration", s.uploadedDuration),
+		zapx.HumanDuration("data_process_duration", s.procesingDataTime),
+		zapx.HumanDuration("avg_upload_duration", s.avgUploadDuration.Average()),
+		zapx.HumanDuration("total_upload_duration", s.avgUploadDuration.Total()),
+		zapx.HumanDuration("avg_boundary_process_duration", s.avgBoundaryProcessDuration.Average()),
+		zapx.HumanDuration("total_boundary_process_duration", s.avgBoundaryProcessDuration.Total()),
+		zapx.HumanDuration("avg_data_process_duration", s.avgDataProcessDuration.Average()),
+		zapx.HumanDuration("total_data_process_duration", s.avgDataProcessDuration.Total()),
 	}
 }
diff --git a/db_changes/db/flush.go b/db_changes/db/flush.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/streamingfast/logging/zapx"
 	sink "github.com/streamingfast/substreams-sink"
 	"go.uber.org/zap"
 )
@@ -42,7 +43,7 @@ func (l *Loader) Flush(ctx context.Context, outputModuleHash string, cursor *sin
 	l.reset()
 
 	// We add + 1 to the table count because the `cursors` table is an implicit table
-	l.logger.Debug("flushed table(s) rows to database", zap.Int("table_count", l.entries.Len()+1), zap.Int("row_count", rowFlushedCount), zap.Duration("took", time.Since(startAt)))
+	l.logger.Debug("flushed table(s) rows to database", zap.Int("table_count", l.entries.Len()+1), zap.Int("row_count", rowFlushedCount), zapx.HumanDuration("took", time.Since(startAt)))
 	return rowFlushedCount, nil
 }
 
diff --git a/db_changes/sinker/generate_csv_sinker.go b/db_changes/sinker/generate_csv_sinker.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/streamingfast/dstore"
 	"github.com/streamingfast/logging"
+	"github.com/streamingfast/logging/zapx"
 	"github.com/streamingfast/shutter"
 	sink "github.com/streamingfast/substreams-sink"
 	pbdatabase "github.com/streamingfast/substreams-sink-database-changes/pb/sf/substreams/sink/database/v1"
@@ -154,7 +155,7 @@ func (s *GenerateCSVSinker) Run(ctx context.Context) {
 	s.stats.Start(logEach, cursor)
 
 	s.logger.Info("starting sql generate CSV sink",
-		zap.Duration("stats_refresh_each", logEach),
+		zapx.HumanDuration("stats_refresh_each", logEach),
 		zap.Stringer("restarting_at", cursor.Block()),
 		zap.String("loader", s.loader.GetIdentifier()),
 	)
diff --git a/db_changes/sinker/sinker.go b/db_changes/sinker/sinker.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/streamingfast/logging"
+	"github.com/streamingfast/logging/zapx"
 	"github.com/streamingfast/shutter"
 	sink "github.com/streamingfast/substreams-sink"
 	pbdatabase "github.com/streamingfast/substreams-sink-database-changes/pb/sf/substreams/sink/database/v1"
@@ -109,7 +110,7 @@ func (s *SQLSinker) Run(ctx context.Context) {
 	s.stats.Start(logEach, cursor)
 
 	s.logger.Info("starting sql sink",
-		zap.Duration("stats_refresh_each", logEach),
+		zapx.HumanDuration("stats_refresh_each", logEach),
 		zap.Stringer("restarting_at", cursor.Block()),
 		zap.String("loader", s.loader.GetIdentifier()),
 	)
@@ -129,7 +130,7 @@ func (s *SQLSinker) flushWithRetry(ctx context.Context, moduleHash string, curso
 			s.logger.Warn("retrying flush after error",
 				zap.Int("attempt", attempt),
 				zap.Int("max_retries", retries),
-				zap.Duration("delay", delay),
+				zapx.HumanDuration("delay", delay),
 				zap.Error(lastErr))
 
 			select {
@@ -194,7 +195,7 @@ func (s *SQLSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstream
 		zap.Bool("block_flush_needed_before_timing_check", blockFlushNeeded))
 
 	if blockFlushNeeded && isLive != nil && *isLive && s.stats.AverageFlushDuration() > data.Clock.Timestamp.AsTime().Sub(s.lastAppliedBlockTime) {
-		s.logger.Debug("skipping a flush because we are LIVE and flush average duration is above time between blocks", zap.Duration("flush_duration_average", s.stats.AverageFlushDuration()), zap.Time("last_block_time", s.lastAppliedBlockTime), zap.Time("block_time", data.Clock.Timestamp.AsTime()))
+		s.logger.Debug("skipping a flush because we are LIVE and flush average duration is above time between blocks", zapx.HumanDuration("flush_duration_average", s.stats.AverageFlushDuration()), zap.Time("last_block_time", s.lastAppliedBlockTime), zap.Time("block_time", data.Clock.Timestamp.AsTime()))
 		blockFlushNeeded = false
 	}
 
@@ -225,7 +226,7 @@ func (s *SQLSinker) HandleBlockScopedData(ctx context.Context, data *pbsubstream
 				level = zap.WarnLevel
 			}
 
-			s.logger.Check(level, "flush to database took a long time to complete, could cause long sync time along the road").Write(zap.Duration("took", flushDuration))
+			s.logger.Check(level, "flush to database took a long time to complete, could cause long sync time along the road").Write(zapx.HumanDuration("took", flushDuration))
 		}
 
 		FlushCount.Inc()
diff --git a/db_proto/sinker.go b/db_proto/sinker.go
@@ -6,6 +6,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/streamingfast/logging/zapx"
 	sink "github.com/streamingfast/substreams-sink"
 	sql "github.com/streamingfast/substreams-sink-sql/db_proto/sql"
 	"github.com/streamingfast/substreams-sink-sql/db_proto/stats"
@@ -120,7 +121,7 @@ func (s *Sinker) HandleBlockScopedData(ctx context.Context, data *pbsubstreamsrp
 	holding = append(holding, holder)
 	if data.Clock.Number > (s.lastAppliedBlockNum+s.blockBatchSize) || s.blockBatchSize == 1 || (isLive != nil && *isLive) {
 		if isLive != nil && *isLive && s.stats.FlushDuration.Average() > data.Clock.Timestamp.AsTime().Sub(s.lastAppliedBlockTime) {
-			s.logger.Debug("skipping a flush because we are LIVE and flush average duration is above time between blocks", zap.Duration("flush_duration_average", s.stats.FlushDuration.Average()), zap.Time("last_block_time", s.lastAppliedBlockTime), zap.Time("block_time", data.Clock.Timestamp.AsTime()))
+			s.logger.Debug("skipping a flush because we are LIVE and flush average duration is above time between blocks", zapx.HumanDuration("flush_duration_average", s.stats.FlushDuration.Average()), zap.Time("last_block_time", s.lastAppliedBlockTime), zap.Time("block_time", data.Clock.Timestamp.AsTime()))
 			return nil
 		}
 
diff --git a/db_proto/sql/click_house/accumulator_inserter.go b/db_proto/sql/click_house/accumulator_inserter.go
@@ -608,7 +608,7 @@ func (i *AccumulatorInserter) flush(database *Database) error {
 	}
 	i.accumulators = accs
 
-	i.logger.Debug("flushing done", zap.Duration("duration", time.Since(start)), zap.Int("rows", rowCount))
+	i.logger.Debug("flushing done", zapx.HumanDuration("duration", time.Since(start)), zap.Int("rows", rowCount))
 
 	return nil
 }
diff --git a/db_proto/sql/click_house/database.go b/db_proto/sql/click_house/database.go
@@ -12,6 +12,7 @@ import (
 
 	"github.com/ClickHouse/ch-go"
 	"github.com/streamingfast/logging"
+	"github.com/streamingfast/logging/zapx"
 	sink "github.com/streamingfast/substreams-sink"
 	"github.com/streamingfast/substreams-sink-sql/bytes"
 	"github.com/streamingfast/substreams-sink-sql/db_changes/db"
@@ -415,7 +416,7 @@ func (d *Database) HandleBlocksUndo(lastValidBlockNum uint64) error {
 	//	return fmt.Errorf("optimizing table: %w", err)
 	//}
 
-	d.logger.Info("undo completed", zap.String("table", "_block_"), zap.Duration("duration", time.Since(start)))
+	d.logger.Info("undo completed", zap.String("table", "_block_"), zapx.HumanDuration("duration", time.Since(start)))
 
 	for _, table := range tables {
 		d.logger.Info("undoing blocks", zap.String("table", table.Name), zap.Uint64("last_valid_block_num", lastValidBlockNum))
@@ -467,7 +468,7 @@ func (d *Database) HandleBlocksUndo(lastValidBlockNum uint64) error {
 		//	Body: fmt.Sprintf("OPTIMIZE TABLE %s FINAL CLEANUP;", tableFullName),
 		//})
 
-		d.logger.Info("undo completed", zap.String("table", table.Name), zap.Duration("duration", time.Since(start)))
+		d.logger.Info("undo completed", zap.String("table", table.Name), zapx.HumanDuration("duration", time.Since(start)))
 	}
 	err = d.CommitTransaction()
 	if err != nil {
diff --git a/db_proto/sql/postgres/database.go b/db_proto/sql/postgres/database.go
@@ -7,6 +7,7 @@ import (
 	"hash/fnv"
 	"time"
 
+	"github.com/streamingfast/logging/zapx"
 	sink "github.com/streamingfast/substreams-sink"
 	"github.com/streamingfast/substreams-sink-sql/bytes"
 	"github.com/streamingfast/substreams-sink-sql/db_changes/db"
@@ -148,7 +149,7 @@ func (d *Database) applyConstraints() error {
 			return fmt.Errorf("executing fk constraint statement: %w %s", err, constraint.Sql)
 		}
 	}
-	d.logger.Info("applying constraints", zap.Duration("duration", time.Since(startAt)))
+	d.logger.Info("applying constraints", zapx.HumanDuration("duration", time.Since(startAt)))
 	return nil
 }
 
diff --git a/db_proto/stats/stats.go b/db_proto/stats/stats.go
@@ -3,6 +3,7 @@ package stats
 import (
 	"time"
 
+	"github.com/streamingfast/logging/zapx"
 	"go.uber.org/zap"
 )
 
@@ -31,11 +32,11 @@ func (a *Average) Average() time.Duration {
 	if len(a.Duration) == 0 {
 		return 0
 	}
-	var total int64
+	var total time.Duration
 	for _, d := range a.Duration {
-		total += d.Nanoseconds()
+		total += d
 	}
-	return time.Duration(total / int64(len(a.Duration)))
+	return time.Duration(total / time.Duration(len(a.Duration)))
 }
 
 func (a *Average) LastItemsAverage(count int) time.Duration {
@@ -53,7 +54,10 @@ func (a *Average) LastItemsAverage(count int) time.Duration {
 }
 
 func (a *Average) Log(logger *zap.Logger) {
-	logger.Info(a.title, zap.Duration("average", a.Average()), zap.Duration("last X average", a.LastItemsAverage(a.lastX)))
+	logger.Info(a.title,
+		zapx.HumanDuration("average", a.Average()),
+		zapx.HumanDuration("last X average", a.LastItemsAverage(a.lastX)),
+	)
 }
 
 type Stats struct {
@@ -97,7 +101,14 @@ func (s *Stats) Log() {
 	if s.BlockCount == 0 {
 		s.logger.Info("Stats: no blocks processed yet")
 	} else {
-		s.logger.Info("Stats", zap.Int("block_count", s.BlockCount), zap.Duration("Processing Time", s.TotalProcessingDuration), zap.Duration("Total Wait Duration", s.TotalDurationBetween), zap.Duration("Total Duration", s.TotalDurationBetween+s.TotalProcessingDuration), zap.Time("Last Block Process At", s.LastBlockProcessAt))
+		s.logger.Info("Stats",
+			zap.Int("block_count", s.BlockCount),
+			zapx.HumanDuration("Processing Time", s.TotalProcessingDuration),
+			zapx.HumanDuration("Total Wait Duration", s.TotalDurationBetween),
+			zapx.HumanDuration("Total Duration", s.TotalDurationBetween+s.TotalProcessingDuration),
+			zap.Time("Last Block Process At", s.LastBlockProcessAt),
+		)
+
 		s.WaitDurationBetweenBlocks.Log(s.logger)
 		s.BlockProcessingDuration.Log(s.logger)
 		s.UnmarshallingDuration.Log(s.logger)
diff --git a/go.mod b/go.mod
@@ -15,7 +15,7 @@ require (
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.15.0
-	github.com/streamingfast/logging v0.0.0-20250728160343-57342b174ace
+	github.com/streamingfast/logging v0.0.0-20251127143054-23a35e5bd633
 	github.com/streamingfast/substreams v1.16.7-0.20250925152521-9d7a8ef0f261
 	github.com/streamingfast/substreams-sink v0.5.3-0.20250818134825-6b25ffb8232c
 	github.com/streamingfast/substreams-sink-database-changes v1.3.2-0.20250509171446-cb64cdbfea72
@@ -27,7 +27,6 @@ require (
 	go.uber.org/zap v1.27.0
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
 	google.golang.org/protobuf v1.36.6
-
 )
 
 require (
diff --git a/go.sum b/go.sum
@@ -614,8 +614,8 @@ github.com/streamingfast/firehose-networks v0.2.0/go.mod h1:/XyroRcAUOO9DdxdgHkt
 github.com/streamingfast/logging v0.0.0-20210811175431-f3b44b61606a/go.mod h1:4GdqELhZOXj4xwc4IaBmzofzdErGynnaSzuzxy0ZIBo=
 github.com/streamingfast/logging v0.0.0-20220222131651-12c3943aac2e/go.mod h1:4GdqELhZOXj4xwc4IaBmzofzdErGynnaSzuzxy0ZIBo=
 github.com/streamingfast/logging v0.0.0-20220304214715-bc750a74b424/go.mod h1:VlduQ80JcGJSargkRU4Sg9Xo63wZD/l8A5NC/Uo1/uU=
-github.com/streamingfast/logging v0.0.0-20250728160343-57342b174ace h1:qiDK6tlIU//4pqTEuxN+jAkmlFJKlrL5Jm8h4TSTXSU=
-github.com/streamingfast/logging v0.0.0-20250728160343-57342b174ace/go.mod h1:hAKr497lpFM43lZ8Ki4uummLGEAEeVAfcwsDWrG0zbE=
+github.com/streamingfast/logging v0.0.0-20251127143054-23a35e5bd633 h1:e9b6GC7qjlMuc8oAzyjd4AYiFONqQ5Z2UuTP5e1a0Zc=
+github.com/streamingfast/logging v0.0.0-20251127143054-23a35e5bd633/go.mod h1:BtDq81Tyc7H8up5aXNi/I95nPmG3C0PLEqGWY/iWQ2E=
 github.com/streamingfast/opaque v0.0.0-20210811180740-0c01d37ea308 h1:xlWSfi1BoPfsHtPb0VEHGUcAdBF208LUiFCwfaVPfLA=
 github.com/streamingfast/opaque v0.0.0-20210811180740-0c01d37ea308/go.mod h1:K1p8Bj/wG34KJvYzPUqtzpndffmpkrVY11u2hkyxCWQ=
 github.com/streamingfast/pbgo v0.0.6-0.20240823134334-812f6a16c5cb h1:Xqt4ned9ELmQMKcg7cFbm56MKG2gBjnE1M+2HObOs6w=
diff --git a/tests/integration/db_proto_clickhouse_test.go b/tests/integration/db_proto_clickhouse_test.go
@@ -117,10 +117,7 @@ func TestDbProtoClickhouseIntegration(t *testing.T) {
 				UseTransactions: true,
 				BlockBatchSize:  1,
 				Parallel:        false,
-				Clickhouse: struct {
-					SinkInfoFolder string
-					CursorFilePath string
-				}{
+				Clickhouse: db_proto.SinkerFactoryClickhouse{
 					SinkInfoFolder: clickhouseStateFolder,
 					CursorFilePath: filepath.Join(clickhouseStateFolder, "cursor.txt"),
 				},
diff --git a/tests/integration/helpers_test.go b/tests/integration/helpers_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/jmoiron/sqlx"
 	"github.com/streamingfast/bstream"
 	"github.com/streamingfast/logging"
+	"github.com/streamingfast/logging/zapx"
 	"github.com/streamingfast/substreams/client"
 	pbsubstreamsrpc "github.com/streamingfast/substreams/pb/sf/substreams/rpc/v2"
 	pbsubstreams "github.com/streamingfast/substreams/pb/sf/substreams/v1"
@@ -178,7 +179,9 @@ func setupClickhouseContainer(t *testing.T, seedDb ClickhouseSeeder) (dbConnecti
 	t.Helper()
 
 	start := time.Now()
-	defer func() { logger.Debug("setupClickhouseContainer duration", zap.Duration("duration", time.Since(start))) }()
+	defer func() {
+		logger.Debug("setupClickhouseContainer duration", zapx.HumanDuration("duration", time.Since(start)))
+	}()
 
 	ctx := context.Background()
 

Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,7 @@ import (`
`19`	`19`	`. "github.com/streamingfast/cli"`
`20`	`20`	`"github.com/streamingfast/cli/sflags"`
`21`	`21`	`"github.com/streamingfast/dstore"`
	`22`	`+ "github.com/streamingfast/logging/zapx"`
`22`	`23`	`db2 "github.com/streamingfast/substreams-sink-sql/db_changes/db"`
`23`	`24`	`"go.uber.org/zap"`
`24`	`25`	`)`
`@@ -79,7 +80,7 @@ func injectCSVE(cmd *cobra.Command, args []string) error {`
`79`	`80`	`return fmt.Errorf("table filler %q: %w", tableName, err)`
`80`	`81`	`}`
`81`	`82`
`82`		`- zlog.Info("table done", zap.Duration("total", time.Since(t0)))`
	`83`	`+ zlog.Info("table done", zapx.HumanDuration("total", time.Since(t0)))`
`83`	`84`	`return nil`
`84`	`85`	`}`
`85`	`86`
`@@ -232,7 +233,7 @@ func (t *TableFiller) injectCSVFromReader(ctx context.Context, fl io.Reader, sou`
`232`	`233`	`zap.String("source", source),`
`233`	`234`	`zap.String("table_name", t.tblName),`
`234`	`235`	`zap.Int64("rows_affected", count),`
`235`		`- zap.Duration("elapsed", elapsed),`
	`236`	`+ zapx.HumanDuration("elapsed", elapsed),`
`236`	`237`	`)`
`237`	`238`
`238`	`239`	`return nil`
Original file line number	Diff line number	Diff line change
`@@ -12,6 +12,7 @@ import (`
`12`	`12`	`. "github.com/streamingfast/cli"`
`13`	`13`	`"github.com/streamingfast/cli/sflags"`
`14`	`14`	`"github.com/streamingfast/dmetrics"`
	`15`	`+ "github.com/streamingfast/logging/zapx"`
`15`	`16`	`"go.uber.org/zap"`
`16`	`17`	`)`
`17`	`18`
`@@ -50,7 +51,7 @@ func preStart(cmd *cobra.Command, _ []string) {`
`50`	`51`
`51`	`52`	`delay := sflags.MustGetDuration(cmd, "delay-before-start")`
`52`	`53`	`if delay > 0 {`
`53`		`- zlog.Info("sleeping to respect delay before start setting", zap.Duration("delay", delay))`
	`54`	`+ zlog.Info("sleeping to respect delay before start setting", zapx.HumanDuration("delay", delay))`
`54`	`55`	`time.Sleep(delay)`
`55`	`56`	`}`
`56`	`57`
Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,7 @@ import (`
`6`	`6`	`"time"`
`7`	`7`
`8`	`8`	`"github.com/ClickHouse/clickhouse-go/v2"`
	`9`	`+ "github.com/streamingfast/logging/zapx"`
`9`	`10`	`sink "github.com/streamingfast/substreams-sink"`
`10`	`11`	`"go.uber.org/zap"`
`11`	`12`	`)`
`@@ -42,7 +43,7 @@ func (l Loader) Flush(ctx context.Context, outputModuleHash string, cursor sin`
`42`	`43`	`l.reset()`
`43`	`44`
`44`	`45`	// We add + 1 to the table count because the `cursors` table is an implicit table
`45`		`- l.logger.Debug("flushed table(s) rows to database", zap.Int("table_count", l.entries.Len()+1), zap.Int("row_count", rowFlushedCount), zap.Duration("took", time.Since(startAt)))`
	`46`	`+ l.logger.Debug("flushed table(s) rows to database", zap.Int("table_count", l.entries.Len()+1), zap.Int("row_count", rowFlushedCount), zapx.HumanDuration("took", time.Since(startAt)))`
`46`	`47`	`return rowFlushedCount, nil`
`47`	`48`	`}`
`48`	`49`
Original file line number	Diff line number	Diff line change
`@@ -608,7 +608,7 @@ func (i AccumulatorInserter) flush(database Database) error {`
`608`	`608`	`}`
`609`	`609`	`i.accumulators = accs`
`610`	`610`
`611`		`- i.logger.Debug("flushing done", zap.Duration("duration", time.Since(start)), zap.Int("rows", rowCount))`
	`611`	`+ i.logger.Debug("flushing done", zapx.HumanDuration("duration", time.Since(start)), zap.Int("rows", rowCount))`
`612`	`612`
`613`	`613`	`return nil`
`614`	`614`	`}`
Original file line number	Diff line number	Diff line change
`@@ -7,6 +7,7 @@ import (`
`7`	`7`	`"hash/fnv"`
`8`	`8`	`"time"`
`9`	`9`
	`10`	`+ "github.com/streamingfast/logging/zapx"`
`10`	`11`	`sink "github.com/streamingfast/substreams-sink"`
`11`	`12`	`"github.com/streamingfast/substreams-sink-sql/bytes"`
`12`	`13`	`"github.com/streamingfast/substreams-sink-sql/db_changes/db"`
`@@ -148,7 +149,7 @@ func (d *Database) applyConstraints() error {`
`148`	`149`	`return fmt.Errorf("executing fk constraint statement: %w %s", err, constraint.Sql)`
`149`	`150`	`}`
`150`	`151`	`}`
`151`		`- d.logger.Info("applying constraints", zap.Duration("duration", time.Since(startAt)))`
	`152`	`+ d.logger.Info("applying constraints", zapx.HumanDuration("duration", time.Since(startAt)))`
`152`	`153`	`return nil`
`153`	`154`	`}`
`154`	`155`