Repository navigation
ArrowBatchIterator turns a failed FetchResults into io.EOF #507
Description
This is the Arrow-batch form of #254. That issue was fixed for database/sql in v1.6.2 (#255): rows.Next returns the fetch error. rows.ArrowBatchIterator still does not.
When FetchResults fails between pages, resultPageIterator.HasNext stores the error and returns false. arrowRecordIterator.checkFinished treats that false as the end of the result and closes the iterator. From then on Next returns io.EOF. ArrowBatchIterator has no Err() method, and the next call to HasNext overwrites the stored error with io.EOF. A caller that stops on HasNext() == false or on io.EOF cannot tell a failed page from a finished result. The rows already returned look like a complete result set.
HasNext stores the error and reports no more pages. A later call replaces it with io.EOF. checkFinished takes that false as the end of the result. Next then returns io.EOF.
Expected: Next returns the page-fetch error instead of io.EOF when the fetch failed.
Reproducer, in internal/rows/arrowbased, next to TestArrowRecordIterator. It uses the same directResultsMultipleFetch fixtures: direct results are page 1, the mock returns page 2, and the following fetch fails. Today the loop ends with lastErr == nil.
func TestArrowRecordIteratorSwallowsPageError(t *testing.T) {
logger := dbsqllog.WithContext("connectionId", "correlationId", "")
executeStatementResp := cli_service.TExecuteStatementResp{}
loadTestData2(t, "directResultsMultipleFetch/ExecuteStatement.json", &executeStatementResp)
fetchResp1 := cli_service.TFetchResultsResp{}
loadTestData2(t, "directResultsMultipleFetch/FetchResults1.json", &fetchResp1)
calls := 0
failing := &client.TestClient{
FnFetchResults: func(ctx context.Context, req *cli_service.TFetchResultsReq) (*cli_service.TFetchResultsResp, error) {
calls++
if calls == 1 {
return &fetchResp1, nil
}
return nil, fmt.Errorf("FetchResults: connection reset")
},
}
ctx := driverctx.NewContextWithConnId(context.Background(), "connectionId")
ctx = driverctx.NewContextWithCorrelationId(ctx, "correlationId")
rpi := rowscanner.NewResultPageIterator(ctx, rowscanner.NewDelimiter(0, 7311), 5000, nil, false, failing, logger)
cfg := *config.WithDefaults()
bi, err := NewLocalBatchIterator(ctx, executeStatementResp.DirectResults.ResultSet.Results.ArrowBatches, 0, executeStatementResp.DirectResults.ResultSetMetadata.ArrowSchema, &cfg)
assert.Nil(t, err)
it := NewArrowRecordIterator(ctx, rpi, bi, executeStatementResp.DirectResults.ResultSetMetadata.ArrowSchema, cfg)
defer it.Close()
var lastErr error
rows := int64(0)
for it.HasNext() {
rec, err := it.Next()
if err != nil {
lastErr = err
break
}
rows += rec.NumRows()
rec.Release()
}
t.Logf("fetch calls=%d rows=%d lastErr=%v", calls, rows, lastErr)
assert.ErrorContains(t, lastErr, "connection reset")
}
Activity