resultForDefaultIter reads batches of rows from the iterator and writes results into the callback function.
(ctx *sql.Context, schema sql.Schema, iter sql.RowIter, callback func(*sql.Context, *Result) error, resultFields []pgproto3.FieldDescription, formatCodes []int16)
| 619 | // resultForDefaultIter reads batches of rows from the iterator |
| 620 | // and writes results into the callback function. |
| 621 | func (h *DoltgresHandler) resultForDefaultIter(ctx *sql.Context, schema sql.Schema, iter sql.RowIter, callback func(*sql.Context, *Result) error, resultFields []pgproto3.FieldDescription, formatCodes []int16) (*Result, bool, error) { |
| 622 | defer trace.StartRegion(ctx, "DoltgresHandler.resultForDefaultIter").End() |
| 623 | |
| 624 | // TODO: use errguard.Go instead? |
| 625 | pan2err := func(err *error) { |
| 626 | if HandlePanics { |
| 627 | if recoveredPanic := recover(); recoveredPanic != nil { |
| 628 | if err == nil { |
| 629 | *err = errors.Errorf("DoltgresHandler caught panic with nil error: %v: %s", recoveredPanic, debug.Stack()) |
| 630 | } else { |
| 631 | // debug.Stack() here prints the stack trace of the original panic, not the lexical stack of this defer function |
| 632 | *err = goerrors.Join(*err, errors.Errorf("DoltgresHandler caught panic: %v: %s", recoveredPanic, debug.Stack())) |
| 633 | } |
| 634 | } |
| 635 | } |
| 636 | } |
| 637 | |
| 638 | wg := sync.WaitGroup{} |
| 639 | wg.Add(3) |
| 640 | eg, ctx := ctx.NewErrgroup() |
| 641 | |
| 642 | // Read rows off the row iterator and send them to the row channel. |
| 643 | var rowChan = make(chan sql.Row, 512) |
| 644 | eg.Go(func() (err error) { |
| 645 | defer pan2err(&err) |
| 646 | defer wg.Done() |
| 647 | defer close(rowChan) |
| 648 | for { |
| 649 | select { |
| 650 | case <-ctx.Done(): |
| 651 | return context.Cause(ctx) |
| 652 | default: |
| 653 | row, iErr := iter.Next(ctx) |
| 654 | if iErr == io.EOF { |
| 655 | return nil |
| 656 | } |
| 657 | if iErr != nil { |
| 658 | return iErr |
| 659 | } |
| 660 | select { |
| 661 | case rowChan <- row: |
| 662 | case <-ctx.Done(): |
| 663 | return nil |
| 664 | } |
| 665 | } |
| 666 | } |
| 667 | }) |
| 668 | |
| 669 | // Default waitTime is one minute if there is no timeout configured, in which case |
| 670 | // it will loop to iterate again unless the socket died by the OS timeout or other problems. |
| 671 | // If there is a timeout, it will be enforced to ensure that Vitess has a chance to |
| 672 | // call DoltgresHandler.CloseConnection() |
| 673 | waitTime := 1 * time.Minute |
| 674 | if h.readTimeout > 0 { |
| 675 | waitTime = h.readTimeout |
| 676 | } |
| 677 | timer := time.NewTimer(waitTime) |
| 678 | defer timer.Stop() |