shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(go): optimize tdengine raw frame queries

Browse Source
This commit is contained in:
lingniu
2026-07-03 14:25:48 +08:00
parent 535acdb2a6
commit 6318c10c41
4 changed files with 123 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
go/vehicle-gateway/internal/history/query.go
View File

@@ -24,6 +24,8 @@ type RawFrameQuery struct {
DeviceID string DeviceID string
MessageID string MessageID string
OrderBy string OrderBy string
IncludeFields bool
IncludePayload bool
IncludeTotal bool IncludeTotal bool
DateFrom string DateFrom string
DateTo string DateTo string
@@ -42,7 +44,7 @@ type RawFrameRow struct {
RawSizeBytes int64 `json:"raw_size_bytes"` RawSizeBytes int64 `json:"raw_size_bytes"`
RawHex string `json:"raw_hex,omitempty"` RawHex string `json:"raw_hex,omitempty"`
RawText string `json:"raw_text,omitempty"` RawText string `json:"raw_text,omitempty"`
ParsedJSON string `json:"parsed_json,omitempty"` ParsedFields string `json:"parsed_fields,omitempty"`
ParseStatus string `json:"parse_status"` ParseStatus string `json:"parse_status"`
ParseError string `json:"parse_error,omitempty"` ParseError string `json:"parse_error,omitempty"`
SourceEndpoint string `json:"source_endpoint"` 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) { func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]RawFrameRow, error) {
query = normalizeRawFrameQuery(query) query = normalizeRawFrameQuery(query)
sqlText, args := buildRawFrameSQL(r.tableName(), query) sqlText, args := buildRawFrameSQL(r.tableName(query), query)
rows, err := r.db.QueryContext(ctx, sqlText, args...) rows, err := r.db.QueryContext(ctx, sqlText, args...)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -136,7 +138,7 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]
&row.RawSizeBytes, &row.RawSizeBytes,
&row.RawHex, &row.RawHex,
&row.RawText, &row.RawText,
&row.ParsedJSON, &row.ParsedFields,
&row.ParseStatus, &row.ParseStatus,
&row.ParseError, &row.ParseError,
&row.SourceEndpoint, &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) { func (r *RawFrameRepository) Count(ctx context.Context, query RawFrameQuery) (int64, error) {
query = normalizeRawFrameQuery(query) query = normalizeRawFrameQuery(query)
sqlText, args := buildRawFrameCountSQL(r.tableName(), query) sqlText, args := buildRawFrameCountSQL(r.tableName(query), query)
return countRows(ctx, r.db, sqlText, args...) 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() 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 == "" { if r.database == "" {
return "raw_frames" return "raw_frames"
} }
return r.database + ".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 { func (r *RawFrameRepository) chunkTableName() string {
if r.database == "" { if r.database == "" {
return "raw_frame_payload_chunks" return "raw_frame_payload_chunks"
@@ -290,7 +316,17 @@ func normalizeLocationQuery(query LocationQuery) LocationQuery {
func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) { func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) {
where := rawFrameWhere(query) 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 { if len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ") 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_hex", value: rows[index].RawHex},
{kind: "raw_text", value: rows[index].RawText}, {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) 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 continue
} }
eventID := strings.TrimSpace(manifest.EventID) eventID := strings.TrimSpace(manifest.EventID)
@@ -400,13 +436,20 @@ func (r *RawFrameRepository) hydratePayloadChunks(ctx context.Context, rows []Ra
rows[target.rowIndex].RawHex = hydrated rows[target.rowIndex].RawHex = hydrated
case "raw_text": case "raw_text":
rows[target.rowIndex].RawText = hydrated rows[target.rowIndex].RawText = hydrated
case "parsed_json": case "parsed_fields":
rows[target.rowIndex].ParsedJSON = hydrated rows[target.rowIndex].ParsedFields = hydrated
} }
} }
return nil 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) { func parsePayloadChunkManifest(value string) (payloadChunkManifest, bool) {
var manifest payloadChunkManifest var manifest payloadChunkManifest
if !strings.Contains(value, `"chunked"`) { if !strings.Contains(value, `"chunked"`) {
@@ -633,6 +676,8 @@ func parseRawFrameQuery(r *http.Request) (RawFrameQuery, error) {
DeviceID: values.Get("deviceId"), DeviceID: values.Get("deviceId"),
MessageID: values.Get("messageId"), MessageID: values.Get("messageId"),
OrderBy: values.Get("orderBy"), 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"), IncludeTotal: strings.EqualFold(strings.TrimSpace(values.Get("includeTotal")), "true"),
DateFrom: values.Get("dateFrom"), DateFrom: values.Get("dateFrom"),
DateTo: values.Get("dateTo"), DateTo: values.Get("dateTo"),

42
go/vehicle-gateway/internal/history/query_test.go
View File

@@ -19,10 +19,10 @@ func TestRawFrameRepositoryQueriesRawFramesWithFilters(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err) t.Fatalf("sqlmock.New() error = %v", err)
} }
defer db.Close() 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{ WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", "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", "protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow( }).AddRow(
time.Date(2026, 7, 1, 23, 25, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)), time.Date(2026, 7, 1, 23, 25, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
@@ -39,6 +39,7 @@ func TestRawFrameRepositoryQueriesRawFramesWithFilters(t *testing.T) {
VIN: "LKLG7C4E3NA774736", VIN: "LKLG7C4E3NA774736",
DateFrom: "2026-07-01 00:00:00", DateFrom: "2026-07-01 00:00:00",
DateTo: "2026-07-01 23:59:59", DateTo: "2026-07-01 23:59:59",
IncludeFields: true,
Limit: 20, Limit: 20,
}) })
if err != nil { if err != nil {
@@ -47,7 +48,7 @@ func TestRawFrameRepositoryQueriesRawFramesWithFilters(t *testing.T) {
if len(rows) != 1 { if len(rows) != 1 {
t.Fatalf("row count = %d", len(rows)) 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]) t.Fatalf("unexpected row: %#v", rows[0])
} }
if rows[0].TS != "2026-07-01 23:25:36" { 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() db, mock, err := sqlmock.New()
if err != nil { if err != nil {
t.Fatalf("sqlmock.New() error = %v", err) t.Fatalf("sqlmock.New() error = %v", err)
} }
defer db.Close() defer db.Close()
manifest := `{"chunked":true,"payload_kind":"parsed_json","event_id":"event-oversized","chunk_count":2}` 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, 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{ WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", "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", "protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow( }).AddRow(
"2026-07-02 10:00:00", "go_frame", "event-oversized", 2, "2026-07-02 10:00:00", "go_frame", "event-oversized", 2,
@@ -79,13 +80,14 @@ func TestRawFrameRepositoryHydratesChunkedParsedJSON(t *testing.T) {
)) ))
mock.ExpectQuery("SELECT event_id, payload_kind, chunk_index, chunk_text FROM lingniu_vehicle_ts.raw_frame_payload_chunks"). 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"}). WillReturnRows(sqlmock.NewRows([]string{"event_id", "payload_kind", "chunk_index", "chunk_text"}).
AddRow("event-oversized", "parsed_json", 0, `{"data_units":[`). AddRow("event-oversized", "parsed_fields", 0, `{"data_units":[`).
AddRow("event-oversized", "parsed_json", 1, `{"field":"value"}]}`)) AddRow("event-oversized", "parsed_fields", 1, `{"field":"value"}]}`))
repository := NewRawFrameRepository(db, "lingniu_vehicle_ts") repository := NewRawFrameRepository(db, "lingniu_vehicle_ts")
rows, err := repository.Query(context.Background(), RawFrameQuery{ rows, err := repository.Query(context.Background(), RawFrameQuery{
Protocol: "GB32960", Protocol: "GB32960",
VIN: "LB9A32A21R0LS1707", VIN: "LB9A32A21R0LS1707",
IncludeFields: true,
Limit: 1, Limit: 1,
}) })
if err != nil { if err != nil {
@@ -94,8 +96,8 @@ func TestRawFrameRepositoryHydratesChunkedParsedJSON(t *testing.T) {
if len(rows) != 1 { if len(rows) != 1 {
t.Fatalf("row count = %d", len(rows)) t.Fatalf("row count = %d", len(rows))
} }
if want := `{"data_units":[{"field":"value"}]}`; rows[0].ParsedJSON != want { if want := `{"data_units":[{"field":"value"}]}`; rows[0].ParsedFields != want {
t.Fatalf("hydrated parsed_json = %q, want %q", rows[0].ParsedJSON, want) t.Fatalf("hydrated parsed_fields = %q, want %q", rows[0].ParsedFields, want)
} }
if err := mock.ExpectationsWereMet(); err != nil { if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err) t.Fatalf("sql expectations: %v", err)
@@ -108,12 +110,12 @@ func TestRawFrameHandlerReturnsRawFrames(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err) t.Fatalf("sqlmock.New() error = %v", err)
} }
defer db.Close() 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)) 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{ WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", "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", "protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow( }).AddRow(
"2026-07-01 22:28:25", "go_frame", "event-2", 2, "2026-07-01 22:28:25", "2026-07-01 22:28:25", "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) t.Fatalf("sqlmock.New() error = %v", err)
} }
defer db.Close() 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)) WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(11))
mock.ExpectQuery("vehicle_key = 'JT808:013307811350'"). mock.ExpectQuery("vehicle_key = 'JT808:013307811350'").
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", "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", "protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow( }).AddRow(
"2026-07-02 00:18:22", "go_frame", "event-3", 0x0200, "2026-07-02 00:18:22", "2026-07-02 00:22:43", "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"). mock.ExpectQuery("ORDER BY received_at DESC LIMIT 1 OFFSET 0").
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", "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", "protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow( }).AddRow(
"2026-07-02 01:26:48", "go_frame", "event-4", 0x0200, "2026-07-02 01:20:46", "2026-07-02 01:26:48", "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"). mock.ExpectQuery("ORDER BY ts DESC LIMIT 1 OFFSET 0").
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", "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", "protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow( }).AddRow(
"2026-07-02 01:26:48", "go_frame", "event-5", 0x0200, "2026-07-02 01:20:46", "2026-07-02 01:26:48", "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) t.Fatalf("sqlmock.New() error = %v", err)
} }
defer db.Close() 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{ WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", "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", "protocol", "vehicle_key", "vin", "phone", "device_id",
})) }))

16
go/vehicle-gateway/internal/history/writer.go
View File

@@ -14,6 +14,7 @@ import (
"unicode/utf8" "unicode/utf8"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime"
) )
type Execer interface { 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) rawHex, rawHexChunks := chunkPayload(env, "raw_hex", env.RawHex)
rawText, rawTextChunks := chunkPayload(env, "raw_text", env.RawText) 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 _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
(ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, (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_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 { if err != nil {
return err return err
} }
@@ -169,7 +170,7 @@ func (w *Writer) ensureLocationChild(ctx context.Context, table string, env enve
return nil 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) received := millis(env.ReceivedAtMS)
eventTime := millis(env.EventTimeMS) eventTime := millis(env.EventTimeMS)
return []any{ return []any{
@@ -182,7 +183,7 @@ func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsed
rawSizeBytes(env), rawSizeBytes(env),
rawHex, rawHex,
rawText, rawText,
parsedJSON, parsedFields,
string(env.ParseStatus), string(env.ParseStatus),
env.ParseError, env.ParseError,
env.SourceEndpoint, env.SourceEndpoint,
@@ -316,11 +317,12 @@ func jsonString(value any) string {
return string(data) return string(data)
} }
func parsedJSONString(value map[string]any) string { func parsedFieldsJSONString(env envelope.FrameEnvelope) string {
if len(value) == 0 { fieldsEnv, ok := realtime.BuildFieldsEnvelope(env)
if !ok || len(fieldsEnv.Fields) == 0 {
return "" return ""
} }
return jsonString(value) return jsonString(fieldsEnv.Fields)
} }
func floatField(env envelope.FrameEnvelope, key string) (float64, bool) { func floatField(env envelope.FrameEnvelope, key string) (float64, bool) {

10
go/vehicle-gateway/internal/history/writer_test.go
View File

@@ -107,7 +107,7 @@ func TestWriterNormalizesPhoneTagAndRefreshesExistingChildTags(t *testing.T) {
} }
} }
func TestWriterChunksOversizedParsedJSON(t *testing.T) { func TestWriterChunksOversizedParsedFields(t *testing.T) {
exec := &recordingExec{} exec := &recordingExec{}
writer := NewWriter(exec) writer := NewWriter(exec)
env := sampleEnvelope() env := sampleEnvelope()
@@ -120,8 +120,8 @@ func TestWriterChunksOversizedParsedJSON(t *testing.T) {
} }
rawInsert := findSQL(exec.calls, "INSERT INTO raw_") rawInsert := findSQL(exec.calls, "INSERT INTO raw_")
if !strings.Contains(rawInsert, `"chunked":true`) || !strings.Contains(rawInsert, `"payload_kind":"parsed_json"`) { if !strings.Contains(rawInsert, `"chunked":true`) || !strings.Contains(rawInsert, `"payload_kind":"parsed_fields"`) {
t.Fatalf("raw insert should store a parsed_json chunk manifest: %s", rawInsert) 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 { if got := countSQL(exec.calls, "USING raw_frame_payload_chunks"); got != 1 {
t.Fatalf("chunk child create count = %d", got) 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{} exec := &recordingExec{}
writer := NewWriter(exec) writer := NewWriter(exec)
env := sampleEnvelope() env := sampleEnvelope()
@@ -145,7 +145,7 @@ func TestWriterLeavesParsedJSONEmptyWhenNoParsedPayload(t *testing.T) {
rawInsert := findSQL(exec.calls, "INSERT INTO raw_") rawInsert := findSQL(exec.calls, "INSERT INTO raw_")
if strings.Contains(rawInsert, "'{}'") || strings.Contains(rawInsert, "'null'") { 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'") { if !strings.Contains(rawInsert, "'invalid frame checksum'") {
t.Fatalf("raw insert should keep parse error: %s", rawInsert) t.Fatalf("raw insert should keep parse error: %s", rawInsert)