diff --git a/docs/go-version-data-flow.html b/docs/go-version-data-flow.html index de2b6598..27d738eb 100644 --- a/docs/go-version-data-flow.html +++ b/docs/go-version-data-flow.html @@ -408,7 +408,7 @@
落库与查询
history-writer -

消费 RAW topic,写 TDengine:raw_frames、payload_chunks、locations、mileage_points。

+

消费 RAW topic,写 TDengine:raw_frames、payload_chunks、locations。

stat-writer diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index ef7ea7d8..1c27a8dc 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -41,7 +41,7 @@ func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { if _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL); err != nil { return err } - return w.dropObsoleteColumns(ctx) + return nil } func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) error { @@ -113,126 +113,6 @@ 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 - } - if err := w.ensureRealtimeVINIndexes(ctx, queryer); err != nil { - return err - } - 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_snapshot", name: "vehicle_key"}, - {table: "vehicle_realtime_snapshot", name: "phone"}, - {table: "vehicle_realtime_snapshot", name: "device_id"}, - {table: "vehicle_realtime_snapshot", name: "message_id"}, - {table: "vehicle_realtime_snapshot", name: "sequence_id"}, - {table: "vehicle_realtime_snapshot", name: "source_endpoint"}, - {table: "vehicle_realtime_location", name: "fields_json"}, - {table: "vehicle_realtime_location", name: "vehicle_key"}, - {table: "vehicle_realtime_location", name: "phone"}, - {table: "vehicle_realtime_location", name: "device_id"}, - {table: "vehicle_realtime_location", name: "message_id"}, - {table: "vehicle_realtime_location", name: "sequence_id"}, - {table: "vehicle_realtime_location", name: "source_endpoint"}, - } { - 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 (w *SnapshotWriter) ensureRealtimeVINIndexes(ctx context.Context, queryer Queryer) error { - for _, item := range []struct { - table string - oldIndex string - newIndex string - }{ - {table: "vehicle_realtime_snapshot", oldIndex: "uk_realtime_snapshot_vehicle", newIndex: "uk_realtime_snapshot_vin"}, - {table: "vehicle_realtime_location", oldIndex: "uk_realtime_location_vehicle", newIndex: "uk_realtime_location_vin"}, - } { - if err := w.normalizeVINRows(ctx, item.table); err != nil { - return err - } - exists, err := indexExists(ctx, queryer, item.table, item.newIndex) - if err != nil { - return err - } - if !exists { - if _, err := w.exec.ExecContext(ctx, "ALTER TABLE "+item.table+" ADD UNIQUE KEY "+item.newIndex+" (protocol, vin)"); err != nil { - return err - } - } - exists, err = indexExists(ctx, queryer, item.table, item.oldIndex) - if err != nil { - return err - } - if exists { - if _, err := w.exec.ExecContext(ctx, "ALTER TABLE "+item.table+" DROP INDEX "+item.oldIndex); err != nil { - return err - } - } - } - return nil -} - -func (w *SnapshotWriter) normalizeVINRows(ctx context.Context, table string) error { - if _, err := w.exec.ExecContext(ctx, "DELETE FROM "+table+" WHERE vin IS NULL OR vin = ''"); err != nil { - return err - } - _, err := w.exec.ExecContext(ctx, `DELETE newer_duplicate -FROM `+table+` newer_duplicate -JOIN `+table+` latest - ON newer_duplicate.protocol = latest.protocol - AND newer_duplicate.vin = latest.vin - AND ( - newer_duplicate.updated_at < latest.updated_at - OR (newer_duplicate.updated_at = latest.updated_at AND newer_duplicate.id < latest.id) - )`) - return err -} - -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 -} - -func indexExists(ctx context.Context, queryer Queryer, table string, index string) (bool, error) { - var count int - err := queryer.QueryRowContext(ctx, ` -SELECT COUNT(*) -FROM information_schema.statistics -WHERE table_schema = DATABASE() - AND table_name = ? - AND index_name = ?`, table, index).Scan(&count) - if err != nil { - return false, err - } - return count > 0, nil -} - type realtimeLocationRow struct { Protocol string VIN string diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 0251c7e8..17f66bec 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -7,6 +7,8 @@ import ( "strings" "testing" + "github.com/DATA-DOG/go-sqlmock" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" ) @@ -79,6 +81,27 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { } } +func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + defer db.Close() + writer := NewSnapshotWriter(db) + + mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location"). + WillReturnResult(sqlmock.NewResult(0, 0)) + + if err := writer.EnsureSchema(context.Background()); err != nil { + t.Fatalf("EnsureSchema() error = %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unexpected compatibility migration query: %v", err) + } +} + func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) { exec := &recordingSnapshotExec{} writer := NewSnapshotWriter(exec)