executor: Fix the parse problematic slow log panic issue due to empty…

… value string (#58258) (#58272)

close #58147

executor/slow_query.go

@@ -598,6 +598,10 @@ func splitByColon(line string) (fields []string, values []string) {
 			fields = append(fields, line[start:current])
 			parseKey = false
 			current += 2 // bypass ": "
+			if current >= lineLength {
+				// last empty value
+				values = append(values, "")
+			}
 		} else {
 			start = current
 			if current < lineLength && (line[current] == '{' || line[current] == '[') {
@@ -611,6 +615,13 @@ func splitByColon(line string) (fields []string, values []string) {
 				for current < lineLength && line[current] != ' ' {
 					current++
 				}
+				// Meet empty value cases: "Key: Key:"
+				if current > 0 && line[current-1] == ':' {
+					values = append(values, "")
+					current = start
+					parseKey = true
+					continue
+				}
 			}
 			values = append(values, line[start:mathutil.Min(current, len(line))])
 			parseKey = true
@@ -620,6 +631,10 @@ func splitByColon(line string) (fields []string, values []string) {
 		logutil.BgLogger().Warn("slow query parse slow log error", zap.String("Error", errMsg), zap.String("Log", line))
 		return nil, nil
 	}
+	if len(fields) != len(values) {
+		logutil.BgLogger().Warn("slow query parse slow log error", zap.Int("field_count", len(fields)), zap.Int("value_count", len(values)), zap.String("Log", line))
+		return nil, nil
+	}
 	return fields, values
 }
 

executor/slow_query_test.go

@@ -499,7 +499,7 @@ func TestSplitbyColon(t *testing.T) {
 		{
 			"123a",
 			[]string{"123a"},
-			[]string{},
+			[]string{""},
 		},
 		{
 			"1a: 2b",
@@ -552,9 +552,16 @@ func TestSplitbyColon(t *testing.T) {
 			[]string{"Time"},
 			[]string{"2021-09-08T14:39:54.506967433+08:00"},
 		},
+		{
+
+			"Cop_proc_avg: 0 Cop_proc_addr: Cop_proc_max: Cop_proc_min: ",
+			[]string{"Cop_proc_avg", "Cop_proc_addr", "Cop_proc_max", "Cop_proc_min"},
+			[]string{"0", "", "", ""},
+		},
 	}
 	for _, c := range cases {
 		resFields, resValues := splitByColon(c.line)
+		logutil.BgLogger().Info(c.line)
 		require.Equal(t, c.fields, resFields)
 		require.Equal(t, c.values, resValues)
 	}