| 383 | } |
| 384 | |
| 385 | func convertToTaskRunLogEntries(logs []*store.TaskRunLog) []*v1pb.TaskRunLogEntry { |
| 386 | var entries []*v1pb.TaskRunLogEntry |
| 387 | for _, l := range logs { |
| 388 | switch l.Payload.Type { |
| 389 | case storepb.TaskRunLog_SCHEMA_DUMP_START: |
| 390 | entries = append(entries, &v1pb.TaskRunLogEntry{ |
| 391 | Type: v1pb.TaskRunLogEntry_SCHEMA_DUMP, |
| 392 | LogTime: timestamppb.New(l.T), |
| 393 | ReplicaId: l.Payload.ReplicaId, |
| 394 | SchemaDump: &v1pb.TaskRunLogEntry_SchemaDump{ |
| 395 | StartTime: timestamppb.New(l.T), |
| 396 | }, |
| 397 | }) |
| 398 | |
| 399 | case storepb.TaskRunLog_SCHEMA_DUMP_END: |
| 400 | if len(entries) == 0 { |
| 401 | continue |
| 402 | } |
| 403 | prev := entries[len(entries)-1] |
| 404 | if prev == nil || prev.Type != v1pb.TaskRunLogEntry_SCHEMA_DUMP { |
| 405 | continue |
| 406 | } |
| 407 | prev.SchemaDump.EndTime = timestamppb.New(l.T) |
| 408 | prev.SchemaDump.Error = l.Payload.SchemaDumpEnd.Error |
| 409 | |
| 410 | case storepb.TaskRunLog_COMMAND_EXECUTE: |
| 411 | entries = append(entries, &v1pb.TaskRunLogEntry{ |
| 412 | Type: v1pb.TaskRunLogEntry_COMMAND_EXECUTE, |
| 413 | LogTime: timestamppb.New(l.T), |
| 414 | ReplicaId: l.Payload.ReplicaId, |
| 415 | CommandExecute: &v1pb.TaskRunLogEntry_CommandExecute{ |
| 416 | LogTime: timestamppb.New(l.T), |
| 417 | Range: convertToRange(l.Payload.CommandExecute.Range), |
| 418 | Statement: l.Payload.CommandExecute.Statement, |
| 419 | }, |
| 420 | }) |
| 421 | |
| 422 | case storepb.TaskRunLog_COMMAND_RESPONSE: |
| 423 | if len(entries) == 0 { |
| 424 | continue |
| 425 | } |
| 426 | prev := entries[len(entries)-1] |
| 427 | if prev == nil || prev.Type != v1pb.TaskRunLogEntry_COMMAND_EXECUTE { |
| 428 | continue |
| 429 | } |
| 430 | prev.CommandExecute.Response = &v1pb.TaskRunLogEntry_CommandExecute_CommandResponse{ |
| 431 | LogTime: timestamppb.New(l.T), |
| 432 | Error: l.Payload.CommandResponse.Error, |
| 433 | AffectedRows: l.Payload.CommandResponse.AffectedRows, |
| 434 | AllAffectedRows: l.Payload.CommandResponse.AllAffectedRows, |
| 435 | } |
| 436 | |
| 437 | case storepb.TaskRunLog_DATABASE_SYNC_START: |
| 438 | entries = append(entries, &v1pb.TaskRunLogEntry{ |
| 439 | Type: v1pb.TaskRunLogEntry_DATABASE_SYNC, |
| 440 | LogTime: timestamppb.New(l.T), |
| 441 | ReplicaId: l.Payload.ReplicaId, |
| 442 | DatabaseSync: &v1pb.TaskRunLogEntry_DatabaseSync{ |