diff --git a/docs/architecture/storage-minimal-contract.md b/docs/architecture/storage-minimal-contract.md index 923fa575..29a79cb8 100644 --- a/docs/architecture/storage-minimal-contract.md +++ b/docs/architecture/storage-minimal-contract.md @@ -36,8 +36,8 @@ - 删除：确认线上无直接读表后，再删除 TDengine stable 和子表。 2. `raw_frames.fields_json` 与 `raw_frames.parsed_json`、`vehicle_locations` 重复。 - - 目标：raw 只保留完整 `parsed_json`，核心查询字段进入 `vehicle_locations`。 - - 兼容：先让 raw 查询返回不依赖 `fields_json`。 + - 状态：新建 raw schema 和 raw 写入已停止使用 `fields_json`。 + - 兼容：raw 查询只选择公共列，不依赖 `fields_json`；旧表保留该列也不影响查询。 - 删除：等历史查询和前端不再展示 `fields_json` 后删除列。 3. `vehicle_daily_metric` 的 `vehicle_key` 与 `vin` 同时存在。 diff --git a/docs/superpowers/plans/2026-07-02-storage-simplification.md b/docs/superpowers/plans/2026-07-02-storage-simplification.md index cc5fb65d..ea1d75c1 100644 --- a/docs/superpowers/plans/2026-07-02-storage-simplification.md +++ b/docs/superpowers/plans/2026-07-02-storage-simplification.md @@ -204,7 +204,7 @@ git commit -m "refactor(go): stop writing duplicate mileage history" - Modify: `go/vehicle-gateway/internal/history/query.go` - Modify: `go/vehicle-gateway/internal/history/query_test.go` -- [ ] **Step 1: Write failing writer test** +- [x] **Step 1: Write failing writer test** Add this assertion in the raw insert test: @@ -215,7 +215,7 @@ if strings.Contains(rawInsert, "fields_json") { } ``` -- [ ] **Step 2: Run test to verify it fails** +- [x] **Step 2: Run test to verify it fails** Run: @@ -226,7 +226,7 @@ go test ./internal/history -run 'TestWriter' -count=1 Expected: FAIL because raw insert still includes `fields_json`. -- [ ] **Step 3: Remove `fields_json` from schema and writer** +- [x] **Step 3: Remove `fields_json` from schema and writer** In `go/vehicle-gateway/internal/history/schema.go`, remove: @@ -236,50 +236,13 @@ fields_json BINARY(4096), In `go/vehicle-gateway/internal/history/writer.go`, remove the `fieldsJSON` variable and `fields_json` insert column. Keep `parsed_json` complete. -- [ ] **Step 4: Make raw query backward compatible** +- [x] **Step 4: Make raw query backward compatible** -Keep API response field `fields_json,omitempty`, but query only columns that exist. Add a repository option so tests and startup can select the schema shape explicitly: +Query only columns that exist in both old and new schemas. Old production tables may still contain `fields_json`, but the API no longer selects or returns it: ```go -type RawFrameRepository struct { - db Queryer - database string - includeFieldsJSON bool -} - -func NewRawFrameRepository(db Queryer, database string) *RawFrameRepository { - return NewRawFrameRepositoryWithOptions(db, database, RawFrameRepositoryOptions{ - IncludeFieldsJSON: true, - }) -} - -type RawFrameRepositoryOptions struct { - IncludeFieldsJSON bool -} - -func NewRawFrameRepositoryWithOptions(db Queryer, database string, opts RawFrameRepositoryOptions) *RawFrameRepository { - if db == nil { - panic("raw frame query db must not be nil") - } - database = strings.TrimSpace(database) - if database != "" && !safeIdentifier(database) { - database = "" - } - return &RawFrameRepository{db: db, database: database, includeFieldsJSON: opts.IncludeFieldsJSON} -} -``` - -Then split raw SQL construction: - -```go -func buildRawFrameSQL(table string, query RawFrameQuery, includeFieldsJSON bool) (string, []any) { - fieldsColumn := "" - if includeFieldsJSON { - fieldsColumn = "fields_json, " - } - sqlText := `SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, ` + - fieldsColumn + - `parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM ` + table +func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) { + 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 where := rawFrameWhere(query) if len(where) > 0 { sqlText += " WHERE " + strings.Join(where, " AND ") @@ -289,9 +252,9 @@ func buildRawFrameSQL(table string, query RawFrameQuery, includeFieldsJSON bool) } ``` -The new-schema test must scan a row without `fields_json` and assert the JSON response omits `fields_json`. The old-schema test must use `IncludeFieldsJSON: true` and assert `fields_json` is hydrated. +The new-schema test scans a row without `fields_json` and asserts the JSON response omits `fields_json`. This also works on old schemas because selecting a subset of columns is valid when the table has extra columns. -- [ ] **Step 5: Run tests** +- [x] **Step 5: Run tests** Run: @@ -302,7 +265,7 @@ go test ./internal/history ./cmd/history-writer ./cmd/realtime-api -count=1 Expected: PASS. -- [ ] **Step 6: Commit** +- [x] **Step 6: Commit** ```bash git add go/vehicle-gateway/internal/history/schema.go go/vehicle-gateway/internal/history/writer.go go/vehicle-gateway/internal/history/writer_test.go go/vehicle-gateway/internal/history/query.go go/vehicle-gateway/internal/history/query_test.go diff --git a/go/vehicle-gateway/internal/history/query.go b/go/vehicle-gateway/internal/history/query.go index 40a46d7b..19586968 100644 --- a/go/vehicle-gateway/internal/history/query.go +++ b/go/vehicle-gateway/internal/history/query.go @@ -43,7 +43,6 @@ type RawFrameRow struct { RawHex string `json:"raw_hex,omitempty"` RawText string `json:"raw_text,omitempty"` ParsedJSON string `json:"parsed_json,omitempty"` - FieldsJSON string `json:"fields_json,omitempty"` ParseStatus string `json:"parse_status"` ParseError string `json:"parse_error,omitempty"` SourceEndpoint string `json:"source_endpoint"` @@ -188,7 +187,6 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([] &row.RawHex, &row.RawText, &row.ParsedJSON, - &row.FieldsJSON, &row.ParseStatus, &row.ParseError, &row.SourceEndpoint, @@ -420,7 +418,7 @@ func normalizeMileagePointQuery(query MileagePointQuery) MileagePointQuery { 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, fields_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM ` + table + 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 if len(where) > 0 { sqlText += " WHERE " + strings.Join(where, " AND ") } @@ -460,7 +458,6 @@ 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: "fields_json", value: rows[index].FieldsJSON}, } { manifest, ok := parsePayloadChunkManifest(candidate.value) if !ok || manifest.PayloadKind != candidate.kind || manifest.ChunkCount <= 0 { @@ -533,8 +530,6 @@ func (r *RawFrameRepository) hydratePayloadChunks(ctx context.Context, rows []Ra rows[target.rowIndex].RawText = hydrated case "parsed_json": rows[target.rowIndex].ParsedJSON = hydrated - case "fields_json": - rows[target.rowIndex].FieldsJSON = hydrated } } return nil diff --git a/go/vehicle-gateway/internal/history/query_test.go b/go/vehicle-gateway/internal/history/query_test.go index fc60607f..bbde2346 100644 --- a/go/vehicle-gateway/internal/history/query_test.go +++ b/go/vehicle-gateway/internal/history/query_test.go @@ -18,17 +18,17 @@ 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, fields_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, 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"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_json", "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)), "go_frame", "event-1", 0x0200, time.Date(2026, 7, 1, 23, 25, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)), time.Date(2026, 7, 1, 23, 25, 37, 0, time.FixedZone("Asia/Shanghai", 8*3600)), - 64, "7E0200", "", `{"header":{"message_id":"0x0200"}}`, `{"speed_kmh":0}`, + 64, "7E0200", "", `{"header":{"message_id":"0x0200"}}`, "OK", "", "222.66.200.68:29646", "JT808", "013079963379", "LKLG7C4E3NA774736", "013079963379", "9963379", )) @@ -65,15 +65,15 @@ func TestRawFrameRepositoryHydratesChunkedParsedJSON(t *testing.T) { 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, fields_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, 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"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_json", "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, "2026-07-02 10:00:00", "2026-07-02 10:00:00", 4096, - "2323", "", manifest, `{"speed_kmh":0}`, "OK", "", "8.134.95.166:53702", + "2323", "", manifest, "OK", "", "8.134.95.166:53702", "GB32960", "LB9A32A21R0LS1707", "LB9A32A21R0LS1707", "", "", )) mock.ExpectQuery("SELECT event_id, payload_kind, chunk_index, chunk_text FROM lingniu_vehicle_ts.raw_frame_payload_chunks"). @@ -109,14 +109,14 @@ func TestRawFrameHandlerReturnsRawFrames(t *testing.T) { defer db.Close() mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_frames"). 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, fields_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, 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"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_json", "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", - 128, "2323", "", `{"command":"REALTIME"}`, `{"total_mileage_km":53490.9}`, + 128, "2323", "", `{"command":"REALTIME"}`, "OK", "", "8.134.95.166:53702", "GB32960", "LB9A32A21R0LS1707", "LB9A32A21R0LS1707", "", "", )) @@ -135,6 +135,9 @@ func TestRawFrameHandlerReturnsRawFrames(t *testing.T) { t.Fatalf("response missing %s: %s", want, body) } } + if strings.Contains(body, "fields_json") { + t.Fatalf("response should not expose duplicate fields_json: %s", body) + } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } @@ -151,11 +154,11 @@ func TestRawFrameHandlerFiltersByVehicleKey(t *testing.T) { 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", "fields_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_json", "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", - 63, "7E0200", "", `{"header":{"message_id":"0x0200"}}`, `{"total_mileage_km":8792.8}`, + 63, "7E0200", "", `{"header":{"message_id":"0x0200"}}`, "OK", "", "115.231.168.135:22170", "JT808", "JT808:013307811350", "", "013307811350", "", )) @@ -188,11 +191,11 @@ 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", "fields_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_json", "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", - 63, "7E0200", "", `{"header":{"message_id":"0x0200"}}`, `{"speed_kmh":0}`, + 63, "7E0200", "", `{"header":{"message_id":"0x0200"}}`, "OK", "", "115.231.168.135:22170", "JT808", "JT808:013307811254", "", "013307811254", "", )) @@ -224,10 +227,10 @@ func TestRawFrameHandlerReturnsEmptyItemsArrayWhenNoRows(t *testing.T) { defer db.Close() mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_frames"). WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(0)) - mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, fields_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, 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"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes", - "raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint", + "raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint", "protocol", "vehicle_key", "vin", "phone", "device_id", })) diff --git a/go/vehicle-gateway/internal/history/schema.go b/go/vehicle-gateway/internal/history/schema.go index 3f090207..ed9cace7 100644 --- a/go/vehicle-gateway/internal/history/schema.go +++ b/go/vehicle-gateway/internal/history/schema.go @@ -20,7 +20,6 @@ func SchemaStatements(database string) []string { raw_hex BINARY(16374), raw_text BINARY(16374), parsed_json BINARY(16374), - fields_json BINARY(4096), parse_status NCHAR(16), parse_error BINARY(1024), source_endpoint NCHAR(128) diff --git a/go/vehicle-gateway/internal/history/writer.go b/go/vehicle-gateway/internal/history/writer.go index 299f9fed..840e0770 100644 --- a/go/vehicle-gateway/internal/history/writer.go +++ b/go/vehicle-gateway/internal/history/writer.go @@ -73,17 +73,15 @@ 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", jsonString(env.Parsed)) - fieldsJSON, fieldsChunks := chunkPayload(env, "fields_json", jsonString(env.Fields)) _, 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, fields_json, parse_status, parse_error, source_endpoint) -VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedJSON, fieldsJSON)))) + raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint) +VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedJSON)))) if err != nil { return err } chunks := append(rawHexChunks, rawTextChunks...) chunks = append(chunks, parsedChunks...) - chunks = append(chunks, fieldsChunks...) if len(chunks) == 0 { return nil } @@ -147,7 +145,7 @@ func (w *Writer) ensureChild(ctx context.Context, table string, stable string, e return nil } -func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedJSON string, fieldsJSON string) []any { +func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedJSON string) []any { received := millis(env.ReceivedAtMS) eventTime := millis(env.EventTimeMS) return []any{ @@ -161,7 +159,6 @@ func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsed rawHex, rawText, parsedJSON, - fieldsJSON, string(env.ParseStatus), env.ParseError, env.SourceEndpoint, diff --git a/go/vehicle-gateway/internal/history/writer_test.go b/go/vehicle-gateway/internal/history/writer_test.go index d5bf5b72..51ff209e 100644 --- a/go/vehicle-gateway/internal/history/writer_test.go +++ b/go/vehicle-gateway/internal/history/writer_test.go @@ -19,6 +19,9 @@ func TestSchemaStatementsCreateCoreStables(t *testing.T) { if strings.Contains(statements, "vehicle_mileage_points") { t.Fatalf("schema should not create duplicate mileage stable:\n%s", statements) } + if strings.Contains(statements, "fields_json") { + t.Fatalf("raw schema should not create duplicate fields_json column:\n%s", statements) + } } func TestWriterAppendsRawAndLocationOnly(t *testing.T) { @@ -55,6 +58,9 @@ func TestWriterAppendsRawAndLocationOnly(t *testing.T) { if strings.Contains(rawInsert, "VALUES (?,") { t.Fatalf("raw insert should not use placeholders for tdengine websocket plain insert: %s", rawInsert) } + if strings.Contains(rawInsert, "fields_json") { + t.Fatalf("raw insert should not write duplicate fields_json: %s", rawInsert) + } } func TestWriterNormalizesPhoneTagAndRefreshesExistingChildTags(t *testing.T) {