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(*Result) error, resultFields []pgproto3.FieldDescription)
| 726 | // resultForDefaultIter reads batches of rows from the iterator |
| 727 | // and writes results into the callback function. |
| 728 | func (h *DuckHandler) resultForDefaultIter(ctx *sql.Context, schema sql.Schema, iter sql.RowIter, callback func(*Result) error, resultFields []pgproto3.FieldDescription) (r *Result, processedAtLeastOneBatch bool, returnErr error) { |
| 729 | defer trace.StartRegion(ctx, "DuckHandler.resultForDefaultIter").End() |
| 730 | |
| 731 | eg, ctx := ctx.NewErrgroup() |
| 732 | |
| 733 | var rowChan = make(chan sql.Row, 512) |
| 734 | |
| 735 | pan2err := func() { |
| 736 | if recoveredPanic := recover(); recoveredPanic != nil { |
| 737 | returnErr = fmt.Errorf("DoltgresHandler caught panic: %v", recoveredPanic) |
| 738 | } |
| 739 | } |
| 740 | |
| 741 | wg := sync.WaitGroup{} |
| 742 | wg.Add(2) |
| 743 | // Read rows off the row iterator and send them to the row channel. |
| 744 | eg.Go(func() error { |
| 745 | defer pan2err() |
| 746 | defer wg.Done() |
| 747 | defer close(rowChan) |
| 748 | for { |
| 749 | select { |
| 750 | case <-ctx.Done(): |
| 751 | return nil |
| 752 | default: |
| 753 | row, err := iter.Next(ctx) |
| 754 | if err == io.EOF { |
| 755 | return nil |
| 756 | } |
| 757 | if err != nil { |
| 758 | return err |
| 759 | } |
| 760 | select { |
| 761 | case rowChan <- row: |
| 762 | case <-ctx.Done(): |
| 763 | return nil |
| 764 | } |
| 765 | } |
| 766 | } |
| 767 | }) |
| 768 | |
| 769 | // Default waitTime is one minute if there is no timeout configured, in which case |
| 770 | // it will loop to iterate again unless the socket died by the OS timeout or other problems. |
| 771 | // If there is a timeout, it will be enforced to ensure that Vitess has a chance to |
| 772 | // call DoltgresHandler.CloseConnection() |
| 773 | waitTime := 1 * time.Minute |
| 774 | if h.readTimeout > 0 { |
| 775 | waitTime = h.readTimeout |
| 776 | } |
| 777 | timer := time.NewTimer(waitTime) |
| 778 | defer timer.Stop() |
| 779 | |
| 780 | // reads rows from the channel, converts them to wire format, |
| 781 | // and calls |callback| to give them to vitess. |
| 782 | eg.Go(func() error { |
| 783 | defer pan2err() |
| 784 | // defer cancelF() |
| 785 | defer wg.Done() |