diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 92867e08..76e8df6d 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -3,7 +3,6 @@ package realtime import ( "context" "database/sql" - "encoding/json" "errors" "strings" "time" @@ -39,8 +38,10 @@ func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { if _, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL); err != nil { return err } - _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL) - return err + if _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL); err != nil { + return err + } + return w.dropObsoleteColumns(ctx) } func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) error { @@ -55,14 +56,6 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) if err != nil { return err } - fieldsJSON, err := marshalObject(env.Fields) - if err != nil { - return err - } - parsedJSON, err := marshalObject(env.Parsed) - if err != nil { - return err - } eventTime := nullableTime(env.EventTimeMS) receivedAt := nullableTime(env.ReceivedAtMS) if _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL, @@ -76,14 +69,12 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) env.Sequence, strings.TrimSpace(env.SourceEndpoint), eventTime, - parsedJSON, - fieldsJSON, receivedAt, env.StableEventID(), ); err != nil { return err } - location, ok := realtimeLocationFromEnvelope(env, vehicleKey, fieldsJSON, plate) + location, ok := realtimeLocationFromEnvelope(env, vehicleKey, plate) if !ok { return nil } @@ -107,7 +98,6 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) location.DirectionDeg, location.AlarmFlag, location.StatusFlag, - location.FieldsJSON, location.ReceivedAt, location.EventID, ) @@ -135,6 +125,47 @@ func (w *SnapshotWriter) plateForEnvelope(ctx context.Context, env envelope.Fram return strings.TrimSpace(plate), nil } +func (w *SnapshotWriter) dropObsoleteColumns(ctx context.Context) error { + queryer, ok := w.exec.(Queryer) + if !ok { + return nil + } + for _, column := range []struct { + table string + name string + }{ + {table: "vehicle_realtime_snapshot", name: "parsed_json"}, + {table: "vehicle_realtime_snapshot", name: "fields_json"}, + {table: "vehicle_realtime_location", name: "fields_json"}, + } { + exists, err := columnExists(ctx, queryer, column.table, column.name) + if err != nil { + return err + } + if !exists { + continue + } + if _, err := w.exec.ExecContext(ctx, "ALTER TABLE "+column.table+" DROP COLUMN "+column.name); err != nil { + return err + } + } + return nil +} + +func columnExists(ctx context.Context, queryer Queryer, table string, column string) (bool, error) { + var count int + err := queryer.QueryRowContext(ctx, ` +SELECT COUNT(*) +FROM information_schema.columns +WHERE table_schema = DATABASE() + AND table_name = ? + AND column_name = ?`, table, column).Scan(&count) + if err != nil { + return false, err + } + return count > 0, nil +} + type realtimeLocationRow struct { Protocol string VehicleKey string @@ -155,12 +186,11 @@ type realtimeLocationRow struct { DirectionDeg any AlarmFlag any StatusFlag any - FieldsJSON string ReceivedAt any EventID string } -func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, fieldsJSON string, plate string) (realtimeLocationRow, bool) { +func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, plate string) (realtimeLocationRow, bool) { latitude, okLat := numberField(env.Fields, envelope.FieldLatitude) longitude, okLon := numberField(env.Fields, envelope.FieldLongitude) if !okLat || !okLon { @@ -186,7 +216,6 @@ func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, DirectionDeg: nullableNumberField(env.Fields, "direction_deg"), AlarmFlag: nullableNumberField(env.Fields, "alarm_flag"), StatusFlag: nullableNumberField(env.Fields, "status_flag"), - FieldsJSON: fieldsJSON, ReceivedAt: nullableTime(env.ReceivedAtMS), EventID: env.StableEventID(), }, true @@ -282,17 +311,6 @@ func safeIdentifier(value string) bool { return true } -func marshalObject(value map[string]any) (string, error) { - if value == nil { - value = map[string]any{} - } - payload, err := json.Marshal(value) - if err != nil { - return "", err - } - return string(payload), nil -} - func nullableTime(ms int64) any { if ms <= 0 { return nil @@ -322,8 +340,6 @@ const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_sn sequence_id INT NOT NULL DEFAULT 0, source_endpoint VARCHAR(128) NOT NULL DEFAULT '', event_time DATETIME(3) NULL, - parsed_json JSON NOT NULL, - fields_json JSON NOT NULL, received_at DATETIME(3) NULL, event_id VARCHAR(64) NOT NULL DEFAULT '', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, @@ -337,8 +353,8 @@ const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_sn const upsertRealtimeSnapshotSQL = ` INSERT INTO vehicle_realtime_snapshot (protocol, vehicle_key, vin, phone, device_id, plate, message_id, sequence_id, - source_endpoint, event_time, parsed_json, fields_json, received_at, event_id) -VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), CAST(? AS JSON), ?, ?) + source_endpoint, event_time, received_at, event_id) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE vin = IF(VALUES(vin) <> '', VALUES(vin), vin), phone = IF(VALUES(phone) <> '', VALUES(phone), phone), @@ -348,8 +364,6 @@ ON DUPLICATE KEY UPDATE sequence_id = VALUES(sequence_id), source_endpoint = VALUES(source_endpoint), event_time = VALUES(event_time), - parsed_json = VALUES(parsed_json), - fields_json = VALUES(fields_json), received_at = VALUES(received_at), event_id = VALUES(event_id), updated_at = CURRENT_TIMESTAMP @@ -376,7 +390,6 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo direction_deg DECIMAL(10,3) NULL, alarm_flag BIGINT NULL, status_flag BIGINT NULL, - fields_json JSON NOT NULL, received_at DATETIME(3) NULL, event_id VARCHAR(64) NOT NULL DEFAULT '', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, @@ -392,9 +405,8 @@ const upsertRealtimeLocationSQL = ` INSERT INTO vehicle_realtime_location (protocol, vehicle_key, vin, phone, device_id, plate, message_id, sequence_id, source_endpoint, event_time, latitude, longitude, speed_kmh, total_mileage_km, - soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, fields_json, - received_at, event_id) -VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?) + soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE vin = IF(VALUES(vin) <> '', VALUES(vin), vin), phone = IF(VALUES(phone) <> '', VALUES(phone), phone), @@ -413,7 +425,6 @@ ON DUPLICATE KEY UPDATE direction_deg = VALUES(direction_deg), alarm_flag = VALUES(alarm_flag), status_flag = VALUES(status_flag), - fields_json = VALUES(fields_json), received_at = VALUES(received_at), event_id = VALUES(event_id), updated_at = CURRENT_TIMESTAMP diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index efcaadaa..95538857 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -10,7 +10,7 @@ import ( "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" ) -func TestSnapshotWriterEnsuresSchemaAndUpsertsParsedJSON(t *testing.T) { +func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { exec := &recordingSnapshotExec{} writer := NewSnapshotWriter(exec) @@ -46,22 +46,26 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsParsedJSON(t *testing.T) { if !strings.Contains(exec.calls[1].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") { t.Fatalf("location schema query = %s", exec.calls[1].query) } + for _, call := range exec.calls[:2] { + if strings.Contains(call.query, "parsed_json") || strings.Contains(call.query, "fields_json") { + t.Fatalf("realtime schema should not contain json payload columns: %s", call.query) + } + } upsert := exec.calls[2] if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") { t.Fatalf("upsert query = %s", upsert.query) } + if strings.Contains(upsert.query, "parsed_json") || strings.Contains(upsert.query, "fields_json") { + t.Fatalf("snapshot upsert should not write json payload columns: %s", upsert.query) + } if got, want := upsert.args[0], "GB32960"; got != want { t.Fatalf("protocol arg = %#v, want %q", got, want) } if got, want := upsert.args[1], "VIN001"; got != want { t.Fatalf("vehicle key arg = %#v, want %q", got, want) } - parsedJSON, ok := upsert.args[10].(string) - if !ok { - t.Fatalf("parsed_json arg type = %T", upsert.args[10]) - } - if !strings.Contains(parsedJSON, "gd_fc_vendor_tlv") || !strings.Contains(parsedJSON, "vehicle") { - t.Fatalf("parsed_json should contain original and vendor parsed data: %s", parsedJSON) + if len(upsert.args) != 12 { + t.Fatalf("snapshot upsert args = %d, want 12", len(upsert.args)) } } @@ -104,6 +108,9 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location") { t.Fatalf("location upsert query = %s", locationUpsert.query) } + if strings.Contains(locationUpsert.query, "fields_json") { + t.Fatalf("location upsert should not write fields_json: %s", locationUpsert.query) + } if got, want := locationUpsert.args[0], "JT808"; got != want { t.Fatalf("protocol arg = %#v, want %q", got, want) } @@ -125,6 +132,9 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) if got, want := locationUpsert.args[14], 88.0; got != want { t.Fatalf("soc arg = %#v, want %v", got, want) } + if len(locationUpsert.args) != 21 { + t.Fatalf("location upsert args = %d, want 21", len(locationUpsert.args)) + } } func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) {