fix(rollups): rollup a batch if more than 2 seconds elapsed since last batch

posting/mvcc.go

@@ -83,16 +83,22 @@ func (ir *incrRollupi) rollUpKey(writer *TxnWriter, key []byte) error {
 	return writer.Write(&bpb.KVList{Kv: kvs})
 }
 
+var lastSentTs int64 // keep track of ts of last batch rolled up.
 func (ir *incrRollupi) addKeyToBatch(key []byte) {
 	batch := ir.keysPool.Get().(*[][]byte)
 	*batch = append(*batch, key)
-	if len(*batch) < 64 {
+	sent := atomic.LoadInt64(&lastSentTs)
+	now := time.Now().Unix()
+
+	// Rollup only if batch len is 64 or it has been more than 2 seconds since last batch.
+	if now-sent < 2 && len(*batch) < 64 {
 		ir.keysPool.Put(batch)
 		return
 	}
 
 	select {
 	case ir.keysCh <- batch:
+		atomic.StoreInt64(&lastSentTs, now)
 	default:
 		// Drop keys and build the batch again. Lossy behavior.
 		*batch = (*batch)[:0]

posting/mvcc_test.go

@@ -25,6 +25,13 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func BenchmarkIncrRollupAtomic(b *testing.B) {
+	buf := []byte("key")
+	for n := 0; n < b.N; n++ {
+		IncrRollup.addKeyToBatch(buf)
+	}
+}
+
 func TestRollupTimestamp(t *testing.T) {
 	key := x.DataKey("rollup", 1)
 	// 3 Delta commits.