diff --git a/go/vehicle-gateway/internal/history/query.go b/go/vehicle-gateway/internal/history/query.go index e84de4ea..c471f964 100644 --- a/go/vehicle-gateway/internal/history/query.go +++ b/go/vehicle-gateway/internal/history/query.go @@ -17,18 +17,20 @@ type Queryer interface { } type RawFrameQuery struct { - Protocol string - VehicleKey string - VIN string - Phone string - DeviceID string - MessageID string - OrderBy string - IncludeTotal bool - DateFrom string - DateTo string - Limit int - Offset int + Protocol string + VehicleKey string + VIN string + Phone string + DeviceID string + MessageID string + OrderBy string + IncludeFields bool + IncludePayload bool + IncludeTotal bool + DateFrom string + DateTo string + Limit int + Offset int } type RawFrameRow struct { @@ -42,7 +44,7 @@ type RawFrameRow struct { RawSizeBytes int64 `json:"raw_size_bytes"` RawHex string `json:"raw_hex,omitempty"` RawText string `json:"raw_text,omitempty"` - ParsedJSON string `json:"parsed_json,omitempty"` + ParsedFields string `json:"parsed_fields,omitempty"` ParseStatus string `json:"parse_status"` ParseError string `json:"parse_error,omitempty"` SourceEndpoint string `json:"source_endpoint"` @@ -113,7 +115,7 @@ func NewLocationRepository(db Queryer, database string) *LocationRepository { func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]RawFrameRow, error) { query = normalizeRawFrameQuery(query) - sqlText, args := buildRawFrameSQL(r.tableName(), query) + sqlText, args := buildRawFrameSQL(r.tableName(query), query) rows, err := r.db.QueryContext(ctx, sqlText, args...) if err != nil { return nil, err @@ -136,7 +138,7 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([] &row.RawSizeBytes, &row.RawHex, &row.RawText, - &row.ParsedJSON, + &row.ParsedFields, &row.ParseStatus, &row.ParseError, &row.SourceEndpoint, @@ -165,7 +167,7 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([] func (r *RawFrameRepository) Count(ctx context.Context, query RawFrameQuery) (int64, error) { query = normalizeRawFrameQuery(query) - sqlText, args := buildRawFrameCountSQL(r.tableName(), query) + sqlText, args := buildRawFrameCountSQL(r.tableName(query), query) return countRows(ctx, r.db, sqlText, args...) } @@ -240,13 +242,37 @@ func countRows(ctx context.Context, db Queryer, sqlText string, args ...any) (in return total, rows.Err() } -func (r *RawFrameRepository) tableName() string { +func (r *RawFrameRepository) tableName(query ...RawFrameQuery) string { + if len(query) > 0 { + if child := r.rawChildTableName(query[0]); child != "" { + return child + } + } if r.database == "" { return "raw_frames" } return r.database + ".raw_frames" } +func (r *RawFrameRepository) rawChildTableName(query RawFrameQuery) string { + protocol := strings.ToUpper(strings.TrimSpace(query.Protocol)) + if protocol == "" { + return "" + } + vehicleKey := strings.TrimSpace(query.VehicleKey) + if vehicleKey == "" { + vehicleKey = strings.TrimSpace(query.VIN) + } + if vehicleKey == "" { + return "" + } + table := "raw_" + strings.ToLower(protocol) + "_" + hash16(vehicleKey) + if r.database != "" { + return r.database + "." + table + } + return table +} + func (r *RawFrameRepository) chunkTableName() string { if r.database == "" { return "raw_frame_payload_chunks" @@ -290,7 +316,17 @@ func normalizeLocationQuery(query LocationQuery) LocationQuery { func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) { where := rawFrameWhere(query) - sqlText := `SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM ` + table + rawHexSelect := "'' AS raw_hex" + rawTextSelect := "'' AS raw_text" + parsedFieldsSelect := "'' AS parsed_fields" + if query.IncludePayload { + rawHexSelect = "raw_hex" + rawTextSelect = "raw_text" + } + if query.IncludeFields { + parsedFieldsSelect = "parsed_json AS parsed_fields" + } + sqlText := `SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, ` + rawHexSelect + `, ` + rawTextSelect + `, ` + parsedFieldsSelect + `, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM ` + table if len(where) > 0 { sqlText += " WHERE " + strings.Join(where, " AND ") } @@ -329,10 +365,10 @@ func (r *RawFrameRepository) hydratePayloadChunks(ctx context.Context, rows []Ra }{ {kind: "raw_hex", value: rows[index].RawHex}, {kind: "raw_text", value: rows[index].RawText}, - {kind: "parsed_json", value: rows[index].ParsedJSON}, + {kind: "parsed_fields", value: rows[index].ParsedFields}, } { manifest, ok := parsePayloadChunkManifest(candidate.value) - if !ok || manifest.PayloadKind != candidate.kind || manifest.ChunkCount <= 0 { + if !ok || !payloadKindMatches(candidate.kind, manifest.PayloadKind) || manifest.ChunkCount <= 0 { continue } eventID := strings.TrimSpace(manifest.EventID) @@ -400,13 +436,20 @@ func (r *RawFrameRepository) hydratePayloadChunks(ctx context.Context, rows []Ra rows[target.rowIndex].RawHex = hydrated case "raw_text": rows[target.rowIndex].RawText = hydrated - case "parsed_json": - rows[target.rowIndex].ParsedJSON = hydrated + case "parsed_fields": + rows[target.rowIndex].ParsedFields = hydrated } } return nil } +func payloadKindMatches(candidate string, manifest string) bool { + if candidate == manifest { + return true + } + return candidate == "parsed_fields" && manifest == "parsed_json" +} + func parsePayloadChunkManifest(value string) (payloadChunkManifest, bool) { var manifest payloadChunkManifest if !strings.Contains(value, `"chunked"`) { @@ -626,18 +669,20 @@ func parseRawFrameQuery(r *http.Request) (RawFrameQuery, error) { return RawFrameQuery{}, err } query := RawFrameQuery{ - Protocol: values.Get("protocol"), - VehicleKey: values.Get("vehicleKey"), - VIN: values.Get("vin"), - Phone: values.Get("phone"), - DeviceID: values.Get("deviceId"), - MessageID: values.Get("messageId"), - OrderBy: values.Get("orderBy"), - IncludeTotal: strings.EqualFold(strings.TrimSpace(values.Get("includeTotal")), "true"), - DateFrom: values.Get("dateFrom"), - DateTo: values.Get("dateTo"), - Limit: limit, - Offset: offset, + Protocol: values.Get("protocol"), + VehicleKey: values.Get("vehicleKey"), + VIN: values.Get("vin"), + Phone: values.Get("phone"), + DeviceID: values.Get("deviceId"), + MessageID: values.Get("messageId"), + OrderBy: values.Get("orderBy"), + IncludeFields: strings.EqualFold(strings.TrimSpace(values.Get("includeFields")), "true"), + IncludePayload: strings.EqualFold(strings.TrimSpace(values.Get("includePayload")), "true"), + IncludeTotal: strings.EqualFold(strings.TrimSpace(values.Get("includeTotal")), "true"), + DateFrom: values.Get("dateFrom"), + DateTo: values.Get("dateTo"), + Limit: limit, + Offset: offset, } if !validDateTime(query.DateFrom) || !validDateTime(query.DateTo) { return RawFrameQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD or YYYY-MM-DD HH:mm:ss") diff --git a/go/vehicle-gateway/internal/history/query_test.go b/go/vehicle-gateway/internal/history/query_test.go index 8afd8776..224c5eaf 100644 --- a/go/vehicle-gateway/internal/history/query_test.go +++ b/go/vehicle-gateway/internal/history/query_test.go @@ -19,10 +19,10 @@ func TestRawFrameRepositoryQueriesRawFramesWithFilters(t *testing.T) { t.Fatalf("sqlmock.New() error = %v", err) } defer db.Close() - mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames"). + mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, .* AS parsed_fields, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_jt808_"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_fields", "parse_status", "parse_error", "source_endpoint", "protocol", "vehicle_key", "vin", "phone", "device_id", }).AddRow( time.Date(2026, 7, 1, 23, 25, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)), @@ -35,11 +35,12 @@ func TestRawFrameRepositoryQueriesRawFramesWithFilters(t *testing.T) { repository := NewRawFrameRepository(db, "lingniu_vehicle_ts") rows, err := repository.Query(context.Background(), RawFrameQuery{ - Protocol: "JT808", - VIN: "LKLG7C4E3NA774736", - DateFrom: "2026-07-01 00:00:00", - DateTo: "2026-07-01 23:59:59", - Limit: 20, + Protocol: "JT808", + VIN: "LKLG7C4E3NA774736", + DateFrom: "2026-07-01 00:00:00", + DateTo: "2026-07-01 23:59:59", + IncludeFields: true, + Limit: 20, }) if err != nil { t.Fatalf("Query() error = %v", err) @@ -47,7 +48,7 @@ func TestRawFrameRepositoryQueriesRawFramesWithFilters(t *testing.T) { if len(rows) != 1 { t.Fatalf("row count = %d", len(rows)) } - if rows[0].MessageID != 512 || rows[0].MessageIDHex != "0x0200" || rows[0].ParsedJSON == "" { + if rows[0].MessageID != 512 || rows[0].MessageIDHex != "0x0200" || rows[0].ParsedFields == "" { t.Fatalf("unexpected row: %#v", rows[0]) } if rows[0].TS != "2026-07-01 23:25:36" { @@ -58,18 +59,18 @@ func TestRawFrameRepositoryQueriesRawFramesWithFilters(t *testing.T) { } } -func TestRawFrameRepositoryHydratesChunkedParsedJSON(t *testing.T) { +func TestRawFrameRepositoryHydratesChunkedParsedFields(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock.New() error = %v", err) } defer db.Close() - manifest := `{"chunked":true,"payload_kind":"parsed_json","event_id":"event-oversized","chunk_count":2}` - mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames"). + manifest := `{"chunked":true,"payload_kind":"parsed_fields","event_id":"event-oversized","chunk_count":2}` + mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, .* AS parsed_fields, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_gb32960_"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_fields", "parse_status", "parse_error", "source_endpoint", "protocol", "vehicle_key", "vin", "phone", "device_id", }).AddRow( "2026-07-02 10:00:00", "go_frame", "event-oversized", 2, @@ -79,14 +80,15 @@ func TestRawFrameRepositoryHydratesChunkedParsedJSON(t *testing.T) { )) mock.ExpectQuery("SELECT event_id, payload_kind, chunk_index, chunk_text FROM lingniu_vehicle_ts.raw_frame_payload_chunks"). WillReturnRows(sqlmock.NewRows([]string{"event_id", "payload_kind", "chunk_index", "chunk_text"}). - AddRow("event-oversized", "parsed_json", 0, `{"data_units":[`). - AddRow("event-oversized", "parsed_json", 1, `{"field":"value"}]}`)) + AddRow("event-oversized", "parsed_fields", 0, `{"data_units":[`). + AddRow("event-oversized", "parsed_fields", 1, `{"field":"value"}]}`)) repository := NewRawFrameRepository(db, "lingniu_vehicle_ts") rows, err := repository.Query(context.Background(), RawFrameQuery{ - Protocol: "GB32960", - VIN: "LB9A32A21R0LS1707", - Limit: 1, + Protocol: "GB32960", + VIN: "LB9A32A21R0LS1707", + IncludeFields: true, + Limit: 1, }) if err != nil { t.Fatalf("Query() error = %v", err) @@ -94,8 +96,8 @@ func TestRawFrameRepositoryHydratesChunkedParsedJSON(t *testing.T) { if len(rows) != 1 { t.Fatalf("row count = %d", len(rows)) } - if want := `{"data_units":[{"field":"value"}]}`; rows[0].ParsedJSON != want { - t.Fatalf("hydrated parsed_json = %q, want %q", rows[0].ParsedJSON, want) + if want := `{"data_units":[{"field":"value"}]}`; rows[0].ParsedFields != want { + t.Fatalf("hydrated parsed_fields = %q, want %q", rows[0].ParsedFields, want) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) @@ -108,12 +110,12 @@ func TestRawFrameHandlerReturnsRawFrames(t *testing.T) { t.Fatalf("sqlmock.New() error = %v", err) } defer db.Close() - mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_frames"). + mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_gb32960_"). WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(38)) - mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames"). + mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, .* AS parsed_fields, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_gb32960_"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_fields", "parse_status", "parse_error", "source_endpoint", "protocol", "vehicle_key", "vin", "phone", "device_id", }).AddRow( "2026-07-01 22:28:25", "go_frame", "event-2", 2, "2026-07-01 22:28:25", "2026-07-01 22:28:25", @@ -150,12 +152,12 @@ func TestRawFrameHandlerFiltersByVehicleKey(t *testing.T) { t.Fatalf("sqlmock.New() error = %v", err) } defer db.Close() - mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_frames"). + mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_jt808_"). WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(11)) mock.ExpectQuery("vehicle_key = 'JT808:013307811350'"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_fields", "parse_status", "parse_error", "source_endpoint", "protocol", "vehicle_key", "vin", "phone", "device_id", }).AddRow( "2026-07-02 00:18:22", "go_frame", "event-3", 0x0200, "2026-07-02 00:18:22", "2026-07-02 00:22:43", @@ -192,7 +194,7 @@ func TestRawFrameHandlerCanSkipTotalCountForFreshnessProbe(t *testing.T) { mock.ExpectQuery("ORDER BY received_at DESC LIMIT 1 OFFSET 0"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_fields", "parse_status", "parse_error", "source_endpoint", "protocol", "vehicle_key", "vin", "phone", "device_id", }).AddRow( "2026-07-02 01:26:48", "go_frame", "event-4", 0x0200, "2026-07-02 01:20:46", "2026-07-02 01:26:48", @@ -229,7 +231,7 @@ func TestRawFrameHandlerSkipsTotalCountByDefault(t *testing.T) { mock.ExpectQuery("ORDER BY ts DESC LIMIT 1 OFFSET 0"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_fields", "parse_status", "parse_error", "source_endpoint", "protocol", "vehicle_key", "vin", "phone", "device_id", }).AddRow( "2026-07-02 01:26:48", "go_frame", "event-5", 0x0200, "2026-07-02 01:20:46", "2026-07-02 01:26:48", @@ -261,10 +263,10 @@ func TestRawFrameHandlerReturnsEmptyItemsArrayWhenNoRows(t *testing.T) { t.Fatalf("sqlmock.New() error = %v", err) } defer db.Close() - mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames"). + mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, .* AS parsed_fields, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_fields", "parse_status", "parse_error", "source_endpoint", "protocol", "vehicle_key", "vin", "phone", "device_id", })) diff --git a/go/vehicle-gateway/internal/history/writer.go b/go/vehicle-gateway/internal/history/writer.go index 43b36ba6..cd87ef1c 100644 --- a/go/vehicle-gateway/internal/history/writer.go +++ b/go/vehicle-gateway/internal/history/writer.go @@ -14,6 +14,7 @@ import ( "unicode/utf8" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime" ) type Execer interface { @@ -72,11 +73,11 @@ func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope) } rawHex, rawHexChunks := chunkPayload(env, "raw_hex", env.RawHex) rawText, rawTextChunks := chunkPayload(env, "raw_text", env.RawText) - parsedJSON, parsedChunks := chunkPayload(env, "parsed_json", parsedJSONString(env.Parsed)) + parsedFields, parsedChunks := chunkPayload(env, "parsed_fields", parsedFieldsJSONString(env)) _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s (ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint) -VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedJSON)))) +VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedFields)))) if err != nil { return err } @@ -169,7 +170,7 @@ func (w *Writer) ensureLocationChild(ctx context.Context, table string, env enve return nil } -func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedJSON string) []any { +func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedFields string) []any { received := millis(env.ReceivedAtMS) eventTime := millis(env.EventTimeMS) return []any{ @@ -182,7 +183,7 @@ func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsed rawSizeBytes(env), rawHex, rawText, - parsedJSON, + parsedFields, string(env.ParseStatus), env.ParseError, env.SourceEndpoint, @@ -316,11 +317,12 @@ func jsonString(value any) string { return string(data) } -func parsedJSONString(value map[string]any) string { - if len(value) == 0 { +func parsedFieldsJSONString(env envelope.FrameEnvelope) string { + fieldsEnv, ok := realtime.BuildFieldsEnvelope(env) + if !ok || len(fieldsEnv.Fields) == 0 { return "" } - return jsonString(value) + return jsonString(fieldsEnv.Fields) } func floatField(env envelope.FrameEnvelope, key string) (float64, bool) { diff --git a/go/vehicle-gateway/internal/history/writer_test.go b/go/vehicle-gateway/internal/history/writer_test.go index 13b6c94a..7d9cfd32 100644 --- a/go/vehicle-gateway/internal/history/writer_test.go +++ b/go/vehicle-gateway/internal/history/writer_test.go @@ -107,7 +107,7 @@ func TestWriterNormalizesPhoneTagAndRefreshesExistingChildTags(t *testing.T) { } } -func TestWriterChunksOversizedParsedJSON(t *testing.T) { +func TestWriterChunksOversizedParsedFields(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec) env := sampleEnvelope() @@ -120,8 +120,8 @@ func TestWriterChunksOversizedParsedJSON(t *testing.T) { } rawInsert := findSQL(exec.calls, "INSERT INTO raw_") - if !strings.Contains(rawInsert, `"chunked":true`) || !strings.Contains(rawInsert, `"payload_kind":"parsed_json"`) { - t.Fatalf("raw insert should store a parsed_json chunk manifest: %s", rawInsert) + if !strings.Contains(rawInsert, `"chunked":true`) || !strings.Contains(rawInsert, `"payload_kind":"parsed_fields"`) { + t.Fatalf("raw insert should store a parsed_fields chunk manifest: %s", rawInsert) } if got := countSQL(exec.calls, "USING raw_frame_payload_chunks"); got != 1 { t.Fatalf("chunk child create count = %d", got) @@ -131,7 +131,7 @@ func TestWriterChunksOversizedParsedJSON(t *testing.T) { } } -func TestWriterLeavesParsedJSONEmptyWhenNoParsedPayload(t *testing.T) { +func TestWriterLeavesParsedFieldsEmptyWhenNoParsedPayload(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec) env := sampleEnvelope() @@ -145,7 +145,7 @@ func TestWriterLeavesParsedJSONEmptyWhenNoParsedPayload(t *testing.T) { rawInsert := findSQL(exec.calls, "INSERT INTO raw_") if strings.Contains(rawInsert, "'{}'") || strings.Contains(rawInsert, "'null'") { - t.Fatalf("raw insert should not store empty parsed_json payload: %s", rawInsert) + t.Fatalf("raw insert should not store empty parsed_fields payload: %s", rawInsert) } if !strings.Contains(rawInsert, "'invalid frame checksum'") { t.Fatalf("raw insert should keep parse error: %s", rawInsert)