[perf] Gracefully handle trace values duplicates

Should fix errors like `pgx - Exec map[args:[] err:ERROR:
com.google.api.gax.rpc.OutOfRangeException:
io.grpc.StatusRuntimeException: OUT_OF_RANGE: Cannot affect a row second
time for key (<unprintable>, 42) (SQLSTATE P0001) pid:44 sql:INSERT
INTO`. There is no reason to drop the whole batch of data because of the
duplicates.

Change-Id: I412ac5d8445b356c150bf4f170776d0758b7aaef
Reviewed-on: https://skia-review.googlesource.com/c/buildbot/+/1124858
Reviewed-by: Maxim Sheshukov <[email protected]>
Commit-Queue: Anri Sidorov <[email protected]>

Author: nevedaren

perf/go/tracestore/sqltracestore/sqltracestore.go

@@ -1189,7 +1189,12 @@ func (s *SQLTraceStore) WriteTraces(ctx context.Context, commitNumber types.Comm
 			continue
 		}
 		traceID := traceIDForSQLFromTraceName(traceName)
-		traceParams[string(traceID)] = p
+		traceIDKey := string(traceID)
+		if _, exists := traceParams[traceIDKey]; exists {
+			sklog.Warningf("Many values for the same trace name found, skipping all but the first: %s", traceName)
+			continue
+		}
+		traceParams[traceIDKey] = p
 		valuesTemplateContext = append(valuesTemplateContext, insertIntoTraceValuesContext{
 			MD5HexTraceID: traceID,
 			CommitNumber:  commitNumber,

perf/go/tracestore/sqltracestore/sqltracestore_test.go

@@ -657,6 +657,34 @@ func TestWriteTraces_InsertDifferentValueAndFile_OverwriteExistingTraceValues(t
 	assert.Equal(t, float32(2.4), trace[1])
 }
 
+func TestWriteTraces_DuplicateTraceIDsInBatch_Succeeds(t *testing.T) {
+	ctx, s := commonTestSetupWithCommits(t)
+	// We are simulating a scenario where the ingestion process sends two values
+	// for the exactly same trace in a single batch.
+	traceNames := []paramtools.Params{
+		{"config": "8888", "arch": "x86"}, // Entry A
+		{"config": "8888", "arch": "x86"}, // Entry A (Duplicate of above)
+	}
+	values := []float32{1.0, 99.0}
+	ps := paramtools.ParamSet{
+		"config": {"8888"},
+		"arch":   {"x86"},
+	}
+
+	err := s.WriteTraces(ctx, types.CommitNumber(2), traceNames, values, ps, file1, time.Time{})
+	require.NoError(t, err)
+
+	keys := []string{",arch=x86,config=8888,"}
+	ts, _, _, err := s.ReadTraces(ctx, types.TileNumber(0), keys)
+	require.NoError(t, err)
+	trace := ts[",arch=x86,config=8888,"]
+	actualValue := trace[2]
+	// Verify that the stored value is one of the inputs.
+	// We don't care if it's 1.0 (first-write-wins) or 99.0 (last-write-wins),
+	// as long as the write succeeded and data is present.
+	assert.Contains(t, []float32{1.0, 99.0}, actualValue, "The stored value should be one of the input values from the batch")
+}
+
 func TestReadTraces_WithDiscontinueCommitNumbers_Succeed(t *testing.T) {
 	ctx, s := commonTestSetupWithCommits(t)