chore: Use first-wins semantics for duplicate fields in logfmt/json parsers (#19568)

Author: benclive

pkg/engine/internal/executor/parse_json.go

@@ -76,7 +76,6 @@ func (j *jsonParser) process(line []byte, requestedKeys []string) (map[string]st
 	err := jsonparser.ObjectEach(line, func(key []byte, value []byte, dataType jsonparser.ValueType, _ int) error {
 		return j.parseObject(key, value, dataType, result, requestedKeyLookup)
 	})
-
 	// If there's an error, return empty result for consistency with malformed JSON handling
 	if err != nil {
 		return make(map[string]string), err
@@ -142,6 +141,11 @@ func (j *jsonParser) parseLabelValue(key, value []byte, dataType jsonparser.Valu
 	// Convert the value to string based on its type
 	parsedValue := parseValue(value, dataType)
 	if parsedValue != "" {
+		// First-wins semantics for duplicates
+		_, exists := result[keyString]
+		if exists {
+			return nil
+		}
 		result[keyString] = parsedValue
 	}
 

pkg/engine/internal/executor/parse_logfmt.go

@@ -13,7 +13,7 @@ func buildLogfmtColumns(input *array.String, requestedKeys []string) ([]string,
 }
 
 // tokenizeLogfmt parses logfmt input using the standard decoder
-// Returns a map of key-value pairs with last-wins semantics for duplicates
+// Returns a map of key-value pairs with first-wins semantics for duplicates
 // If requestedKeys is provided, the result will be filtered to only include those keys
 func tokenizeLogfmt(input string, requestedKeys []string) (map[string]string, error) {
 	result := make(map[string]string)
@@ -37,11 +37,15 @@ func tokenizeLogfmt(input string, requestedKeys []string) (map[string]string, er
 
 		val := decoder.Value()
 		if len(val) == 0 {
-			//TODO: retain empty values if --keep-empty is set
+			// TODO: retain empty values if --keep-empty is set
 			continue
 		}
 
-		// Last-wins semantics for duplicates
+		// First-wins semantics for duplicates
+		_, exists := result[key]
+		if exists {
+			continue
+		}
 		result[key] = unsafeString(decoder.Value())
 	}
 

pkg/engine/internal/executor/parse_test.go

@@ -237,6 +237,24 @@ func TestNewParsePipeline_logfmt(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "handle duplicate keys with first-wins semantics",
+			schema: arrow.NewSchema([]arrow.Field{
+				semconv.FieldFromFQN("utf8.builtin.message", true),
+			}, nil),
+			input: arrowtest.Rows{
+				{colMsg: "level=info status=200 level=debug"},
+			},
+			requestedKeys:  nil,
+			expectedFields: 3, // 3 columns: message, level, status
+			expectedOutput: arrowtest.Rows{
+				{
+					colMsg:               "level=info status=200 level=debug",
+					"utf8.parsed.level":  "info",
+					"utf8.parsed.status": "200",
+				},
+			},
+		},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			// Create input data with message column containing logfmt
@@ -625,6 +643,23 @@ func TestNewParsePipeline_JSON(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "duplicate field name takes first value",
+			schema: arrow.NewSchema([]arrow.Field{
+				semconv.FieldFromFQN("utf8.builtin.message", true),
+			}, nil),
+			input: arrowtest.Rows{
+				{colMsg: `{"app": "foo", "app": "duplicate"}`},
+			},
+			requestedKeys:  nil, // Extract all keys
+			expectedFields: 2,   // message, app
+			expectedOutput: arrowtest.Rows{
+				{
+					colMsg:            `{"app": "foo", "app": "duplicate"}`,
+					"utf8.parsed.app": "foo",
+				},
+			},
+		},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			// Create input data with message column containing JSON

pkg/logql/log/parser_test.go

@@ -57,7 +57,8 @@ func Test_jsonParser_Parse(t *testing.T) {
 			},
 			NoParserHints(),
 		},
-		{"numeric",
+		{
+			"numeric",
 			[]byte(`{"counter":1, "price": {"_net_":5.56909}}`),
 			labels.EmptyLabels(),
 			labels.FromStrings("counter", "1",
@@ -665,6 +666,18 @@ func TestJSONExpressionParser(t *testing.T) {
 			),
 			NoParserHints(),
 		},
+		{
+			"duplicate field name takes first value",
+			[]byte(`{"app":"foo","app":"duplicate"}`),
+			[]LabelExtractionExpr{
+				NewLabelExtractionExpr("app", `app`),
+			},
+			labels.FromStrings("foo", "bar"),
+			labels.FromStrings("foo", "bar",
+				"app", "foo",
+			),
+			NoParserHints(),
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -1428,7 +1441,6 @@ func TestLogfmtConsistentPrecedence(t *testing.T) {
 		require.Equal(t, ParsedLabel, cat)
 		require.True(t, ok)
 	})
-
 }
 
 func TestLogfmtExpressionParser(t *testing.T) {