Merge branch 'public-single-node' into pmm-6401-read-prometheus-data-files

lib/mergeset/table.go

@@ -644,17 +644,20 @@ func (ris *rawItemsShard) appendBlocksToFlush(dst []*inmemoryBlock, tb *Table, i
 		flushSeconds = 1
 	}
 	lastFlushTime := atomic.LoadUint64(&ris.lastFlushTime)
-	if isFinal || currentTime-lastFlushTime > uint64(flushSeconds) {
+	if !isFinal && currentTime <= lastFlushTime+uint64(flushSeconds) {
-		ris.mu.Lock()
+		// Fast path - nothing to flush
-		ibs := ris.ibs
+		return dst
-		dst = append(dst, ibs...)
-		for i := range ibs {
-			ibs[i] = nil
-		}
-		ris.ibs = ibs[:0]
-		atomic.StoreUint64(&ris.lastFlushTime, currentTime)
-		ris.mu.Unlock()
 	}
+	// Slow path - move ris.ibs to dst
+	ris.mu.Lock()
+	ibs := ris.ibs
+	dst = append(dst, ibs...)
+	for i := range ibs {
+		ibs[i] = nil
+	}
+	ris.ibs = ibs[:0]
+	atomic.StoreUint64(&ris.lastFlushTime, currentTime)
+	ris.mu.Unlock()
 	return dst
 }
 

lib/storage/partition.go

@@ -473,23 +473,17 @@ func (rrs *rawRowsShard) addRows(pt *partition, rows []rawRow) {
 
 	rrs.mu.Lock()
 	if cap(rrs.rows) == 0 {
-		n := getMaxRawRowsPerShard()
+		rrs.rows = newRawRowsBlock()
-		rrs.rows = make([]rawRow, 0, n)
 	}
-	maxRowsCount := cap(rrs.rows)
+	n := copy(rrs.rows[len(rrs.rows):cap(rrs.rows)], rows)
-	capacity := maxRowsCount - len(rrs.rows)
+	rrs.rows = rrs.rows[:len(rrs.rows)+n]
-	if capacity >= len(rows) {
+	rows = rows[n:]
-		// Fast path - rows fit rrs.rows capacity.
+	if len(rows) > 0 {
-		rrs.rows = append(rrs.rows, rows...)
+		// Slow path - rows did't fit rrs.rows capacity.
-	} else {
+		// Convert rrs.rows to rowsToFlush and convert it to a part,
-		// Slow path - rows don't fit rrs.rows capacity.
+		// then try moving the remaining rows to rrs.rows.
-		// Fill rrs.rows with rows until capacity,
-		// then put rrs.rows to rowsToFlush and convert it to a part.
-		n := copy(rrs.rows[:cap(rrs.rows)], rows)
-		rows = rows[n:]
 		rowsToFlush = rrs.rows
-		n = getMaxRawRowsPerShard()
+		rrs.rows = newRawRowsBlock()
-		rrs.rows = make([]rawRow, 0, n)
 		if len(rows) <= n {
 			rrs.rows = append(rrs.rows[:0], rows...)
 		} else {
@@ -504,6 +498,11 @@ func (rrs *rawRowsShard) addRows(pt *partition, rows []rawRow) {
 	pt.flushRowsToParts(rowsToFlush)
 }
 
+func newRawRowsBlock() []rawRow {
+	n := getMaxRawRowsPerShard()
+	return make([]rawRow, 0, n)
+}
+
 func (pt *partition) flushRowsToParts(rows []rawRow) {
 	maxRows := getMaxRawRowsPerShard()
 	wg := getWaitGroup()
@@ -747,13 +746,16 @@ func (rrs *rawRowsShard) appendRawRowsToFlush(dst []rawRow, pt *partition, isFin
 		flushSeconds = 1
 	}
 	lastFlushTime := atomic.LoadUint64(&rrs.lastFlushTime)
-	if isFinal || currentTime-lastFlushTime > uint64(flushSeconds) {
+	if !isFinal && currentTime <= lastFlushTime+uint64(flushSeconds) {
-		rrs.mu.Lock()
+		// Fast path - nothing to flush
-		dst = append(dst, rrs.rows...)
+		return dst
-		rrs.rows = rrs.rows[:0]
-		atomic.StoreUint64(&rrs.lastFlushTime, currentTime)
-		rrs.mu.Unlock()
 	}
+	// Slow path - move rrs.rows to dst.
+	rrs.mu.Lock()
+	dst = append(dst, rrs.rows...)
+	rrs.rows = rrs.rows[:0]
+	atomic.StoreUint64(&rrs.lastFlushTime, currentTime)
+	rrs.mu.Unlock()
 	return dst
 }