fix: fix metrics (#3147)

Signed-off-by: Song Gao <[email protected]>

.gitignore

@@ -41,3 +41,4 @@ corss_build_for_rpm.tar
 
 .vscode/
 tools/lint/bin/
+go.work.sum

internal/topo/node/batch_op.go

@@ -117,7 +117,6 @@ func (b *BatchOp) ingest(ctx api.StreamContext, item any, checkSize bool) {
 	b.currIndex++
 	if checkSize && b.currIndex >= b.batchSize {
 		b.send()
-		b.statManager.IncTotalRecordsOut()
 	}
 	b.statManager.ProcessTimeEnd()
 	b.statManager.IncTotalMessagesProcessed(1)
@@ -133,6 +132,7 @@ func (b *BatchOp) send() {
 	})
 
 	b.Broadcast(b.buffer)
+	b.statManager.IncTotalRecordsOut()
 	// Reset buffer
 	b.buffer = &xsql.WindowTuples{
 		Content: make([]xsql.Row, 0, b.batchSize),