chore: Dedup log objects during build (#19378)

Author: benclive

pkg/dataobj/sections/logs/table_build.go

@@ -1,6 +1,7 @@
 package logs
 
 import (
+	"bytes"
 	"cmp"
 	"slices"
 
@@ -22,21 +23,30 @@ func buildTable(buf *tableBuffer, pageSize, pageRowCount int, compressionOpts da
 		messageBuilder   = buf.Message(pageSize, pageRowCount, compressionOpts)
 	)
 
-	for i, record := range records {
+	var prev Record
+	row := 0
+	for _, record := range records {
+		if equalRecords(prev, record) {
+			// Skip equal records
+			continue
+		}
+		prev = record
+
 		// Append only fails if given out-of-order data, where the provided row
 		// number is less than the previous row number. That can't happen here, so
 		// to keep the code readable we ignore the error values.
 
-		_ = streamIDBuilder.Append(i, dataset.Int64Value(record.StreamID))
-		_ = timestampBuilder.Append(i, dataset.Int64Value(record.Timestamp.UnixNano()))
-		_ = messageBuilder.Append(i, dataset.BinaryValue(record.Line))
+		_ = streamIDBuilder.Append(row, dataset.Int64Value(record.StreamID))
+		_ = timestampBuilder.Append(row, dataset.Int64Value(record.Timestamp.UnixNano()))
+		_ = messageBuilder.Append(row, dataset.BinaryValue(record.Line))
 
 		record.Metadata.Range(func(md labels.Label) {
 			// Passing around md.Value as an unsafe slice is safe here: appending
 			// values is always read-only and the byte slice will never be mutated.
 			metadataBuilder := buf.Metadata(md.Name, pageSize, pageRowCount, compressionOpts)
-			_ = metadataBuilder.Append(i, dataset.BinaryValue(unsafeSlice(md.Value, 0)))
+			_ = metadataBuilder.Append(row, dataset.BinaryValue(unsafeSlice(md.Value, 0)))
 		})
+		row++
 	}
 
 	table, err := buf.Flush()
@@ -56,3 +66,16 @@ func sortRecords(records []Record) {
 		return cmp.Compare(a.StreamID, b.StreamID)
 	})
 }
+
+func equalRecords(a, b Record) bool {
+	if a.StreamID != b.StreamID {
+		return false
+	}
+	if a.Timestamp != b.Timestamp {
+		return false
+	}
+	if !labels.Equal(a.Metadata, b.Metadata) {
+		return false
+	}
+	return bytes.Equal(a.Line, b.Line)
+}

pkg/dataobj/sections/logs/table_merge.go

@@ -1,6 +1,7 @@
 package logs
 
 import (
+	"bytes"
 	"cmp"
 	"context"
 	"errors"
@@ -9,6 +10,7 @@ import (
 	"math"
 
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/dataset"
+	"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/datasetmd"
 	"github.com/grafana/loki/v3/pkg/dataobj/internal/result"
 	"github.com/grafana/loki/v3/pkg/util/loser"
 )
@@ -60,9 +62,7 @@ func mergeTables(buf *tableBuffer, pageSize, pageRowCount int, compressionOpts d
 		messageBuilder   = buf.Message(pageSize, pageRowCount, compressionOpts)
 	)
 
-	var (
-		tableSequences = make([]*tableSequence, 0, len(tables))
-	)
+	tableSequences := make([]*tableSequence, 0, len(tables))
 	for _, t := range tables {
 		dsetColumns, err := result.Collect(t.ListColumns(context.Background()))
 		if err != nil {
@@ -96,6 +96,7 @@ func mergeTables(buf *tableBuffer, pageSize, pageRowCount int, compressionOpts d
 	tree := loser.New(tableSequences, maxValue, tableSequenceAt, CompareForSortOrder(sort), tableSequenceClose)
 	defer tree.Close()
 
+	var prev dataset.Row
 	for tree.Next() {
 		seq := tree.Winner()
 
@@ -104,6 +105,12 @@ func mergeTables(buf *tableBuffer, pageSize, pageRowCount int, compressionOpts d
 			return nil, err
 		}
 
+		if equalRows(prev, row) {
+			// Skip equal rows
+			continue
+		}
+		prev = row
+
 		for i, column := range seq.columns {
 			// column is guaranteed to be a *tableColumn since we got it from *table.
 			column := column.(*tableColumn)
@@ -258,3 +265,41 @@ func CompareRows(a, b dataset.Row) int {
 	}
 	return cmp.Compare(bTimestamp, aTimestamp)
 }
+
+// equalRows compares two rows for equality, column by column.
+// a row is considered equal if all the columns are equal.
+func equalRows(a, b dataset.Row) bool {
+	if len(a.Values) != len(b.Values) {
+		return false
+	}
+
+	// The first two columns of each row are *always* stream ID and timestamp, so they will be checked first.
+	// This means equalRows will exit quickly for rows with different timestamps without reading the rest of the columns.
+	for i := 0; i < len(a.Values); i++ {
+		aType, bType := a.Values[i].Type(), b.Values[i].Type()
+		if aType != bType {
+			return false
+		}
+
+		switch aType {
+		case datasetmd.PHYSICAL_TYPE_INT64:
+			if a.Values[i].Int64() != b.Values[i].Int64() {
+				return false
+			}
+		case datasetmd.PHYSICAL_TYPE_UINT64:
+			if a.Values[i].Uint64() != b.Values[i].Uint64() {
+				return false
+			}
+		case datasetmd.PHYSICAL_TYPE_BINARY:
+			if !bytes.Equal(a.Values[i].Binary(), b.Values[i].Binary()) {
+				return false
+			}
+		case datasetmd.PHYSICAL_TYPE_UNSPECIFIED:
+			continue
+		default:
+			return false
+		}
+	}
+
+	return true
+}

pkg/dataobj/sections/logs/table_test.go

@@ -49,7 +49,8 @@ func initBuffer(buf *tableBuffer) {
 func Test_mergeTables(t *testing.T) {
 	var buf tableBuffer
 
-	// tables need to be sorted by Timestamp DESC and StreamID ASC
+	// tables need to be sorted by Timestamp DESC and StreamID ASC.
+	// duplicates are added to ensure the resulting objects are deduplicated if all attributes match.
 	var (
 		tableA = buildTable(&buf, pageSize, pageRows, dataset.CompressionOptions{}, []Record{
 			{StreamID: 3, Timestamp: time.Unix(3, 0), Line: []byte("hello")},
@@ -58,16 +59,21 @@ func Test_mergeTables(t *testing.T) {
 		})
 
 		tableB = buildTable(&buf, pageSize, pageRows, dataset.CompressionOptions{}, []Record{
+			{StreamID: 3, Timestamp: time.Unix(3, 0), Line: []byte("hello")}, // Duplicate in tableA
 			{StreamID: 1, Timestamp: time.Unix(2, 0), Line: []byte("world")},
 			{StreamID: 3, Timestamp: time.Unix(1, 0), Line: []byte("goodbye")},
 		})
 
 		tableC = buildTable(&buf, pageSize, pageRows, dataset.CompressionOptions{}, []Record{
 			{StreamID: 3, Timestamp: time.Unix(2, 0), Line: []byte("are")},
+			{StreamID: 3, Timestamp: time.Unix(2, 0), Line: []byte("are")}, // Duplicate within tableC
 			{StreamID: 2, Timestamp: time.Unix(1, 0), Line: []byte("doing?")},
 		})
 	)
 
+	// TableC should have been initially deduped by buildTable
+	require.Equal(t, tableC.Timestamp.Desc.RowsCount, 2)
+
 	mergedTable, err := mergeTables(&buf, pageSize, pageRows, dataset.CompressionOptions{}, []*table{tableA, tableB, tableC}, SortTimestampDESC)
 	require.NoError(t, err)